当前位置：首页 > news >正文

FastAPI 基础篇：类型注解驱动的 Python Web 开发范式

news 2026/7/1 3:06:58

FastAPI 基础篇：类型注解驱动的 Python Web 开发范式

1. 引子：写 API 为什么这么累？

先问一个问题：用 Flask 或 Django 写一个带参数校验和文档的 POST 接口，需要写多少代码？

# Flask 写一个带校验的 POST 接口fromflaskimportFlask,request,jsonifyfrommarshmallowimportSchema,fields,ValidationError app=Flask(__name__)classBookSchema(Schema):title=fields.String(required=True,validate=validate.Length(min=2,max=50))price=fields.Float(required=True,validate=validate.Range(min=0))@app.route('/books',methods=['POST'])defcreate_book():# 手动解析 JSONdata=request.get_json()# 手动校验schema=BookSchema()try:book=schema.load(data)exceptValidationErrorase:returnjsonify({'errors':e.messages}),422# 业务逻辑returnjsonify({'title':book['title'],'price':book['price']})# 还没写文档呢……

这份代码暴露了传统框架的典型痛点：

手动解析请求体
手动定义校验规则（且校验和路由是分离的）
手动维护API 文档（或者额外装 flasgger/Swagger）
手动序列化响应

如果每个接口都这样写，项目中 30% 的代码都在做"参数搬运"。更要命的是，校验代码、文档注释、业务逻辑散落在三个地方，修改一个字段要改三处。

FastAPI 解决这个问题的思路很激进：你只需要写 Python 类型注解，剩下的框架替你搞定。

# FastAPI 做同样的事fromfastapiimportFastAPIfrompydanticimportBaseModel,Field app=FastAPI()classBookIn(BaseModel):title:str=Field(...,min_length=2,max_length=50)price:float=Field(...,gt=0)@app.post('/books')asyncdefcreate_book(book:BookIn):return{'title':book.title,'price':book.price}

做完了。校验、文档、序列化全部自动完成，打开http://127.0.0.1:8000/docs就能看到交互式文档。

2. 核心概念：FastAPI 凭什么能这么干？

FastAPI 不是凭空冒出来的东西，它的能力建立在三个技术支柱上：

FastAPI = Starlette（异步网络能力） + Pydantic（数据校验） + 类型注解驱动

2.1 ASGI 与异步

在 FastAPI 之前，Python Web 框架基本都跑在 WSGI（Web Server Gateway Interface）上，这是 Python 在 2003 年提出的标准。WSGI 的工作方式是：来一个请求，开一个线程，处理完再响应——同步阻塞。

ASGI（Asynchronous Server Gateway Interface）是 2016 年提出的新一代标准，它允许服务器在处理一个请求的同时，去处理另一个请求。这对 IO 密集型场景（数据库查询、外部 API 调用、文件读写）提升巨大。

对比	WSGI（Flask/Django 传统模式）	ASGI（FastAPI/Starlette）
处理方式	同步阻塞，一个请求占一个线程	异步非阻塞，事件循环调度
并发能力	靠多线程，线程切换开销大	靠协程，轻量级切换
长连接	不支持 WebSocket 原生	原生支持 WebSocket/SSE
性能	中等	高（接近 Node.js/Go 的水平）

FastAPI 跑在 Starlette 之上，Starlette 是一个轻量级的 ASGI 框架/工具包，提供了路由、中间件、WebSocket 等底层能力。FastAPI 在 Starlette 之上加了一层类型注解驱动的 API 层，这才是它的核心竞争力。

2.2 类型注解驱动

Python 的类型注解（Type Hints）从 Python 3.5 开始引入，最初只是为了给 IDE 做静态检查。FastAPI 把这个特性玩出了新高度：类型注解不再是"注释"，而是框架行为的"指令"。

