搭建一个 FastAPI 服务器

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

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

../_images/gino-fastapi.svg

创建一个新项目

这里我们尝试用亮瞎眼的 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 库,因为 StarletteTestClient 要用到它:

$ 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"
../_images/gino-fastapi-poetry.svg

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

$ git add pyproject.toml poetry.lock
$ 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 的内容:

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 扩展

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

现在让我们把 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
../_images/gino-fastapi-env.svg

也可以在 .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 仍然能够覆盖前者的值。

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

然后创建一个 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

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

现在轮到实现 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")
../_images/gino-fastapi-views.svg

很简单的 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
../_images/gino-fastapi-alembic.svg

这句命令会生产一个新的文件夹 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"])
../_images/gino-fastapi-tests.svg

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

下面是一个简单的测试例子,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 搭配 UvicornUvicornWorker 的方式来获取最佳生产级别可靠性。

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

../_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 的升级流程,比如在每次新版本部署之后执行。

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

  • 编写压力测试。

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