4. FastAPI核心机制与请求处理
FastAPI 的请求处理链路可以概括为一条“声明式装配”的流水线:开发者用类型注解和 Pydantic 模型描述接口契约,框架在启动时把这些声明编译成校验器、序列化器和依赖图,请求到达后再按图执行。理解这条链路,比记住某个装饰器的写法更重要。本节以“用户管理接口”的增删改查(CRUD)为例,把路由声明、路径参数、查询参数、请求体、响应模型和依赖注入串成一个可运行的实现。
一、路由声明与路径参数:把资源标识写进 URL
路由声明使用 @app.get、@app.post 等装饰器,第一个参数是路径模板。路径参数用花括号占位,例如 /users/{user_id}。关键点是类型注解:user_id: int 会让 FastAPI 自动把字符串片段转换为整数,转换失败直接返回 422,而不是进入函数体。这属于“声明即校验”,无需手写 try/except。
路径参数适合表达“唯一标识某个资源”,因此用户 ID、订单号这类值应放在路径中;而过滤、分页、排序等修饰性条件应放在查询参数中。若把 user_id 写成查询参数,URL 语义会变得含糊,也不利于缓存与日志分析。
二、查询参数:默认值、可选性与约束
函数签名中未出现在路径模板里的普通参数,会被 FastAPI 识别为查询参数。例如 skip: int = 0 与 limit: int = 10 提供默认值,调用 GET /users 时即使不带参数也能工作;q: str | None = None 表示可选搜索关键字。若需要更严格的约束,可用 Query(ge=1, le=100) 限定范围,用 Query(min_length=1, max_length=32) 限定字符串长度。约束失败同样返回 422,并在响应体中给出字段路径与错误原因,便于前端定位。
三、请求体与响应模型:输入输出分离
请求体用于承载结构化数据,通常用 Pydantic 模型声明。以用户为例,可以定义三个模型:UserCreate 用于创建(含 name、email、age),UserUpdate 用于更新(所有字段可选),UserOut 用于响应(含 id,但不含密码等敏感字段)。这种拆分让“输入契约”和“输出契约”各自独立,避免把内部字段意外暴露给客户端。
响应模型通过 response_model=UserOut 指定。FastAPI 会用该模型对返回值做过滤与序列化:即使函数返回了包含密码的字典,响应中也只会出现 UserOut 声明的字段。这是比“手动删字段”更可靠的安全边界。同时,response_model 还会驱动 OpenAPI 文档生成,使 /docs 中的响应结构准确可用。
四、依赖注入系统:把公共逻辑抽离为可复用单元
依赖注入(Dependency Injection)是 FastAPI 最具工程价值的部分。用 Depends 声明的函数会在请求进入路径函数前被调用,其结果作为参数注入。典型用途包括:数据库会话管理、当前用户解析、分页参数复用、权限校验。依赖可以嵌套:get_current_user 依赖 get_db,路径函数再依赖 get_current_user,框架会自动按拓扑顺序解析并缓存同一请求内的重复依赖。
下面给出一个可运行的用户管理接口实现,使用内存字典模拟存储,便于聚焦机制本身:
from fastapi import FastAPI, Depends, HTTPException, Query
from pydantic import BaseModel, EmailStr, Field
app = FastAPI(title="User Service")
class UserCreate(BaseModel):
name: str = Field(min_length=1, max_length=32)
email: EmailStr
age: int = Field(ge=0, le=150)
class UserUpdate(BaseModel):
name: str | None = Field(default=None, min_length=1, max_length=32)
email: EmailStr | None = None
age: int | None = Field(default=None, ge=0, le=150)
class UserOut(BaseModel):
id: int
name: str
email: EmailStr
age: int
_db: dict[int, dict] = {}
_seq = 0
def get_db() -> dict[int, dict]:
return _db
def next_id() -> int:
global _seq
_seq += 1
return _seq
@app.post("/users", response_model=UserOut, status_code=201)
def create_user(payload: UserCreate, db=Depends(get_db)):
uid = next_id()
db[uid] = {"id": uid, **payload.model_dump()}
return db[uid]
@app.get("/users", response_model=list[UserOut])
def list_users(
skip: int = Query(0, ge=0),
limit: int = Query(10, ge=1, le=100),
q: str | None = None,
db=Depends(get_db),
):
items = list(db.values())
if q:
items = [u for u in items if q.lower() in u["name"].lower()]
return items[skip: skip + limit]
@app.get("/users/{user_id}", response_model=UserOut)
def get_user(user_id: int, db=Depends(get_db)):
if user_id not in db:
raise HTTPException(status_code=404, detail="user not found")
return db[user_id]
@app.put("/users/{user_id}", response_model=UserOut)
def update_user(user_id: int, payload: UserUpdate, db=Depends(get_db)):
if user_id not in db:
raise HTTPException(status_code=404, detail="user not found")
data = payload.model_dump(exclude_unset=True)
db[user_id].update(data)
return db[user_id]
@app.delete("/users/{user_id}", status_code=204)
def delete_user(user_id: int, db=Depends(get_db)):
if user_id not in db:
raise HTTPException(status_code=404, detail="user not found")
db.pop(user_id)
这段代码体现了几个容易忽略的细节。payload.model_dump(exclude_unset=True) 只返回客户端真正提交的字段,因此 PUT 可以安全地做部分更新,而不会把未传字段覆盖为 None。status_code=201 与 204 分别对应创建成功与删除成功,符合 HTTP 语义。HTTPException 抛出后由框架统一转换为 JSON 错误响应,无需手动构造。
五、请求处理链路与结果判读
把上述机制串起来,一次 POST /users 的完整处理顺序是:路由匹配 → 解析并校验请求体 → 解析依赖(get_db)→ 执行路径函数 → 用 UserOut 过滤序列化 → 返回 201。若请求体缺少 email 或 age 超出范围,会在进入函数前返回 422,函数体不会执行。若访问不存在的用户,则返回 404 与 {"detail": "user not found"}。判断接口是否按预期工作,可以依次检查:状态码是否符合语义、响应字段是否与 UserOut 一致、错误响应是否包含可定位的字段信息。
最后补充一个工程习惯:把 get_db 换成真实的数据库会话依赖时,应使用 yield 形式,在请求结束后自动关闭会话;把 get_current_user 作为依赖挂到需要鉴权的路由上,可以让权限校验与业务逻辑解耦。依赖注入的价值不在于少写几行代码,而在于让数据库、认证、分页这些横切关注点拥有统一的声明位置和可测试的替换点。