搭建一个 FastAPI 服务器

在这篇教程里，我们会一起搭建一个用于生产环境的 FastAPI 服务器。完整的示例代码在这里 。

写好之后，整个应用技术栈会是这样的：

../_images/gino-fastapi.svg

创建一个新项目

这里我们尝试用亮瞎眼的 Poetry 来管理我们的项目，而不是传统的 pip。请跟随链接安装 Poetry，并且在一个空文件夹中创建我们的新项目：

  1. $ mkdir gino-fastapi-demo
  2. $ cd gino-fastapi-demo
  3. $ git init
  4. $ poetry init

然后跟着 Poetry 的向导完成初始化——关于交互式创建依赖的两个问题，您可以回答“no”，因为我们会在下面手动创建。其他问题都可以用默认值，只是一定保证包的名字是 gino-fastapi-demo。

添加依赖关系

FastAPI 底层用的是 Starlette 框架，所以我们就可以直接使用 GINO 的 Starlette 扩展。执行以下命令即可：

  1. $ poetry add gino[starlette]

接着我们添加 FastAPI，以及快成一道闪电的 ASGI 服务器 Uvicorn，还有用作生产环境的应用服务器 Gunicorn：

  1. $ poetry add fastapi uvicorn gunicorn

我们将用 Alembic 来管理数据库表结构变更。因为 Alembic 只兼容传统的 DB-API 驱动，所以我们还得加上 psycopg：

  1. $ poetry add alembic psycopg2

最后，测试框架选用 pytest，我们将其添加到开发环境的依赖关系中。同时也加上 requests 库，因为 Starlette 的 TestClient 要用到它：

  1. $ poetry add -D pytest requests

提示

经过了上面的步骤，Poetry 会悄没声地帮我们自动创建一个 virtualenv，并且把所有的依赖关系装到这个虚拟环境里。在本教程后面的步骤里，我们会假定继续使用这个环境。但是，您也可以创建自己的 virtualenv，只要激活了 Poetry 就会用它。

以上。下面是 Poetry 给我创建出来的 pyproject.toml 文件内容，您的应该也长得差不多：

  1. [tool.poetry]
  2. name = "gino-fastapi-demo"
  3. version = "0.1.0"
  4. description = ""
  5. authors = ["Fantix King <fantix.king@gmail.com>"]
  7. [tool.poetry.dependencies]
  8. python = "^3.8"
  9. gino = {version = "^1.0", extras = ["starlette"]}
  10. fastapi = "^0.54.1"
  11. uvicorn = "^0.11.3"
  12. gunicorn = "^20.0.4"
  13. alembic = "^1.4.2"
  14. psycopg2 = "^2.8.5"
  16. [tool.poetry.dev-dependencies]
  17. pytest = "^5.4.1"
  18. requests = "^2.23.0"
  20. [build-system]
  21. requires = ["poetry>=0.12"]
  22. build-backend = "poetry.masonry.api"

../_images/gino-fastapi-poetry.svg

同时自动生成的还有一个叫 poetry.lock 的文件，内容是当前完整依赖关系树的精确版本号，当前的目录结构如右图所示。现在让我们把这两个文件加到 Git 仓库中（以后的步骤就不再演示 Git 的操作了）：

  1. $ git add pyproject.toml poetry.lock
  2. $ git commit -m 'add project dependencies'

编写一个简单的服务器

现在让我们写一点 Python 的代码吧。

我们要创建一个 src 文件夹，用来装所有的 Python 文件，如下图所示。这种目录结构叫做“src 布局”，能让项目结构更清晰。

../_images/gino-fastapi-src.svg

我们项目的顶层 Python 包叫做 gino_fastapi_demo，我们在里面创建两个 Python 模块：

  • asgi 作为 ASGI 的入口，将被 ASGI 服务器直接使用

  • main 用来初始化我们自己的服务器

下面是 main.py 的内容：

  1. from fastapi import FastAPI
  3. def get_app():
  4.     app = FastAPI(title="GINO FastAPI Demo")
  5.     return app

在 asgi.py 里，我们只需要实例化我们的应用即可：

  1. from .main import get_app
  3. app = get_app()

