搭建一个 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[pg,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 = ["pg", "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 的升级流程，比如在每次新版本部署之后执行。
编写压力测试。
最后再贴一次，实例程序的源码在这里，本教程的文档源码在`这里<https://github.com/python-gino/gino/blob/master/docs/tutorials/fastapi.rst>`__，请敞开了提 PR，修问题或者分享想法都行。祝玩得愉快！
