FastAPI 最佳实践

FastAPI 最佳实践

虽然FastAPI是一个有着出色文档的伟大框架,但对于初学者来说,如何构建大型项目并不是那么明显。

在过去的1.5年中,我们做出了好的和坏的决定,这些决定对我们的开发人员体验产生了巨大的影响。其中一些值得分享。

目录

项目结构。一致且可预测
过度使用Pydantic进行数据验证
使用依赖项进行数据验证而不是DB
解耦和重用依赖项。依赖调用被缓存
如果只有阻塞I/O操作,请勿使你的路由异步
迁移。Alembic
BackgroundTasks > asyncio.create_task
小心动态的Pydantic字段
分块保存文件
如果必须使用同步SDK,请在线程池中运行。 本文仅包含我们遵循的准则的一部分,所以请随意找到带有详细最佳实践完整列表的 原始GitHub存储库,它已经获得了一些积极的反馈(在 r/Python 中成为了一天的热门帖子,并在第一周内获得了250个星标).
  1. 项目结构。一致且可预测

有许多方法可以构建项目,但最好的结构是一种一致、简单明了且没有惊喜的结构。

如果查看项目结构不能让你了解项目的内容,则结构可能不清晰。
如果你必须打开包才能了解其中包含的模块,则你的结构不清晰。
如果文件的频率和位置感觉随机,则你的项目结构很糟糕。
如果查看模块的位置和名称不能让你了解其中的内容,则你的结构非常糟糕。

虽然我们使用的项目结构是由 Sebastián Ramírez 提出的,该结构将文件按其类型(例如API、CRUD、模型、模式)分开,适用于微服务或具有较少范围的项目,但是我们无法将其适用于具有许多域和模块的单块。

我发现更具可扩展性和可演变性的结构是受Netflix的 Dispatch启发,有一些小修改。

fastapi-project
├── alembic/
├── src
│ ├── auth
│ │ ├── router.py
│ │ ├── schemas.py # pydantic models
│ │ ├── models.py # db models
│ │ ├── dependencies.py
│ │ ├── config.py # local configs
│ │ ├── constants.py
│ │ ├── exceptions.py
│ │ ├── service.py
│ │ └── utils.py
│ ├── aws
│ │ ├── client.py # client model for external service communication
│ │ ├── schemas.py
│ │ ├── config.py
│ │ ├── constants.py
│ │ ├── exceptions.py
│ │ └── utils.py
│ └── posts
│ │ ├── router.py
│ │ ├── schemas.py
│ │ ├── models.py
│ │ ├── dependencies.py
│ │ ├── constants.py
│ │ ├── exceptions.py
│ │ ├── service.py
│ │ └── utils.py
│ ├── config.py # global configs
│ ├── models.py # global models
│ ├── exceptions.py # global exceptions
│ ├── pagination.py # global module e.g. pagination
│ ├── database.py # db connection related stuff
│ └── main.py
├── tests/
│ ├── auth
│ ├── aws
│ └── posts
├── templates/
│ └── index.html
├── requirements
│ ├── base.txt
│ ├── dev.txt
│ └── prod.txt
├── .env
├── .gitignore
├── logging.ini
└── alembic.ini

当一个包需要其他包的服务或依赖项时,请使用显式模块名称导入它们。

from src.auth import constants as auth_constants
from src.notifications import service as notification_service
from src.posts.constants import ErrorCode as PostsErrorCode

  1. 过度使用Pydantic进行数据验证

Pydantic具有丰富的功能,可验证和转换数据。

除了常规功能,如具有默认值的必需和非必需字段之外,Pydantic还具有内置的全面数据处理工具,如正则表达式、限制允许选项的枚举、长度验证、电子邮件验证等。

  1. 使用依赖项进行数据验证而不是DB

Pydantic只能验证客户端输入的值。使用依赖项验证数据是否符合数据库约束,例如电子邮件已存在、未找到用户等。

作为奖励,使用常见依赖项可以消除为每个这些路由编写测试以验证post_id的需要。

  1. 解耦和重用依赖项。依赖调用被缓存

依赖项可以多次重复使用,它们不会被重新计算——FastAPI默认情况下在请求范围内缓存依赖项的结果,即如果我们有一个调用服务 get_post_by_id 的依赖项,则每次调用此依赖项时我们不会访问DB-仅在第一次函数调用时。

了解这一点,我们可以轻松地将依赖项解耦为多个较小的函数,这些函数在较小的域上运行,更容易在其他路由中重用。例如,在下面的代码中,我们使用 parse_jwt_data 依赖项三次:

valid_owned_post
valid_active_creator
get_user_post, 但 parse_jwt_data 仅在第一个调用中调用。
  1. 如果只有阻塞I/O操作,请勿使你的路由异步

在幕后,FastAPI可以有效地处理异步和同步I/O操作。

FastAPI在线程池中运行 sync 路由,阻塞I/O操作不会阻止事件循环执行任务。
否则,如果路由定义为 async,则会通过 await 正常调用,FastAPI相信你只会执行非阻塞I/O操作。

注意事项是,如果你不信任并在异步路由中执行阻塞操作,则事件循环将无法运行下一个任务,直到该阻塞操作完成。