然后执行 poetry install 来把我们的 Python 包以开发模式链接到 PYTHONPATH 中，接下来就可以启动 Uvicorn 的开发服务器了：

  1. $ poetry install
  2. Installing dependencies from lock file
  4. No dependencies to install or update
  6.   - Installing gino-fastapi-demo (0.1.0)
  8. $ poetry run uvicorn gino_fastapi_demo.asgi:app --reload
  9. INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
  10. INFO:     Started reloader process [53010]
  11. INFO:     Started server process [53015]
  12. INFO:     Waiting for application startup.
  13. INFO:     Application startup complete.

这里的 --reload 选项会启用 Uvicorn 的自动加载功能，当我们的 Python 代码发生变动的时候，Uvicorn 会自动加载使用新代码。现在可以访问 http://127.0.0.1:8000/docs 了，试一下我们新 FastAPI 服务器的 Swagger UI 接口文档。

提示

正如之前提到的，如果您使用自己的虚拟环境，那么此处的 poetry run uvicorn 就可以简化为 uvicorn。

poetry run 是一个快捷命令，用于在 Poetry 管理的虚拟环境中执行后续的命令。

添加 GINO 扩展

../_images/gino-fastapi-config.svg

现在让我们把 GINO 添加到服务器里。

首先，我们需要有办法来配置数据库。在本教程中，我们选用 Starlette 的配置系统。创建文件 src/gino_fastapi_demo/config.py，内容为：

  1. from sqlalchemy.engine.url import URL, make_url
  2. from starlette.config import Config
  3. from starlette.datastructures import Secret
  5. config = Config(".env")
  7. DB_DRIVER = config("DB_DRIVER", default="postgresql")
  8. DB_HOST = config("DB_HOST", default=None)
  9. DB_PORT = config("DB_PORT", cast=int, default=None)
  10. DB_USER = config("DB_USER", default=None)
  11. DB_PASSWORD = config("DB_PASSWORD", cast=Secret, default=None)
  12. DB_DATABASE = config("DB_DATABASE", default=None)
  13. DB_DSN = config(
  14.     "DB_DSN",
  15.     cast=make_url,
  16.     default=URL(
  17.         drivername=DB_DRIVER,
  18.         username=DB_USER,
  19.         password=DB_PASSWORD,
  20.         host=DB_HOST,
  21.         port=DB_PORT,
  22.         database=DB_DATABASE,
  23.     ),
  24. )
  25. DB_POOL_MIN_SIZE = config("DB_POOL_MIN_SIZE", cast=int, default=1)
  26. DB_POOL_MAX_SIZE = config("DB_POOL_MAX_SIZE", cast=int, default=16)
  27. DB_ECHO = config("DB_ECHO", cast=bool, default=False)
  28. DB_SSL = config("DB_SSL", default=None)
  29. DB_USE_CONNECTION_FOR_REQUEST = config(
  30.     "DB_USE_CONNECTION_FOR_REQUEST", cast=bool, default=True
  31. )
  32. DB_RETRY_LIMIT = config("DB_RETRY_LIMIT", cast=int, default=1)
  33. DB_RETRY_INTERVAL = config("DB_RETRY_INTERVAL", cast=int, default=1)

这个配置文件会首先从环境变量中加载配置参数，如果没找到，则会从当前路径（通常是项目顶层目录）下一个叫 .env 的文件中加载，最后不行才会使用上面定义的默认值。比如，您即可以在命令行中设置：

  1. $ DB_HOST=localhost DB_USER=postgres poetry run uvicorn gino_fastapi_demo.asgi:app --reload

../_images/gino-fastapi-env.svg

也可以在 .env 文件中设置（一定不要将该文件提交到 Git 中，记得在 .gitignore 里加上它）：

  1. DB_HOST=localhost
  2. DB_USER=postgres

接下来就该创建 PostgreSQL 数据库实例并且将连接参数设置好了。创建数据库实例的命令通常是 createdb yourdbname，但不同平台可能有不同的方式，此教程里就不具体写了。

小技巧

另外，您也可以使用 DB_DSN 来定义数据库连接参数，比如 postgresql://user:password@localhost:5432/dbname，它会覆盖出现在它前面的单个的配置，比如 DB_HOST。

除了默认值不算之外，只要您定义了 DB_DSN ——不管是在环境变量中还是在 .env 文件中，它都比单个的连接参数有更高的优先级。比如哪怕环境变量中定义了 DB_HOST，.env 文件中的 DB_DSN 仍然能够覆盖前者的值。

../_images/gino-fastapi-models.svg

