Fast Api最佳实践指南

这是我在初创公司使用的一系列最佳实践和约定。
在过去几年的生产实践中,我们做过一些好的和不好的决策,这些决策极大地影响了开发者体验。其中一些经验值得分享。
Fast Api最佳实践指南
目录
项目结构
异步路由
I/O密集型任务
CPU密集型任务
Pydantic
大量使用Pydantic
自定义基础模型
拆分Pydantic BaseSettings
依赖项
超越依赖注入
链式依赖
拆分并复用依赖项。依赖调用会被缓存
优先使用async依赖项
其他
遵循REST规范
FastAPI响应序列化
如果必须使用同步SDK,请在线程池中运行它。
ValueErrors可能会变成Pydantic ValidationError
文档
迁移工具Alembic
设置数据库键命名约定
SQL优先,Pydantic次之
从一开始就设置异步测试客户端
使用ruff
额外部分
项目结构有很多种,但最好的结构是一致、直观且没有意外的。
许多示例项目和教程按文件类型(如crud、routers、models)划分项目,这种方式对于微服务或范围较小的项目很有效。但是,这种方法并不适合我们这个包含许多领域和模块的单体应用。
我发现对于这类情况,更具可扩展性和可演进性的结构是受Netflix的Dispatch启发,并做了一些小修改。
fastapi-project
├── alembic/
├── src
│ ├── auth
│ │ ├── router.py
│ │ ├── schemas.py # pydantic模型
│ │ ├── models.py # 数据库模型
│ │ ├── dependencies.py
│ │ ├── config.py # 本地配置
│ │ ├── constants.py
│ │ ├── exceptions.py
│ │ ├── service.py
│ │ └── utils.py
│ ├── aws
│ │ ├── client.py # 用于外部服务通信的客户端模型
│ │ ├── 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 # 全局配置
│ ├── models.py # 全局模型
│ ├── exceptions.py # 全局异常
│ ├── pagination.py # 全局模块,如分页
│ ├── database.py # 数据库连接相关内容
│ └── main.py
├── tests/
│ ├── auth
│ ├── aws
│ └── posts
├── templates/
│ └── index.html
├── requirements
│ ├── base.txt
│ ├── dev.txt
│ └── prod.txt
├── .env
├── .gitignore
├── logging.ini
└── alembic.ini
将所有领域目录存储在src文件夹中
src/ – 应用的最高级别,包含通用模型、配置和常量等。
src/main.py – 项目的根文件,用于初始化FastAPI应用
每个包都有自己的路由、模式、模型等。
router.py – 每个模块的核心,包含所有端点
schemas.py – 用于pydantic模型
models.py – 用于数据库模型
service.py – 模块特定的业务逻辑
dependencies.py – 路由依赖项
constants.py – 模块特定的常量和错误代码
config.py – 例如环境变量
utils.py – 非业务逻辑函数,例如响应规范化、数据丰富等
exceptions.py – 模块特定的异常,例如PostNotFound、InvalidUserData
当包需要其他包的服务、依赖项或常量时,使用显式的模块名导入
from src.auth import constants as auth_constants
from src.notifications import service as notification_service
from src.posts.constants import ErrorCode as PostsErrorCode # 以防每个包的constants模块中都有标准的ErrorCode
FastAPI首先是一个异步框架。它设计用于处理异步I/O操作,这也是它如此快速的原因。
然而,FastAPI并不限制你只能使用async路由,开发者也可以使用同步路由。这可能会让初学者误以为它们是一样的,但实际上并非如此。
在底层,FastAPI可以有效地处理异步和同步I/O操作。
FastAPI在线程池中运行同步路由,阻塞的I/O操作不会阻止事件循环执行任务。
如果路由定义为async,那么它会通过await正常调用,FastAPI相信你只会执行非阻塞的I/O操作。
需要注意的是,如果你违反了这种信任,在异步路由中执行阻塞操作,事件循环将无法在阻塞操作完成之前运行后续任务。
import asyncio
import time
from fastapi import APIRouter
router = APIRouter()
@router.get(“/terrible-ping”)
async def terrible_ping():
time.sleep(10) # 10秒的I/O阻塞操作,整个进程都会被阻塞
return {“pong”: True}
@router.get(“/good-ping”)
def good_ping():
time.sleep(10) # 10秒的I/O阻塞操作,但在单独的线程中运行整个`good_ping`路由
return {“pong”: True}
@router.get(“/perfect-ping”)
async def perfect_ping():
await asyncio.sleep(10) # 非阻塞I/O操作
return {“pong”: True}
当我们调用时会发生什么:
GET /terrible-ping
FastAPI服务器接收请求并开始处理
服务器的事件循环和队列中的所有任务都将等待time.sleep()完成
服务器认为time.sleep()不是I/O任务,所以会等待它完成
等待期间,服务器不会接受任何新请求
服务器返回响应。
响应之后,服务器开始接受新请求
GET /good-ping
FastAPI服务器接收请求并开始处理
FastAPI将整个路由good_ping发送到线程池,工作线程将在那里运行该函数
在good_ping执行期间,事件循环从队列中选择下一个任务并处理它们(例如接受新请求、调用数据库)
独立于主线程(即我们的FastAPI应用),工作线程将等待time.sleep完成。
同步操作只阻塞子线程,而不是主线程。
当good_ping完成工作后,服务器向客户端返回响应
GET /perfect-ping
FastAPI服务器接收请求并开始处理
FastAPI等待asyncio.sleep(10)
事件循环从队列中选择下一个任务并处理它们(例如接受新请求、调用数据库)
当asyncio.sleep(10)完成后,服务器完成路由的执行并向客户端返回响应
Warning
关于线程池的注意事项:
线程比协程需要更多资源,因此它们不像异步I/O操作那样轻量。
线程池的线程数量是有限的,也就是说,你可能会耗尽线程,导致应用变慢。了解更多(外部链接)
第二个需要注意的是,非阻塞的可等待对象或发送到线程池的操作必须是I/O密集型任务(例如打开文件、数据库调用、外部API调用)。
等待CPU密集型任务(例如繁重的计算、数据处理、视频转码)是没有意义的,因为CPU必须工作才能完成这些任务,而I/O操作是外部的,服务器在等待这些操作完成时什么也不做,因此它可以处理下一个任务。
在其他线程中运行CPU密集型任务也不是有效的,因为GIL(全局解释器锁)的存在。简而言之,GIL只允许一个线程同时工作,这使得它对CPU任务毫无用处。
如果你想优化CPU密集型任务,你应该将它们发送到另一个进程中的工作节点。
困惑用户的相关StackOverflow问题