FastAPI — 타입이 곧 계약이다 · 계약과 수명 · 讲解
타입 힌트 하나가 만드는 네 가지
一句话总结
在 FastAPI 中,类型提示不是注释,而是会执行的契约。写下一处类型提示,就能同时获得验证、序列化、文档和编辑器自动补全。
为什么要把契约写进代码
只存在于文档中的 API 规范,必然会与代码产生偏差。因为把某个字段改为可选时,很少有人会同步修改文档。
于是前端会在聊天中询问“这个字段可能返回 null 吗”,服务器也可能收到意料之外的请求体并以 500 崩溃。日志中只留下一行 KeyError,根本无法知道是谁发送了什么错误内容。如果开始在处理器里手写验证逻辑,这些代码又会在不同端点之间逐渐产生差异。
用类型提示把契约写在代码内部,文档、验证、错误响应和客户端类型都会来自同一处,从根本上消除了发生偏差的位置。
它究竟有什么不同
@app.post("/items")def create(item: Item) -> ItemOut: ...这一行会完成以下工作。
1. 请求验证——如果请求体不符合 Item 的形态,就会在进入处理器前以 422 拒绝
2. 序列化——按照 ItemOut 过滤返回值并生成 JSON
3. 文档——自动生成 /openapi.json 和 /docs
4. 类型检查——mypy 和编辑器会实际进行检查
如果使用 Flask,就要从 request.json 取出内容,手写 if "name" not in body:,再单独把规则写进文档,二者随后就会开始偏离。
422 与 400 不同
- 400 Bad Request——违反了业务规则(余额不足、邮箱重复)
- 422 Unprocessable Entity——输入的形态错误(应为字符串却传入数字)
FastAPI 会自动对模式违规返回 422,响应体中还会指出哪个字段为什么错误。
{"detail":[{"type":"int_parsing","loc":["body","qty"], "msg":"Input should be a valid integer","input":"many"}]}把 loc 原样交给客户端,就能免费获得按表单字段显示错误的能力。
response_model 是用于“剔除”字段的工具
这是最实用、也最容易被忽略的功能。
class User(BaseModel): email: str hashed_password: str # DB 모델에는 있다class UserOut(BaseModel): email: str # 나가는 쪽에는 없다@app.get("/me", response_model=UserOut)def me() -> User: ... # User 를 돌려줘도 UserOut 으로 걸러진다即使处理器误返回整个对象,响应中也不会出现 hashed_password。如果没有这一层保护,总有一天会有人写下 return user,使哈希值从 API 泄露。这是实际工作中经常发生的事故。
> 只遵守一条规则,就能挡住一半问题——输入模型和输出模型绝不能使用同一个类。
async def 与 def——服务器会在这里停住
FastAPI 同时支持两种声明方式,但它们的运行方式完全不同。
| 声明 | 在哪里执行 | 内部发生阻塞时 |
|---|---|---|
| async def | 直接在事件循环上 | 整个服务器停止响应 |
| def | 发送到线程池 | 只有该线程被阻塞 |
@app.get("/slow")async def slow(): time.sleep(1) # ❌ 이 1초 동안 모든 요청이 대기한다在 async def 中不能调用不带 await 的阻塞操作,包括 time.sleep、requests.get、同步数据库驱动和繁重的 CPU 计算。
有两种修复方式。
- 直接声明为
def→ FastAPI 会自动发送到线程池 - 使用异步库 →
httpx.AsyncClient、asyncpg
初学者为了“更快”而加上 async,结果反而让服务器串行运行,这是该框架最常见的陷阱。没有把握时,使用 def 通常更安全。
依赖注入
Depends 用于声明“这个处理器需要该对象”。
def get_db(): con = connect() try: yield con # 핸들러가 쓰는 동안 finally: con.close() # 응답을 보낸 뒤 정리된다@app.get("/items")def items(db = Depends(get_db)): ...使用 yield 后,清理代码一定会在响应完成后执行。而它真正的价值会在测试中体现出来。
app.dependency_overrides[get_db] = lambda: FakeDB()无需修改任何一行生产代码就能替换依赖,也不需要猴子补丁。
lifespan——on_event 已经过时
@asynccontextmanagerasync def lifespan(app): app.state.pool = await make_pool() # 시작할 때 yield await app.state.pool.close() # 끝날 때app = FastAPI(lifespan=lifespan)@app.on_event("startup") 已计划废弃。新代码应使用 lifespan,这里适合打开连接池和缓存客户端。
实务中会踩到的问题
BackgroundTasks 不是队列。它会在响应发送后,继续在同一个进程中执行。工作进程重启时,任务就会消失。它只适合发送邮件这类可以丢失的任务;不能丢失的任务应发送到 Redis/Kafka。
工作进程数量。uvicorn --workers N 会启动 N 个进程。每个进程拥有独立内存,不会共享全局变量。如果把内存缓存放在全局变量中,每个工作进程都会持有不同的值。
同步端点的线程池大小是有限的(默认 40)。当所有线程都在阻塞时,后续请求只能在队列中等待。def 不是万能方案,只是一层缓冲。