前一章里,我们已经让一个最小应用跑了起来。现在先别急着继续堆功能,我们把注意力放到一次请求身上:浏览器或者命令行发出请求之后,FastAPI 怎么知道该调用哪个函数?路径里的数字从哪里来?请求体为什么会自动变成 Python 对象?返回字典之前,框架又做了哪些检查?
这一章会围绕一个小型商品目录接口逐步展开。我们会创建商品、筛选商品、读取单个商品、修改库存,再故意发送几次错误请求,直接观察状态码、响应体和响应头。每看到一个现象,再向里拆一层机制。这样学完之后,你记住的不会只是一组装饰器,而是一条完整的请求处理链。
本章示例使用 Python 3.10 及以上版本的类型语法和 Pydantic v2 的方法名。示例数据暂存在内存中,服务一重启就会恢复初始值。这正好让我们集中观察框架行为;数据库、认证和依赖注入会在后面的章节里单独展开。
先看一条最短的路由:
from fastapi import FastAPI
app = FastAPI()
@app.get("/health")
def health():
return {"ready": True}运行服务后访问 /health,你会收到:
{
"ready": true
}这几行代码真正登记的是一个二元条件:请求方法是 GET,路径模式是 /health。两个条件同时匹配,FastAPI 才会调用 health。因此,路由不只是一个网址,也不只是一个函数名。你可以把它理解成通讯录里的一行记录:
GET + /health -> health()同一路径可以按请求方法分流。比如 GET /api/products 表示读取商品列表,POST /api/products 表示创建商品。路径相同,意图不同,进入的处理函数也不同:
@app.get("/api/products")
def list_products():
return []
@app.post("/api/products")
def create_product():
return {"message": "created"}
HTTP 把“想对资源做什么”放在请求方法里。常见方法可以先按下面的工程语义理解:
这些不是为了让接口看起来整齐才约定的。客户端、缓存、代理和接口工具都会根据请求方法理解请求意图。比如读取操作不应该偷偷删除数据;删除成功并且没有返回内容时,204 比“返回一个空字典再配 200”表达得更准确。
FastAPI 提供了与这些方法对应的装饰器:@app.get()、@app.post()、@app.put()、@app.patch() 和 @app.delete()。装饰器中的路径负责识别资源,装饰器本身负责识别操作。
商品接口常同时拥有这两条路由:
@app.get("/api/products/featured")
def featured_product():
return {"name": "彩色键盘"}
@app.get("/api/products/{product_id}")
def get_product(product_id: int):
return {"id": product_id}第一条是静态路径,第二条含有动态片段 {product_id}。请求 /api/products/8 时,字符串 8 会被捕获并交给 product_id。请求 /api/products/featured 时,我们希望它进入推荐商品路由,而不是把 featured 当成商品编号。
路由按注册顺序参与匹配,所以更具体的静态路径应该写在更宽泛的动态路径之前。否则动态路由可能先接住请求,然后因为 featured 不能转换为整数而返回校验错误。这个规则在 /users/me 与 /users/{user_id}、/files/latest 与 /files/{file_id} 这样的组合中同样适用。
在路径里写 {product_id},只是告诉路由器这里允许变化;在函数签名里写 product_id: int,才进一步告诉 FastAPI 这个值必须按整数解析:
from typing import Annotated
from fastapi import Path
@app.get("/api/products/{product_id}")
def get_product(
product_id: Annotated[int, Path(ge=1)],
):
return {"id": product_id}这里的 Path(ge=1) 又增加了一条运行时约束:商品编号必须大于等于 1。于是下面三个请求会得到不同结果:
/api/products/8 -> product_id 是整数 8
/api/products/abc -> 无法解析为整数,校验失败
/api/products/0 -> 能解析为整数,但不满足大于等于 1路径参数一般不做成可选项。路径少一段,命中的就是另一条 URL;如果某个筛选条件可以省略,它更适合放在查询参数里。
当商品路由越来越多,如果每条都从 /api/products 开始手写,容易出现前缀不一致,也会让主应用文件变得拥挤。APIRouter 可以先收集一组相关路由,再一次性挂到应用上:
from fastapi import APIRouter, FastAPI
app = FastAPI()
router = APIRouter(prefix="/api/products", tags=["商品"])
@router.get("")
def list_products():
return []
@router.get("/{product_id}")
def get_product(product_id: int):
prefix 会加在路由器里的每条路径前面,tags 会把这些操作归到接口文档的同一组。APIRouter 自己不是第二个 Web 应用,也不会单独启动;它只是帮助我们组织路由定义。到了大型项目,可以把商品、订单、用户分别放进不同模块,最后仍由同一个 FastAPI 实例统一接收请求。
请求进入正确的路由之后,FastAPI 会读取处理函数的签名,判断每个参数应该从哪里获取。最常见的三种来源是路径、查询字符串和请求体:
@router.patch("/{product_id}")
def patch_product(
product_id: int,
notify: bool = False,
payload: ProductPatch | None = None,
):
...FastAPI 会按规则解释这三个参数:
product_id 的名字出现在路径模板中,所以来自路径。notify 是一个简单类型,不在路径模板中,所以来自查询字符串。payload 是 Pydantic 模型,所以来自 JSON 请求体。对应的请求可以写成:
PATCH /api/products/8?notify=true
Content-Type: application/json
{"price": 269}函数里拿到的已经不是原始文本:product_id 是 int,notify 是 bool,payload 是 ProductPatch 实例。参数定位、读取、转换和校验都发生在处理函数执行之前。