第二个注意事项是,非阻塞可等待操作或发送到线程池的操作必须是I/O密集型任务(例如,打开文件、DB调用、外部API调用)。

等待CPU密集型任务(例如,重型计算、数据处理、视频转码)是无用的,因为CPU必须工作才能完成任务,而I/O操作是外部的,服务器在等待这些操作完成时什么都不做,因此可以转到下一个任务。
在其他线程中运行CPU密集型任务也不起作用,因为GIL。简而言之,GIL一次只允许一个线程工作,这使得它对CPU任务无用。
如果要优化CPU密集型任务,应将它们发送到另一个进程中的工作程序。
  1. 迁移 Alembic。

    迁移必须是静态且可还原的。如果你的迁移依赖于动态生成的数据,请确保唯一动态的是数据本身,而不是其结构。
    使用具有描述性名称和短标识的迁移。标识是必需的,应解释更改。
    为新迁移设置可读的文件模板。我们使用 date_slug.py 模式,例如 2022-08-24_post_content_idx.py

alembic.ini

file_template = %%(year)d-%%(month).2d-%%(day).2d_%%(slug)s

  1. BackgroundTasks > asyncio.create_task

BackgroundTasks 可以有效地运行阻塞和非阻塞 I/O 操作,就像处理路由一样(同步函数在线程池中运行,而异步函数稍后会被等待执行)。

不要欺骗工作线程,也不要将阻塞 I/O 操作标记为“async”。
不要将其用于重量级 CPU 密集型任务。
  1. 分块保存文件

不要指望客户端只发送小文件。

  1. 动态 pydantic 字段需谨慎处理

如果你有一个 pydantic 字段可以接受类型的联合,请确保验证器明确知道这些类型之间的区别。

非常糟糕的解决方案:

适当地对字段类型进行排序:从最严格的到最宽松的。

验证输入是否只包含有效字段。

Pydantic 会忽略联合类型的 ValueError 并迭代它们。如果没有类型是有效的,则会引发最后一个异常。

如果字段简单,则使用 Pydantic 的智能联合(>v1.9)

如果字段很简单,比如 int 或 bool,这是一个很好的解决方案,但它对于像类这样的复杂字段无效。

没有智能联合的情况:

使用智能联合的情况:

  1. 如果必须使用同步 SDK,则在线程池中运行

如果必须使用 SDK 与外部服务进行交互,并且它不是“async”,则在外部工作线程中进行 HTTP 调用。

举个简单的例子,我们可以使用我们熟知的 starlette 的 run_in_threadpool。

FastAPI 是一个工具,可以轻松构建简单和复杂的项目。不是缺少上述规则而导致了难以维护的项目,而是缺乏一致性。

无论你有什么规则,唯一应该遵循的规则就是遵循你的规则的一致性。找到一组有效的约定,对其进行迭代,并将其向他人宣传。如果你已经有了这样的约定,请在 问题页面 上与其他人分享。

译自:https://betterprogramming.pub/fastapi-best-practices-1f0deeba4fce

FastAPI 开发模板

FastAPI Example Project

Some people were searching my GitHub profile for project examples after reading the article on FastAPI best practices.
Unfortunately, I didn't have useful public repositories, but only my old proof-of-concept projects.

Hence, I have decided to fix that and show how I start projects nowadays, after getting some real-world experience.
This repo is kind of a template I use when starting up new FastAPI projects:

  • production-ready

    • gunicorn with dynamic workers configuration (stolen from @tiangolo)
    • Dockerfile optimized for small size and fast builds with a non-root user
    • JSON logs
    • sentry for deployed envs
  • easy local development

    • environment with configured postgres and redis
    • script to lint code with black, autoflake, isort (also stolen from @tiangolo)
    • configured pytest with async-asgi-testclient, pytest-env, pytest-asyncio
    • fully typed to comply with mypy
  • SQLAlchemy with slightly configured alembic

    • async db calls with asyncpg
    • set up sqlalchemy2-stubs
    • migrations set in easy to sort format (YYYY-MM-DD_slug)
  • pre-installed JWT authorization

    • short-lived access token
    • long-lived refresh token which is stored in http-only cookies
    • salted password storage with bcrypt
  • global pydantic model with

    • orjson
    • explicit timezone setting during JSON export
  • and some other extras like global exceptions, sqlalchemy keys naming convention, shortcut scripts for alembic, etc.

Local Development

First Build Only

  1. cp .env.example .env
  2. docker network create app_main
  3. docker-compose up -d --build

Linters

Format the code

docker compose exec app format

Migrations

  • Create an automatic migration from changes in src/database.py

    docker compose exec app makemigrations *migration_name*
  • Run migrations

    docker compose exec app migrate
  • Downgrade migrations

    docker compose exec app downgrade -1  # or -2 or base or hash of the migration

    Tests

    All tests are integrational and require DB connection.

One of the choices I've made is to use default database (postgres), separated from app's app database.

  • Using default database makes it easier to run tests in CI/CD environments, since there is no need to setup additional databases
  • Tests are run with force_rollback=True, i.e. every transaction made is then reverted

Run tests

docker compose exec app pytest

Documentation

swagger - http://localhost:8000/docs
redoc - http://localhost:8000/redoc