搭建一个 FastAPI 服务器

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

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

创建一个新项目

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

$ mkdir gino-fastapi-demo
$ cd gino-fastapi-demo
$ git init
$ poetry init

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

添加依赖关系

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

$ poetry add gino[starlette]

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

$ poetry add fastapi uvicorn gunicorn

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

$ poetry add alembic psycopg2

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

$ poetry add -D pytest requests

提示

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

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

[tool.poetry]
name = "gino-fastapi-demo"
version = "0.1.0"
description = ""
authors = ["Fantix King <fantix.king@gmail.com>"]

[tool.poetry.dependencies]
python = "^3.8"
gino = {version = "^1.0", extras = ["starlette"]}
fastapi = "^0.54.1"
uvicorn = "^0.11.3"
gunicorn = "^20.0.4"
alembic = "^1.4.2"
psycopg2 = "^2.8.5"

[tool.poetry.dev-dependencies]
pytest = "^5.4.1"
requests = "^2.23.0"

[build-system]
requires = ["poetry>=0.12"]
build-backend = "poetry.masonry.api"

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

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

编写一个简单的服务器

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

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

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

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

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

下面是 main.py 的内容：

from fastapi import FastAPI

def get_app():
    app = FastAPI(title="GINO FastAPI Demo")
    return app

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

from .main import get_app

app = get_app()

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

$ poetry install
Installing dependencies from lock file

No dependencies to install or update

  - Installing gino-fastapi-demo (0.1.0)

$ poetry run uvicorn gino_fastapi_demo.asgi:app --reload
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO:     Started reloader process [53010]
INFO:     Started server process [53015]
INFO:     Waiting for application startup.
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 扩展

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

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

from sqlalchemy.engine.url import URL, make_url
from starlette.config import Config
from starlette.datastructures import Secret

config = Config(".env")

DB_DRIVER = config("DB_DRIVER", default="postgresql")
DB_HOST = config("DB_HOST", default=None)
DB_PORT = config("DB_PORT", cast=int, default=None)
DB_USER = config("DB_USER", default=None)
DB_PASSWORD = config("DB_PASSWORD", cast=Secret, default=None)
DB_DATABASE = config("DB_DATABASE", default=None)
DB_DSN = config(
    "DB_DSN",
    cast=make_url,
    default=URL(
        drivername=DB_DRIVER,
        username=DB_USER,
        password=DB_PASSWORD,
        host=DB_HOST,
        port=DB_PORT,
        database=DB_DATABASE,
    ),
)
DB_POOL_MIN_SIZE = config("DB_POOL_MIN_SIZE", cast=int, default=1)
DB_POOL_MAX_SIZE = config("DB_POOL_MAX_SIZE", cast=int, default=16)
DB_ECHO = config("DB_ECHO", cast=bool, default=False)
DB_SSL = config("DB_SSL", default=None)
DB_USE_CONNECTION_FOR_REQUEST = config(
    "DB_USE_CONNECTION_FOR_REQUEST", cast=bool, default=True
)
DB_RETRY_LIMIT = config("DB_RETRY_LIMIT", cast=int, default=1)
DB_RETRY_INTERVAL = config("DB_RETRY_INTERVAL", cast=int, default=1)

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

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

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

DB_HOST=localhost
DB_USER=postgres

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

小技巧

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

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

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

from gino.ext.starlette import Gino

from .. import config

db = Gino(
    dsn=config.DB_DSN,
    pool_min_size=config.DB_POOL_MIN_SIZE,
    pool_max_size=config.DB_POOL_MAX_SIZE,
    echo=config.DB_ECHO,
    ssl=config.DB_SSL,
    use_connection_for_request=config.DB_USE_CONNECTION_FOR_REQUEST,
    retry_limit=config.DB_RETRY_LIMIT,
    retry_interval=config.DB_RETRY_INTERVAL,
)

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

 from fastapi import FastAPI
+
+from .models import db

 def get_app():
     app = FastAPI(title="GINO FastAPI Demo")
+    db.init_app(app)
     return app

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

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

创建 model 及 API

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

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

from . import db

class User(db.Model):
    __tablename__ = "users"

    id = db.Column(db.BigInteger(), primary_key=True)
    nickname = db.Column(db.Unicode(), default="unnamed")

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

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

from fastapi import APIRouter
from pydantic import BaseModel

from ..models.users import User

router = APIRouter()