然后创建一个 Python 的二级包 gino_fastapi_demo.models，用来封装数据库相关的代码。将下面的代码添加到 src/gino_fastapi_demo/models/__init__.py：

  1. from gino.ext.starlette import Gino
  3. from .. import config
  5. db = Gino(
  6.     dsn=config.DB_DSN,
  7.     pool_min_size=config.DB_POOL_MIN_SIZE,
  8.     pool_max_size=config.DB_POOL_MAX_SIZE,
  9.     echo=config.DB_ECHO,
  10.     ssl=config.DB_SSL,
  11.     use_connection_for_request=config.DB_USE_CONNECTION_FOR_REQUEST,
  12.     retry_limit=config.DB_RETRY_LIMIT,
  13.     retry_interval=config.DB_RETRY_INTERVAL,
  14. )

最后，修改 src/gino_fastapi_demo/main.py，安装 GINO 扩展：

  1.  from fastapi import FastAPI
  2. +
  3. +from .models import db
  5.  def get_app():
  6.      app = FastAPI(title="GINO FastAPI Demo")
  7. +    db.init_app(app)
  8.      return app

保存该文件后，您应该可以看到 Uvicorn 服务器重载了我们的变更，然后连上了数据库：

  1. WARNING:  Detected file change in 'src/gino_fastapi_demo/main.py'. Reloading...
  2. INFO:     Shutting down
  3. INFO:     Waiting for application shutdown.
  4. INFO:     Application shutdown complete.
  5. INFO:     Finished server process [63562]
  6. INFO:     Started server process [63563]
  7. INFO:     Waiting for application startup.
  8. INFO:     Connecting to the database: postgresql://fantix:***@localhost
  9. INFO:     Database connection pool created: <asyncpg.pool.Pool max=16 min=1 cur=1 use=0>
  10. INFO:     Application startup complete.

创建 model 及 API

../_images/gino-fastapi-models-users.svg

现在轮到实现 API 逻辑了。比方说我们打算做一个用户管理的服务，可以添加、查看和删除用户。

首先，我们需要一张数据库表 users，用于存储数据。在 gino_fastapi_demo.models.users 模块中添加一个映射这张表的 model User：

  1. from . import db
  3. class User(db.Model):
  4.     __tablename__ = "users"
  6.     id = db.Column(db.BigInteger(), primary_key=True)
  7.     nickname = db.Column(db.Unicode(), default="unnamed")

../_images/gino-fastapi-views.svg

很简单的 model 定义，一切尽在不言中。

然后我们只需要在 API 的实现中正确使用它即可。创建一个新的 Python 二级包 gino_fastapi_demo.models.views，在其中添加一个模块 gino_fastapi_demo.views.users，内容为：

  1. from fastapi import APIRouter
  2. from pydantic import BaseModel
  4. from ..models.users import User
  6. router = APIRouter()
  8. @router.get("/users/{uid}")
  9. async def get_user(uid: int):
  10.     user = await User.get_or_404(uid)
  11.     return user.to_dict()
  13. class UserModel(BaseModel):
  14.     name: str
  16. @router.post("/users")
  17. async def add_user(user: UserModel):
  18.     rv = await User.create(nickname=user.name)
  19.     return rv.to_dict()
  21. @router.delete("/users/{uid}")
  22. async def delete_user(uid: int):
  23.     user = await User.get_or_404(uid)
  24.     await user.delete()
  25.     return dict(id=uid)
  27. def init_app(app):
  28.     app.include_router(router)

APIRouter 用来收集新接口的定义，然后在 init_app 里集成到 FastAPI 应用中去。这里我们加一点反转控制 ：把接口做成模块化的，用 Entry Points 功能进行拼装，避免需要手动一一 import 将来可能有的其他接口。将下面的代码添加到 gino_fastapi_demo.main：

  1. import logging
  2. from importlib.metadata import entry_points
  4. logger = logging.getLogger(__name__)
  6. def load_modules(app=None):
  7.     for ep in entry_points()["gino_fastapi_demo.modules"]:
  8.         logger.info("Loading module: %s", ep.name)
  9.         mod = ep.load()
  10.         if app:
  11.             init_app = getattr(mod, "init_app", None)
  12.             if init_app:
  13.                 init_app(app)

提示

如果您的 Python 版本低于 3.8，您还需要这个 importlib-metadata 的移植。