@app.get('/items/{item_id}')asyncdefread_item(item_id:int,# 路径参数，自动转 intq:str|None=None,# 查询参数，可选book:BookIn,# 请求体，自动解析 JSONtoken:str=Header(None),# 请求头session_id:str=Cookie(None),# Cookie):return{'item_id':item_id,'q':q}

每一处注解都在告诉 FastAPI 三件事：

参数从哪里来（路径、查询、请求体、请求头……）
应该是什么类型（自动校验 + 转换）
是否可选（有默认值 = 可选，没有 = 必填）

2.3 Pydantic：背后的守门员

Pydantic 是 FastAPI 用来做数据校验的引擎。每当 FastAPI 接收到请求数据，都会交给 Pydantic 模型去做校验。

Pydantic 的核心机制是BaseModel——你定义一个类，声明字段和类型，Pydantic 自动完成校验、转换和序列化：

frompydanticimportBaseModel,Field,EmailStrfromdatetimeimportdatetimeclassUser(BaseModel):id:intname:str=Field(...,min_length=2,max_length=20)email:EmailStr created_at:datetime|None=None# 传入的 JSON 会自动校验和转换# {"id": "123", "name": "张三", "email": "test@example.com"}# → id 自动从字符串 "123" 转为整数 123# → email 格式不对直接返回 422 错误# 输出的对象自动序列化为 JSON# → 日期时间自动转为 ISO 格式字符串# → 定义 response_model 时可以过滤敏感字段

2.4 自动文档的原理

FastAPI 在应用启动时，遍历所有注册的路由，根据你的类型注解自动生成 OpenAPI（原 Swagger）规范的 JSON 文件。然后基于这个 JSON 渲染出两个交互式文档：

Swagger UI（/docs）：可以在这个页面直接调试接口
ReDoc（/redoc）：更清晰的可读性文档

这意味着：你不需要单独维护一份文档。修改了参数类型，文档自动更新。代码即文档。

3. 核心体系：FastAPI 的请求生命周期

理解 FastAPI 的最好方式，是跟踪一个请求从进入到返回的全过程。下图展示了这个生命周期：

在这个生命周期中，核心知识点分布在五个层面，下面逐一拆解。

3.1 路由与参数体系

FastAPI 的路由定义非常直观：@app.get()、@app.post()等装饰器绑定 URL 路径和 HTTP 方法。参数从哪里来，由函数签名中的类型和默认值决定：

路径参数：用{}包裹变量名，FastAPI 自动从 URL 中提取，按类型注解做转换。

@app.get('/users/{user_id}')asyncdefget_user(user_id:int):# 访问 /users/abc 自动返回 422return{'user_id':user_id}

查询参数：函数参数中不属于路径占位符的，自动识别为查询参数。

@app.get('/items/')asyncdeflist_items(skip:int=0,# 可选，默认 0limit:int=10,# 可选，默认 10category:str,# 必选，没有默认值):return{'skip':skip,'limit':limit,'category':category}

请求体参数：Pydantic 模型类型的参数，自动从 JSON 请求体中解析。

@app.post('/books/')asyncdefcreate_book(book:BookIn):# BookIn 继承自 BaseModelreturn{'id':1,**book.model_dump()}

请求头与 Cookie：使用Header()和Cookie()显式声明。

@app.get('/secure')asyncdefsecure_endpoint(token:str=Header(...,alias='Authorization'),session_id:str=Cookie(None),):return{'valid':True}

表单与文件：使用Form()和UploadFile。

fromfastapiimportForm,File,UploadFile@app.post('/login')asyncdeflogin(username:str=Form(...),password:str=Form(...),avatar:UploadFile=File(None),):return{'username':username}

3.2 参数校验体系

FastAPI 的参数校验分两层：

第一层：类型注解自带的校验

item_id:int# 自动校验是否为整数price:float# 自动校验是否为浮点数

第二层：Query()/Path()/Field()提供的增强校验

fromtypingimportAnnotatedfromfastapiimportQuery,Path# Annotated 写法（推荐，Python 3.9+）@app.get('/items/')asyncdefread_items(q:Annotated[str|None,Query(min_length=3,max_length=50,pattern='^[a-zA-Z0-9]+$',description='搜索关键词',)]=None,page:Annotated[int,Query(ge=1)]=1,):pass

注意：Annotated写法的优势是类型信息完整，IDE 能正确提示。旧的q: str = Query(default=None, min_length=3)写法把默认值和校验规则混在一起，IDE 可能会误以为q总是字符串。

在 Pydantic 模型中，Field()承担同样的职责：

classBook(BaseModel):title:str=Field(...,min_length=2,max_length=100,description='书名')price:float=Field(...,gt=0,le=10000,description='价格')tags:list[str]=Field(default=[],max_length=5)

gt（大于）、ge（大于等于）、lt（小于）、le（小于等于）、min_length、max_length、pattern（正则）——这些校验参数覆盖了 90% 的日常需求。如果还不够，可以用AfterValidator写自定义校验函数。

3.3 响应处理

FastAPI 的响应处理有三大机制：

响应模型（response_model）：这是 FastAPI 的杀手锏之一。通过声明response_model，框架会：

自动过滤掉不在模型中的字段
自动做类型转换和校验
自动生成文档

classUserIn(BaseModel):username:strpassword:str# 输入时需要email:strclassUserOut(BaseModel):username:stremail:str# 响应时不暴露密码@app.post('/users/',response_model=UserOut)asyncdefcreate_user(user:UserIn):returnuser# password 会被自动过滤掉

响应状态码：通过status_code指定，推荐使用fastapi.status中的常量。

fromfastapiimportstatus@app.post('/items/',status_code=status.HTTP_201_CREATED)asyncdefcreate_item():return{'message':'created'}

响应类型家族：FastAPI 提供了多种响应类，覆盖不同场景：

响应类	Content-Type	适用场景
`JSONResponse`（默认）	`application/json`	常规 JSON 数据
`HTMLResponse`	`text/html`	返回 HTML 页面
`PlainTextResponse`	`text/plain`	返回纯文本
`RedirectResponse`	3xx 状态码	URL 重定向
`FileResponse`	自动识别	文件下载
`StreamingResponse`	自定义	流式输出、大文件下载、SSE

流式响应（StreamingResponse）值得单独拿出来讲，因为它在 LLM 应用、大文件下载场景下非常实用：

fromfastapi.responsesimportStreamingResponseasyncdeffile_iterator(file_path:str,chunk_size:int=8192):"""逐块读取文件，避免内存溢出"""withopen(file_path,'rb')asf:whilechunk:=f.read(chunk_size):yieldchunk@app.get('/stream/file')asyncdefstream_large_file():returnStreamingResponse(content=file_iterator('./large_file.zip'),media_type='application/octet-stream',)

核心思想是：不用一次性把整个文件读到内存，而是边读边发。文件大小和内存占用无关。

SSE（Server-Sent Events）是另一种流式响应，用于服务端向客户端单向推送：

@app.get('/stream/sse')asyncdefsse_stream():asyncdefsse_generator():foriinrange(10):yieldf'data: 第{i}条消息\n\n'.encode('utf-8')returnStreamingResponse(content=sse_generator(),media_type='text/event-stream',headers={'Cache-Control':'no-cache','Connection':'keep-alive',})

3.4 异常处理

FastAPI 的异常处理遵循"即抛即停"原则：raise HTTPException后，当前请求立即终止，返回指定的状态码和错误信息。

fromfastapiimportHTTPException@app.get('/items/{item_id}')asyncdefread_item(item_id:str):ifitem_idnotinitems:raiseHTTPException(status_code=404,detail='Item not found',headers={'X-Error':'not_found'},)return{'item':items[item_id]}

自定义异常处理器可以统一处理业务异常：

fromfastapiimportRequestfromfastapi.responsesimportJSONResponseclassBusinessError(Exception):def__init__(self,code:int,message:str):self.code=code self.message=message@app.exception_handler(BusinessError)asyncdefbusiness_error_handler(request:Request,exc:BusinessError):returnJSONResponse(status_code=exc.code,content={'code':exc.code,'message':exc.message},)# 使用raiseBusinessError(400,'库存不足')

这样整个应用的错误响应格式就统一了。

3.5 依赖注入：FastAPI 的设计精髓

依赖注入是 FastAPI 与其他框架拉开差距的关键特性。一句话解释：

“你的函数需要什么，就声明什么，FastAPI 负责帮你取来。”

基础用法：定义一个普通的函数作为依赖，通过Depends()注入到路径操作中。

fromtypingimportAnnotatedfromfastapiimportDepends# 定义依赖函数asyncdefcommon_params(skip:int=0,limit:int=10):return{'skip':skip,'limit':limit}# 注入到路径操作@app.get('/items/')asyncdefread_items(params:Annotated[dict,Depends(common_params)]):returnparams

依赖链：依赖可以依赖其他依赖，形成链式调用。

defget_query(q:str|None=None):returnqdefget_query_or_empty(q:Annotated[str,Depends(get_query)]):returnqor'empty'@app.get('/search/')asyncdefsearch(query:Annotated[str,Depends(get_query_or_empty)]):return{'query':query}

yield 依赖（资源管理）：当一个依赖需要管理资源（如数据库连接）时，用yield代替return，yield之前的代码在请求前执行，之后的代码在响应后执行。

asyncdefget_db():db=DatabaseSession()try:yielddb# 请求期间使用这个 dbfinally:awaitdb.close()# 请求结束后自动关闭@app.get('/users/')asyncdefread_users(db:Annotated[DatabaseSession,Depends(get_db)]):returndb.query(User).all()

依赖的作用范围：

范围	写法	生效范围
全局	`app = FastAPI(dependencies=[Depends(auth)])`	所有路由
模块级	`APIRouter(dependencies=[Depends(auth)])`	该模块所有路由
路由级	`@router.get('/', dependencies=[Depends(auth)])`	单个路由，不注入返回值
参数级	`func(param = Depends(func))`	单个函数参数

依赖覆盖（测试用）：测试时可以替换依赖，比如用 Mock 数据库替换真实数据库：

app.dependency_overrides[get_db]=override_get_db# 测试期间，所有用到 get_db 的地方都会使用 override_get_db

4. 避坑指南 / 最佳实践

4.1 推荐使用 Annotated 写法

新旧两种写法对比：

# 旧写法：默认值和校验混在一起q:str=Query(default=None,min_length=3,max_length=50)# 新写法：类型、校验、默认值位置清晰q:Annotated[str|None,Query(min_length=3,max_length=50)]=None

新写法有两大好处：

IDE 类型提示更准确（旧写法中 IDE 可能认为q总是str，导致后续代码误报）
参数结构更清晰：类型注解中的Query()只负责校验规则，最后的= None才是默认值

4.2 response_model 的安全问题

用response_model过滤敏感字段时要注意：FastAPI 是在数据返回前做过滤的，如果你在路径操作函数内部就把用户密码打印到了日志里，response_model可帮不了你。

# 错误做法：敏感字段输出了才过滤classUserOut(BaseModel):username:stremail:str@app.post('/users/',response_model=UserOut)asyncdefcreate_user(user:UserIn):print(user.password)# 密码已经在内存中了returnuser

最佳实践：在定义 Pydantic 模型时，输入模型和输出模型分开设计，输入模型包含所有字段（包括敏感信息），输出模型只包含需要暴露的字段。

4.3 Depends 的作用范围选择

遵循最小范围原则：能用参数级就不用全局。全局依赖会影响到所有路由，包括健康检查接口、静态文件等本不需要鉴权的路径。

# 比较好的分层做法# 1. 公开路由：不需要依赖@app.get('/health')asyncdefhealth_check():return{'status':'ok'}# 2. 业务路由模块：统一加模块级鉴权admin_router=APIRouter(prefix='/admin',dependencies=[Depends(verify_admin_token)],)

4.4 用 APIRouter 组织项目

不要让main.py变成一个上千行的文件。按业务模块拆分：

app/ ├── main.py # 入口：初始化 App，挂载路由 ├── api/ │ └── v1/ │ ├── api.py # 汇总层 │ └── endpoints/ │ ├── books.py # 图书模块 │ └── users.py # 用户模块 ├── schemas/ # Pydantic 模型 ├── models/ # SQLAlchemy 模型（后续数据库篇会讲） └── core/ # 配置、安全等

4.5 SSE vs WebSocket 选型

场景	选哪个	原因
推送通知、实时数据	SSE	单向足够，浏览器自动重连，实现简单
聊天、协作编辑	WebSocket	双向通信，低延迟
LLM 流式输出	SSE	服务端单向推送，客户端用 EventSource 接收
在线游戏	WebSocket	需要高频双向交互

4.6 官方文档

FastAPI 官方文档（中文）
Pydantic 官方文档
Starlette 官方文档

5. 总结

下表总结了 FastAPI 基础篇的核心概念和对应的 API：

你要做什么	用 FastAPI 的什么	一句话
定义路由	`@app.get()`/`@app.post()`	装饰器绑定 URL + HTTP 方法
路径参数	`{id}`+ 类型注解	自动提取 URL 变量并校验类型
查询参数	`Query()`/ 默认值	自动提取`?key=value`
请求体验证	Pydantic`BaseModel`	自动解析 JSON + 校验 + 文档
响应过滤	`response_model`	自动剔除敏感字段
流式输出	`StreamingResponse`	逐块返回，不占内存
异常处理	`HTTPException`	即抛即停，返回标准错误
依赖管理	`Depends()`	一次定义，随处注入
模块化路由	`APIRouter`	按业务拆分，`include_router`聚合
跨域	`CORSMiddleware`	`add_middleware`一行配置