@router.get("/users/{uid}")
async def get_user(uid: int):
    user = await User.get_or_404(uid)
    return user.to_dict()

class UserModel(BaseModel):
    name: str

@router.post("/users")
async def add_user(user: UserModel):
    rv = await User.create(nickname=user.name)
    return rv.to_dict()

@router.delete("/users/{uid}")
async def delete_user(uid: int):
    user = await User.get_or_404(uid)
    await user.delete()
    return dict(id=uid)

def init_app(app):
    app.include_router(router)

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

import logging
from importlib.metadata import entry_points

logger = logging.getLogger(__name__)

def load_modules(app=None):
    for ep in entry_points()["gino_fastapi_demo.modules"]:
        logger.info("Loading module: %s", ep.name)
        mod = ep.load()
        if app:
            init_app = getattr(mod, "init_app", None)
            if init_app:
                init_app(app)

提示

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

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

 def get_app():
     app = FastAPI(title="GINO FastAPI Demo")
     db.init_app(app)
+    load_modules(app)
     return app

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

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

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

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

集成 Alembic

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

$ poetry run alembic init migrations

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

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

 # add your model's MetaData object here
 # for 'autogenerate' support
 # from myapp import mymodel
 # target_metadata = mymodel.Base.metadata
-target_metadata = None
+from gino_fastapi_demo.config import DB_DSN
+from gino_fastapi_demo.main import db, load_modules
+
+load_modules()
+config.set_main_option("sqlalchemy.url", str(DB_DSN))
+target_metadata = db

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

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

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

def upgrade():
    op.create_table(
        "users",
        sa.Column("id", sa.BigInteger(), nullable=False),
        sa.Column("nickname", sa.Unicode(), nullable=True),
        sa.PrimaryKeyConstraint("id"),
    )

def downgrade():
    op.drop_table("users")

提示

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

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

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

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

编写测试

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

 config = Config(".env")

+TESTING = config("TESTING", cast=bool, default=False)

 DB_DRIVER = config("DB_DRIVER", default="postgresql")
 DB_HOST = config("DB_HOST", default=None)
 DB_PORT = config("DB_PORT", cast=int, default=None)
 DB_USER = config("DB_USER", default=None)
 DB_PASSWORD = config("DB_PASSWORD", cast=Secret, default=None)
 DB_DATABASE = config("DB_DATABASE", default=None)
+if TESTING:
+    if DB_DATABASE:
+        DB_DATABASE += "_test"
+    else:
+        DB_DATABASE = "gino_fastapi_demo_test"
 DB_DSN = config(

提示

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

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

import pytest
from alembic.config import main
from starlette.config import environ
from starlette.testclient import TestClient

environ["TESTING"] = "TRUE"

@pytest.fixture
def client():
    from gino_fastapi_demo.main import db, get_app

    main(["--raiseerr", "upgrade", "head"])

    with TestClient(get_app()) as client:
        yield client

    main(["--raiseerr", "downgrade", "base"])

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

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

import uuid

def test_crud(client):
    # create
    nickname = str(uuid.uuid4())
    r = client.post("/users", json=dict(name=nickname))
    r.raise_for_status()

    # retrieve
    url = f"/users/{r.json()['id']}"
    assert client.get(url).json()["nickname"] == nickname

    # delete
    client.delete(url).raise_for_status()
    assert client.get(url).status_code == 404

测试跑起来：

$ poetry run pytest
=========================== test session starts ===========================
platform darwin -- Python 3.8.2, pytest-5.4.1, py-1.8.1, pluggy-0.13.1
rootdir: gino-fastapi-demo
collected 1 item

tests/test_users.py .                                               [100%]

============================ 1 passed in 1.21s ============================

生产环境注意事项

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

FROM python:3.8-alpine as base

FROM base as builder
RUN apk add --no-cache gcc musl-dev libffi-dev openssl-dev make postgresql-dev
RUN pip install poetry
COPY . /src/
WORKDIR /src
RUN python -m venv /env && . /env/bin/activate && poetry install

FROM base
RUN apk add --no-cache postgresql-libs
COPY --from=builder /env /env
COPY --from=builder /src /src
WORKDIR /src
CMD ["/env/bin/gunicorn", "gino_fastapi_demo.asgi:app", "-b", "0.0.0.0:80", "-k", "uvicorn.workers.UvicornWorker"]

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

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

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

• 将 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，修问题或者分享想法都行。祝玩得愉快！