然后在我们的应用工厂函数中调用它：

  1.  def get_app():
  2.      app = FastAPI(title="GINO FastAPI Demo")
  3.      db.init_app(app)
  4. +    load_modules(app)
  5.      return app

最后，根据 Poetry 插件文档，在 pyproject.toml 文件中定义 Entry Point：

  1. [tool.poetry.plugins."gino_fastapi_demo.modules"]
  2. "users" = "gino_fastapi_demo.views.users"

再执行一次 poetry install 来激活这些 Entry Point——这次您可能需要亲自重启 Uvicorn 的开发服务器了，因为自动重载机制无法识别 pyproject.toml 文件的变更。

现在您应该可以在 Swagger UI 中看到那 3 个新接口了，但是它们还都不能用，因为我们还没有创建数据库表。

集成 Alembic

请在项目顶层文件夹中执行下面的命令，以开始使用 Alembic：

  1. $ poetry run alembic init migrations

../_images/gino-fastapi-alembic.svg

这句命令会生产一个新的文件夹 migrations，包含了 Alembic 用于数据库表结构变更追踪的版本文件。同时创建的还有一个在顶层文件夹下面的 alembic.ini 文件，我们把这些文件都添加到 Git 中。

为了能让 Alembic 用上我们用 GINO 定义的 model，我们需要修改 migrations/env.py 文件去链接 GINO 实例：

  1.  # add your model's MetaData object here
  2.  # for 'autogenerate' support
  3.  # from myapp import mymodel
  4.  # target_metadata = mymodel.Base.metadata
  5. -target_metadata = None
  6. +from gino_fastapi_demo.config import DB_DSN
  7. +from gino_fastapi_demo.main import db, load_modules
  8. +
  9. +load_modules()
  10. +config.set_main_option("sqlalchemy.url", str(DB_DSN))
  11. +target_metadata = db

然后就可以创建我们的第一个变更版本了：

  1. $ poetry run alembic revision --autogenerate -m 'add users table'
  2. INFO  [alembic.runtime.migration] Context impl PostgresqlImpl.
  3. INFO  [alembic.runtime.migration] Will assume transactional DDL.
  4. INFO  [alembic.autogenerate.compare] Detected added table 'users'
  5.   Generating migrations/versions/32c0feba61ea_add_users_table.py ...  done

生成的版本文件大体上应该长这样：

  1. def upgrade():
  2.     op.create_table(
  3.         "users",
  4.         sa.Column("id", sa.BigInteger(), nullable=False),
  5.         sa.Column("nickname", sa.Unicode(), nullable=True),
  6.         sa.PrimaryKeyConstraint("id"),
  7.     )
  9. def downgrade():
  10.     op.drop_table("users")

提示

以后需要再次修改数据库表结构的时候，您只需要修改 GINO model 然后执行 alembic revision --autogenerate 命令来生成对应改动的新版本即可。提交前记得看一下生成的版本文件，有时需要调整。

我们终于可以应用此次变更了，执行下面的命令将数据库表结构版本升级至最高：

  1. $ poetry run alembic upgrade head
  2. INFO  [alembic.runtime.migration] Context impl PostgresqlImpl.
  3. INFO  [alembic.runtime.migration] Will assume transactional DDL.
  4. INFO  [alembic.runtime.migration] Running upgrade  -> 32c0feba61ea, add users table

到这里，所有的接口应该都可以正常工作了，您可以在 Swagger UI 中试一下。

编写测试