问号后面的内容属于查询字符串。下面这条请求含有四个查询值,其中 tag 出现了两次:
/api/products?q=键盘&tag=外设&tag=桌面&limit=10可以这样接收:
from typing import Annotated, Literal
from fastapi import Query
@router.get("")
def list_products(
q: Annotated[str | None, Query(min_length=1, max_length=20)] = None,
tag: Annotated[list[str] | None, Query()] = None,
limit: Annotated[
这里同时出现了四类约束:
q 默认是 None,所以可以不传;一旦传入,长度必须在 1 到 20 之间。tag 可以重复出现,FastAPI 会把它们收集成 list[str]。limit 默认是 10,并且只能在 1 到 50 之间。order 只能是 price 或 name,其他字符串会被拒绝。“类型允许 None”和“参数可以不传”是两个相关但不同的概念。真正决定请求中能否缺少该参数的,是有没有默认值。q: str | None = None 同时表达了“缺少时用 None”和“值允许是 None”;如果只写 q: str | None 而没有默认值,客户端仍然必须提供 q。
创建商品时,客户端发送的不再是一个零散值,而是一组有内部结构的数据。Pydantic 模型正适合描述这个边界:
from pydantic import BaseModel, Field
class ProductCreate(BaseModel):
name: str = Field(min_length=2, max_length=40)
price: float = Field(gt=0)
stock: int = Field(ge=0)
tags: list[str] = Field(把它放到函数参数中,FastAPI 就会从请求体读取 JSON:
@router.post("")
def create_product(payload: ProductCreate):
return payload下面的请求会通过校验:
{
"name": "便携支架",
"price": 88,
"stock": 5,
"tags": ["桌面"]
}进入函数后,payload.price 是浮点数,payload.stock 是整数,payload.tags 是字符串列表。需要得到普通字典时,Pydantic v2 使用 payload.model_dump():
product_data = payload.model_dump()default_factory=list 会为每个模型实例创建一份新列表,比把可变列表直接当默认值更能清楚表达意图。Pydantic 会妥善处理模型字段的默认值,但在普通 Python 函数和类里,我们仍应避免共享可变默认对象。
创建商品时,名称、价格和库存都应该出现;部分更新时,客户端可能只发一个字段。与其把创建模型的全部字段都改成可选,不如为更新单独定义一个模型:
class ProductPatch(BaseModel):
name: str | None = Field(default=None, min_length=2, max_length=40)
price: float | None = Field(default=None, gt=0)
stock: int | None = Field(default
处理 PATCH 时,只提取客户端真正提供过的字段:
@router.patch("/{product_id}")
def patch_product(product_id: int, payload: ProductPatch):
product = PRODUCTS[product_id]
update_data = payload.model_dump(exclude_unset=True)
product.update(update_data)
return productexclude_unset=True 很关键。假设客户端只发送 {"stock": 12},更新字典里就只有 stock,不会把其余字段的默认 None 覆盖到旧数据上。
还要留意“没有提供”和“明确提供 null”的区别:
{}表示没有要求修改任何字段;而:
{
"name": null
}表示客户端明确把 name 设为 null。当前模型允许 None,所以第二种输入会进入更新字典。真实项目里是否允许清空名称,是业务规则;如果不允许,就应让类型和验证规则直接拒绝它,而不是在路由里猜客户端的意图。
在普通 Python 代码里,类型标注通常不会自动阻止错误参数。但 FastAPI 会读取类型标注,再结合 Pydantic 完成一条运行时流程:
路由器先根据请求方法和路径找到处理函数,并从路径、查询字符串、请求头或请求体中收集原始输入。
框架根据函数签名和模型字段,把原始文本或 JSON 数据交给相应的类型适配与校验逻辑。
校验成功后,处理函数拿到转换后的 Python 值;校验失败时,处理函数不会执行,框架直接构造错误响应。
同一份类型与约束还会进入接口描述,使交互式文档知道参数位置、必填状态、数据结构和允许范围。
所以,product_id: int 不是给编辑器看的装饰。它同时影响运行时输入、错误信息和接口文档。类型写得越含糊,框架能替你做的事情就越少。
现在故意访问一个错误地址:
curl -i http://127.0.0.1:8000/api/products/not-a-number响应状态是 422 Unprocessable Content,响应体类似这样:
{
"detail": [
{
"type": "int_parsing",
"loc": ["path", "product_id"],
"msg": "Input should be a valid integer, unable to parse string as an integer",
"input": "not-a-number"
}
]
}先别被字段数量吓到。定位这类错误,最有用的是三部分:
loc 说明错误位置。这里是路径中的 product_id。type 说明错误类别。这里是整数解析失败。input 保留导致失败的输入,便于客户端对照。如果请求体里的 price 是负数,loc 会指向 body 和 price;如果漏掉必填字段,错误类型会变成缺少字段。统一的结构让前端可以按字段展示错误,也让测试不必依赖一整句可能随版本调整的英文消息。
422 表示请求已经到达应用,媒体类型和整体语法可以被处理,但内容不满足接口声明。它不是“服务器崩了”,也不等同于业务上的“商品不存在”。输入结构错误交给自动校验,资源不存在则应由业务代码返回 404,两者要分开。
输入从 URL 过来时天然是字符串,FastAPI 和 Pydantic 会按声明尝试转换。例如 limit=10 可以变成整数 10,常见形式的布尔字符串可以变成布尔值。转换的目的,是把合理的传输格式变成函数需要的 Python 类型。
但不要把它理解成“任何相近内容都会被猜对”。无法确定含义的字符串仍会失败,字段范围仍会检查,Literal 仍只接受列出的值。对于金额、标识符或安全相关字段,如果业务要求不接受隐式转换,还可以使用更严格的类型配置。核心原则是:接口边界应该明确,转换只能处理约定内的表示方式,不能替业务规则做猜测。
price > 0、name 长度至少为 2、stock >= 0 这类规则只依赖请求数据本身,很适合放在模型字段里。它们在进入路由之前就能完成。
“商品编号是否存在”“当前库存能不能预留”“用户名是否已经占用”则需要查询当前系统状态,属于业务校验。它们应该在处理函数或业务服务中完成,并返回恰当的业务错误。把两层分开之后,模型保持可复用,路由也更容易说明为什么失败。
输入有模型,输出同样需要边界。假设内存里的商品记录包含供应商成本:
PRODUCTS = {
1: {
"id": 1,
"name": "彩色键盘",
"price": 299.0,
"stock": 8,
"tags": ["外设", "桌面"],
"supplier_cost": 168.0,
}
}客户端不应该看到 supplier_cost。我们先定义公开模型:
from pydantic import BaseModel, ConfigDict
class ProductPublic(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: int
name: str
price: float
stock: int
tags: list[str]再把它交给 response_model:
@router.get("/{product_id}", response_model=ProductPublic)
def get_product(product_id: int):
return PRODUCTS[product_id]虽然函数返回的原始字典里有 supplier_cost,客户端收到的内容只包含公开模型声明的字段:
{
"id": 1,
"name": "彩色键盘",
"price": 299.0,
"stock": 8,
"tags": ["外设", "桌面"]
}
response_model 至少承担三件事:
因此,它是一道真正的输出边界。密码哈希、内部成本、风控标记这类字段不应先放进“大而全”的响应模型,再到每个路由里临时排除。更稳妥的做法是为公开场景定义专门模型,让默认行为就是不泄露。
ConfigDict(from_attributes=True) 让模型也可以从对象属性读取字段。当前示例返回的是字典,不依赖这项配置;等后面返回数据库对象时,同一个公开模型仍然能继续使用。
你也可以这样声明返回类型:
@router.get("/{product_id}")
def get_product(product_id: int) -> ProductPublic:
return ProductPublic.model_validate(PRODUCTS[product_id])返回类型标注能帮助编辑器和类型检查器,FastAPI 也可以据此生成响应结构。response_model 则允许函数内部返回更宽的对象,再由框架按公开模型过滤。实际项目可以选一种清晰一致的风格;如果内部返回类型和公开类型确实不同,显式写 response_model 往往更直观。
创建成功时,使用 201 Created:
from fastapi import status
@router.post(
"",
response_model=ProductPublic,
status_code=status.HTTP_201_CREATED,
)
def create_product(payload: ProductCreate):
...删除成功又没有响应体时,使用 204 No Content:
from fastapi import Response
@router.delete(
"/{product_id}",
status_code=status.HTTP_204_NO_CONTENT,
)
def delete_product(product_id: int) -> Response:
del PRODUCTS[product_id]
return Response(status_code=status.HTTP_204_NO_CONTENT)204 的语义就是没有响应内容,所以不要再返回 {"message": "deleted"}。如果产品需要给客户端一段删除结果,就改用 200 并定义响应模型;不要让状态码和响应体互相矛盾。
响应正文适合业务数据,响应头适合请求追踪、缓存策略和处理时间等元数据。可以让 FastAPI 注入 Response,再给它添加响应头:
from fastapi import Response
@app.get("/health")
def health(response: Response):
response.headers["X-Service-Ready"] = "true"
return {"ready": True}如果很多路由都要添加同一个头,就不该在每个处理函数里复制这段代码,中间件更适合做这类跨路由逻辑。后面我们会把追踪头移到中间件里。
类型校验只能说明“输入长得对不对”,不能判断商品是否存在。读取商品时,我们需要主动处理资源缺失:
from fastapi import HTTPException
@router.get("/{product_id}", response_model=ProductPublic)
def get_product(product_id: int):
product = PRODUCTS.get(product_id)
if product is None:
raise HTTPException(status_code=404, detail="商品不存在")
return product访问 /api/products/999,你会看到:
HTTP/1.1 404 Not Found
content-type: application/json{
"detail": "商品不存在"
}这里用 raise,不是 return HTTPException(...)。抛出异常会立刻中断当前处理函数,交给 FastAPI 的异常处理层生成响应;把异常对象返回,只会把它当作普通返回值继续序列化。
可以先用下面这组判断建立直觉:
不要把所有失败都塞进 200 响应,再在 JSON 里写 success: false。这样会让 HTTP 层告诉客户端“成功”,业务正文又告诉客户端“失败”,监控、重试和客户端 SDK 都很难正确处理。
库存不足可能在多个接口里出现。我们可以先定义一个只表达业务事实的异常:
class StockConflict(Exception):
def __init__(self, product_id: int) -> None:
self.product_id = product_id再注册一个全局处理器,把它统一转换成 409:
from fastapi import Request, status
from fastapi.responses import JSONResponse
@app.exception_handler(StockConflict)
async def stock_conflict_handler(
request: Request,
exc: StockConflict,
):
return JSONResponse(
status_code=status.HTTP_409_CONFLICT,
content={"detail": f"商品 {exc.product_id} 库存不足"},
)路由里只保留业务判断:
@router.post("/{product_id}/reserve", response_model=ProductPublic)
def reserve_product(product_id: int):
product = PRODUCTS.get(product_id)
if product is None:
raise HTTPException(status_code=404, detail="商品不存在")
if product["stock"] < 1:
raise StockConflict(product_id)
这样做的价值不是少写两行代码,而是让同一种业务失败始终拥有相同状态码和响应结构。异常处理器还可以统一记录请求路径,但不要把堆栈、数据库语句或内部配置直接回传给客户端。
不要用一个捕获所有 Exception 的处理器把任何错误都改成 200 或模糊的 400。未预期异常应该被日志和监控看见,并以服务端错误处理。过度吞掉异常会让故障表面上“很安静”,排查时却失去最重要的证据。
FastAPI 允许路由处理函数写成 async def,也允许写成普通 def。两种都能响应请求,差别出现在函数等待外部资源时。
import asyncio
@app.get("/wait")
async def wait_for_io():
await asyncio.sleep(0.01)
return {"mode": "async", "result": "done"}执行到 await 时,当前协程暂停,把事件循环的执行机会让给其他就绪任务。等计时器、数据库或网络操作完成,协程再从原位置继续。这不是把一段函数拆到多个 CPU 核心上,也不是让单个请求的计算自动变快;它是在等待期间不占住同一条事件循环通道。

如果数据库驱动或 HTTP 客户端提供异步方法,路由可以写成:
@app.get("/reports/{report_id}")
async def get_report(report_id: int):
report = await report_repository.fetch(report_id)
summary = await remote_service.fetch_summary(report_id)
return {"report": report, "summary": summary}await 只能用于可等待对象,普通函数不能因为前面加了 await 就变成异步。判断的关键不是“这一步看起来耗时”,而是所用库有没有提供真正的异步接口。
如果你使用的是同步数据库驱动或者同步 SDK,可以把处理函数写成普通 def:
@app.get("/legacy-report/{report_id}")
def get_legacy_report(report_id: int):
return legacy_client.fetch(report_id)FastAPI 底层的 Starlette 会把同步端点放到线程池中运行,避免它直接阻塞事件循环。线程池不是无限资源;同步请求很多、每个又阻塞很久时,线程池仍可能排队。因此,“写成 def”是正确隔离同步库的办法,不代表同步阻塞从此没有成本。
下面的写法看起来只有一个词不同,运行特性却完全不同:
import time
@app.get("/bad-wait")
async def bad_wait():
time.sleep(2)
return {"done": True}time.sleep(2) 不会把控制权交回事件循环。它会在这两秒里阻塞当前事件循环线程,其他协程也无法正常推进。异步函数里要使用对应的异步等待:
import asyncio
@app.get("/good-wait")
async def good_wait():
await asyncio.sleep(2)
return {"done": True}真实项目中的同步文件处理、同步网络 SDK 和耗时压缩同样需要留意。可以改用异步库,也可以明确放到线程池或任务系统中,不能只把函数签名改成 async def 就假设它不会阻塞。
图片编码、大规模数值计算和复杂压缩主要消耗 CPU,不是在等网络。即使把它们写进异步函数,也没有可让出的等待点。较长的 CPU 任务通常应交给进程池、独立任务进程或专用计算服务,避免占住 Web 请求处理资源。
可以用一个简化判断表:
这里没有“所有路由都必须加 async”的规则。选对库、识别等待点,比统一追求某种函数写法更重要。
当应用运行后,访问 /openapi.json 会得到一份机器可读的接口描述。FastAPI 会把路由、参数来源、Pydantic 模型、状态码、标签和说明整理成 OpenAPI 文档;/docs 与 /redoc 再读取这份描述,渲染成两种不同的页面。

这条关系可以写成:
路由声明 + 类型标注 + 数据模型 + 响应说明
↓
OpenAPI 描述
↓
交互式文档、阅读文档、客户端工具所以,自动文档不是扫描函数名后拼出来的一张静态网页。真正的源头仍是代码里的接口契约。你把 limit 的上限从 50 改成 100,重新启动应用后,接口描述和文档表单都会跟着变化。
创建应用时可以填写标题、版本和描述:
app = FastAPI(
title="商品目录接口",
version="1.0.0",
description="观察 FastAPI 核心请求与响应流程的最小应用。",
)请求 /openapi.json 时,info 部分会是:
{
"title": "商品目录接口",
"description": "观察 FastAPI 核心请求与响应流程的最小应用。",
"version": "1.0.0"
}这些信息面向接口使用者,应该描述服务是什么,而不是写实现日志或部署口令。
@router.post(
"",
response_model=ProductPublic,
status_code=status.HTTP_201_CREATED,
tags=["商品"],
summary="创建商品",
response_description="创建后的公开商品数据",
)
async def create_product(payload: ProductCreate):
"""接收一个新商品,并分配商品编号。"""
...装饰器参数控制接口描述的结构化部分,函数文档字符串可以补充较长说明。摘要要短,描述要说明行为和边界,不要把源代码逐句翻译一遍。
如果路由可能返回 404 或 409,仅仅在代码里 raise HTTPException,交互式测试当然能看到真实响应,但成功模型之外的错误结构不一定被完整描述。生产接口可以用装饰器的 responses 参数补充错误响应模型。这里先理解一个原则:自动文档会忠实反映你声明过的内容,不会猜出所有业务分支。
另外,公开生产服务是否开放 /docs、/redoc 和 /openapi.json,应根据安全和运维策略决定。隐藏文档不能替代认证授权;开放文档也不意味着敏感接口可以没有访问控制。
假设我们希望每个响应都有追踪标记和处理时间。把同样的两行代码塞进所有路由当然能运行,但很快就会重复。中间件可以在路由处理前后统一执行:
from time import perf_counter
from fastapi import Request
@app.middleware("http")
async def add_trace_header(request: Request, call_next):
started = perf_counter()
response = await call_next(request)
response.headers["X-Lesson-Trace"] = "fastapi-core"
response.headers["X-Process-Time"] = f"{perf_counter() - started:.6f
请求进入时,先执行 call_next 之前的部分;call_next(request) 把请求继续传给后续中间件和路由;路由产生响应后,控制权沿调用链返回,再执行添加响应头的部分。

常见用途包括:
业务规则通常不适合塞进中间件。比如“当前用户能不能修改这一笔订单”需要知道路由语义、资源和用户权限,更适合依赖注入或业务服务。中间件越靠外,掌握的业务上下文越少。
多个中间件叠加时,请求方向逐层向内,响应方向逐层向外。假设外层负责追踪,内层负责计时,执行顺序可以理解成:
追踪中间件:请求阶段
计时中间件:请求阶段
路由处理函数
计时中间件:响应阶段
追踪中间件:响应阶段这也是为什么中间件顺序会影响结果。比如异常处理、跨域响应头和自定义响应修改需要结合完整调用链考虑。不要只凭“添加顺序”猜请求和响应两边的执行顺序,最好写一个小测试观察关键头部与异常响应。
浏览器前端和 API 不同源时,常用 CORSMiddleware:
from fastapi.middleware.cors import CORSMiddleware
app.add_middleware(
CORSMiddleware,
allow_origins=["https://shop.example.com"],
allow_credentials=True,
allow_methods=["GET", "POST", "PATCH", "DELETE"],
allow_headers=["Authorization", "Content-Type"],
)开发阶段把所有来源、方法和头部全部放开很省事,但生产环境应列出真正需要访问 API 的前端来源。跨域限制是浏览器安全模型的一部分,不是服务端认证;非浏览器客户端并不会因为没有跨域响应头就失去请求能力。
中间件处理的是“每一个请求”,生命周期处理的是“整个应用从启动到关闭”。数据库连接池、共享 HTTP 客户端、模型加载和后台任务管理器都不应该在每次请求里重新创建。
FastAPI 推荐用异步上下文管理器定义生命周期:
from contextlib import asynccontextmanager
from fastapi import FastAPI
@asynccontextmanager
async def lifespan(app: FastAPI):
app.state.ready = True
app.state.next_id = 3
yield
app.state.ready = False
app = FastAPI(lifespan=lifespan)yield 之前是启动阶段。它完成后,应用才开始正常接收请求。yield 之后是关闭阶段,适合释放连接、停止任务和刷新缓冲数据。
启动逻辑只运行一次,而不是每次请求运行一次。共享资源可以放到 app.state,请求里通过 request.app.state 访问:
from fastapi import Request
@app.get("/health")
def health(request: Request):
return {
"ready": request.app.state.ready,
"mode": "sync",
}这里的 app.state 是应用级容器。它适合保存连接池或客户端等共享对象,不适合把每个用户的临时状态堆在里面。多个请求会并发访问共享对象,写入共享可变状态时必须考虑竞争条件。内存字典只是教学演示,不能直接替代数据库。
使用 TestClient 时,推荐把它放在 with 块中:
from fastapi.testclient import TestClient
with TestClient(app) as client:
response = client.get("/health")
assert response.json()["ready"] is True
assert app.state.ready is False进入 with 时运行启动阶段,离开时运行关闭阶段。直接创建客户端却不进入上下文,容易让依赖生命周期的测试出现与真实启动过程不一致的行为。
现在把前面的零件合到一个文件里。保存为 main.py:
from __future__ import annotations
import asyncio
from contextlib import asynccontextmanager
from time import perf_counter
from typing import Annotated, Literal
from fastapi import (
APIRouter,
FastAPI,
HTTPException,
Path,
Query,
Request,
Response,
status,
)
from fastapi.responses import JSONResponse
这份代码把本章的核心机制放在了同一条链上:
response_model 负责公开输出。HTTPException 处理直接的 HTTP 失败,自定义异常处理业务冲突。await 处让出执行权。安装依赖后启动应用:
python -m uvicorn main:app --reload终端保持运行,再开一个终端发送请求。先看同步健康检查:
curl -i http://127.0.0.1:8000/health状态行和关键响应如下:
HTTP/1.1 200 OK
content-type: application/json
x-lesson-trace: fastapi-core
x-process-time: 0.108438
{"ready":true,"mode":"sync"}处理时间每次都会不同,不要把示例中的数字当作性能基准。我们真正要确认的是:路由返回了 JSON,中间件添加的两个响应头也存在。
curl -i -X POST http://127.0.0.1:8000/api/products \
-H 'Content-Type: application/json' \
-d '{"name":"桌面灯带","price":49.9,"stock":6,"tags":["桌面"]}'响应是:
HTTP/1.1 201 Created
content-type: application/json
x-lesson-trace: fastapi-core{
"id": 3,
"name": "桌面灯带",
"price": 49.9,
"stock": 6,
"tags": ["桌面"]
}函数内部还添加了 supplier_cost,但响应里没有它。这不是我们在 return 前手动删除的结果,而是 ProductPublic 过滤了公开边界之外的字段。
curl -G http://127.0.0.1:8000/api/products \
--data-urlencode 'tag=外设' \
--data-urlencode 'tag=桌面' \
--data-urlencode 'limit=1'结果只保留同时含有两个标签的商品,并按 limit=1 截取:
[
{
"id": 1,
"name": "彩色键盘",
"price": 299.0,
"stock": 8,
"tags": ["外设", "桌面"]
}
]路径无法转换为整数:
curl -i http://127.0.0.1:8000/api/products/not-a-number得到 422 Unprocessable Content,错误位置指向 path 中的 product_id。
编号格式正确,但资源不存在:
curl -i http://127.0.0.1:8000/api/products/999得到:
HTTP/1.1 404 Not Found{
"detail": "商品不存在"
}商品存在,但库存已经是 0:
curl -i -X POST http://127.0.0.1:8000/api/products/2/reserve得到:
HTTP/1.1 409 Conflict{
"detail": "商品 2 库存不足"
}这三个状态码分别来自三层:框架输入校验、路由资源判断、业务异常处理器。只看“请求失败了”会把它们混在一起;看清失败层次,客户端才知道应该修改输入、换一个资源,还是等待资源状态改变。
curl -i -X DELETE http://127.0.0.1:8000/api/products/1关键输出是:
HTTP/1.1 204 No Content
x-lesson-trace: fastapi-core
x-process-time: 0.000515状态行后没有 JSON 正文,这与 204 的语义一致。中间件仍然能给响应添加头部,因为响应仍会沿调用链向外返回。
手动 curl 适合观察,自动测试适合防止以后改代码时把契约悄悄改坏。新建 test_main.py:
from fastapi.testclient import TestClient
from main import app
def test_core_request_flow():
with TestClient(app) as client:
health = client.get("/health")
assert health.status_code == 200
assert health.json() == {
"ready": True,
"mode": "sync",
}
assert
这里的测试没有直接调用 get_product()。它让请求经过 ASGI 应用,因此路由匹配、参数校验、中间件、异常处理、响应模型和生命周期都会参与。测试通过时,不只是证明某个函数返回了字典,而是证明客户端能观察到的接口行为仍符合约定。
测试错误响应时,优先断言状态码、错误位置和稳定的业务字段,不要把整个自动校验消息逐字锁死。Pydantic 的错误说明可能改进措辞,而 loc 所表达的字段位置更接近我们真正关心的契约。
现在再访问一次:
PATCH /api/products/1?notify=true
Content-Type: application/json
{"price": 269}你应该能按顺序说出发生了什么:
请求先进入应用最外层的中间件。中间件记录开始时间,再把请求交给下一层。
路由器同时比较请求方法和路径。它找到 PATCH /api/products/{product_id} 对应的处理函数,并捕获路径中的 1。
FastAPI 读取函数签名,把 product_id 从路径转换为整数,把 notify 从查询字符串转换为布尔值,把 JSON 请求体交给 ProductPatch。
自动文档并不在每次请求经过的主链路中,但它与这条链使用同一份声明:路由、参数类型、模型约束和响应结构。正因为运行时和文档共享契约,代码变动后文档才能同步变化。
写 FastAPI 接口时,可以先问四个问题:请求怎样命中路由,输入从哪里来,失败用什么状态表达,输出允许公开哪些字段。四个边界清楚之后,再决定同步或异步、是否放进中间件、哪些资源交给生命周期管理,代码通常就不会散。
下一章会把依赖注入接进这条请求链。到那时,数据库会话、当前用户和通用查询条件不需要在每个路由里手动创建;但它们仍会遵守本章建立的路由、校验、异常、响应与生命周期边界。
Pydantic 检查 price 大于 0。校验成功后,处理函数拿到模型实例;如果失败,函数不会执行,应用直接返回 422。
处理函数查找商品,只用 model_dump(exclude_unset=True) 取出客户端提供的 price,然后更新当前记录。
返回值经过 ProductPublic 校验与过滤,内部字段不会进入公开响应,序列化结果变成 JSON。
响应沿中间件链返回。中间件添加追踪标记和处理时间,服务器再把状态行、响应头和响应体写回客户端。