LabHub
学习 学习路径 课程

FastAPI — 타입이 곧 계약이다 · 계약과 수명 · 讲解

타입 힌트 하나가 만드는 네 가지

在 LabHub 中继续学习

一句话总结

在 FastAPI 中,类型提示不是注释,而是会执行的契约。写下一处类型提示,就能同时获得验证、序列化、文档和编辑器自动补全。

流程图: 类型提示不是注释,而是会执行的契约。 · 代码内部 · 请求验证 · 422

为什么要把契约写进代码

只存在于文档中的 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 不同

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.sleeprequests.get、同步数据库驱动和繁重的 CPU 计算。

有两种修复方式。

初学者为了“更快”而加上 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 不是万能方案,只是一层缓冲。