为了不影响开发环境的数据库，我们需要为测试创建单独的数据库。根据下面的补丁修改 gino_fastapi_demo.config：

  1.  config = Config(".env")
  3. +TESTING = config("TESTING", cast=bool, default=False)
  5.  DB_DRIVER = config("DB_DRIVER", default="postgresql")
  6.  DB_HOST = config("DB_HOST", default=None)
  7.  DB_PORT = config("DB_PORT", cast=int, default=None)
  8.  DB_USER = config("DB_USER", default=None)
  9.  DB_PASSWORD = config("DB_PASSWORD", cast=Secret, default=None)
  10.  DB_DATABASE = config("DB_DATABASE", default=None)
  11. +if TESTING:
  12. +    if DB_DATABASE:
  13. +        DB_DATABASE += "_test"
  14. +    else:
  15. +        DB_DATABASE = "gino_fastapi_demo_test"
  16.  DB_DSN = config(

提示

您需要执行 createdb 来创建数据库实例。比如说，如果您在 .env 文件中定义了 DB_DATABASE=mydb，那么测试数据库的名字就是 mydb_test。否则如果没定义的话，默认就是 gino_fastapi_demo_test。

然后在 tests/conftest.py 中创建 pytest fixture：

  1. import pytest
  2. from alembic.config import main
  3. from starlette.config import environ
  4. from starlette.testclient import TestClient
  6. environ["TESTING"] = "TRUE"
  8. @pytest.fixture
  9. def client():
  10.     from gino_fastapi_demo.main import db, get_app
  12.     main(["--raiseerr", "upgrade", "head"])
  14.     with TestClient(get_app()) as client:
  15.         yield client
  17.     main(["--raiseerr", "downgrade", "base"])

../_images/gino-fastapi-tests.svg

这个 fixture 的作用是，在跑测试之前创建所有的数据库表、提供一个 Starlette 的 TestClient、并且在测试跑完之后删除所有的表及其数据，为后续测试保持一个干净的环境。

下面是一个简单的测试例子，tests/test_users.py：

  1. import uuid
  3. def test_crud(client):
  4.     # create
  5.     nickname = str(uuid.uuid4())
  6.     r = client.post("/users", json=dict(name=nickname))
  7.     r.raise_for_status()
  9.     # retrieve
  10.     url = f"/users/{r.json()['id']}"
  11.     assert client.get(url).json()["nickname"] == nickname
  13.     # delete
  14.     client.delete(url).raise_for_status()
  15.     assert client.get(url).status_code == 404

测试跑起来：

  1. $ poetry run pytest
  2. =========================== test session starts ===========================
  3. platform darwin -- Python 3.8.2, pytest-5.4.1, py-1.8.1, pluggy-0.13.1
  4. rootdir: gino-fastapi-demo
  5. collected 1 item
  7. tests/test_users.py .                                               [100%]
  9. ============================ 1 passed in 1.21s ============================

生产环境注意事项

最近 Docker/Kubernetes 挺火，我们也写一个 Dockerfile：

  1. FROM python:3.8-alpine as base
  3. FROM base as builder
  4. RUN apk add --no-cache gcc musl-dev libffi-dev openssl-dev make postgresql-dev
  5. RUN pip install poetry
  6. COPY . /src/
  7. WORKDIR /src
  8. RUN python -m venv /env && . /env/bin/activate && poetry install
  10. FROM base
  11. RUN apk add --no-cache postgresql-libs
  12. COPY --from=builder /env /env
  13. COPY --from=builder /src /src
  14. WORKDIR /src
  15. CMD ["/env/bin/gunicorn", "gino_fastapi_demo.asgi:app", "-b", "0.0.0.0:80", "-k", "uvicorn.workers.UvicornWorker"]

这个 Dockerfile 里，为了降低目标镜像文件的大小，我们分成了两步来分别进行源码构建和生产镜像的组装。另外，我们还采用了 Gunicorn 搭配 Uvicorn 的 UvicornWorker 的方式来获取最佳生产级别可靠性。

回头看一下项目里一共有哪些文件。

../_images/gino-fastapi-layout.svg

至此，我们就完成了演示项目的开发。下面是上生产可以用到的一个不完整检查清单：

  • 将 DB_RETRY_LIMIT 设置成一个稍微大一点的数字，以支持在数据库就绪前启动应用服务器的情况。

  • 在 migrations/env.py 中实现同样的重连尝试逻辑，这样 Alembic 也能拥有同样的特性。

  • 如果需要的话，启用 DB_SSL。

  • 写一个 docker-compose.yml，用于其他开发人员快速尝鲜，甚至可以用于开发。

  • 启用持续集成 <CI_>，安装 pytest-cov 并且用 --cov-fail-under 参数来保障测试覆盖率。

  • 集成静态代码检查工具和安全性/CVE筛查工具。

  • 正确自动化 Alembic 的升级流程，比如在每次新版本部署之后执行。

  • 注意针对诸如 CSRF、XSS 等常见攻击的安全性防护。

  • 编写压力测试。

最后再贴一次，实例程序的源码在这里，本教程的文档源码在`这里<https://github.com/python-gino/gino/blob/master/docs/tutorials/fastapi.rst>`__，请敞开了提 PR，修问题或者分享想法都行。祝玩得愉快！