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

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

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

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

# Flask 写一个带校验的 POST 接口
from flask import Flask, request, jsonify
from marshmallow import Schema, fields, ValidationError

app = Flask(__name__)

class BookSchema(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'])
def create_book():
# 手动解析 JSON
data = request.get_json()

# 手动校验
schema = BookSchema()
try:
book = schema.load(data)
except ValidationError as e:
return jsonify({'errors': e.messages}), 422

# 业务逻辑
return jsonify({'title': book['title'], 'price': book['price']})

# 还没写文档呢……

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

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

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

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

# FastAPI 做同样的事
from fastapi import FastAPI
from pydantic import BaseModel, Field

app = FastAPI()

class BookIn(BaseModel):
title: str = Field(..., min_length=2, max_length=50)
price: float = Field(..., gt=0)

@app.post('/books')
async def create_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 调用、文件读写）提升巨大。

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

2.2 类型注解驱动

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

@app.get('/items/{item_id}')
async def read_item(
item_id: int, # 路径参数，自动转 int
q: str | None = None, # 查询参数，可选
book: BookIn, # 请求体，自动解析 JSON
token: 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 自动完成校验、转换和序列化：

from pydantic import BaseModel, Field, EmailStr
from datetime import datetime

class User(BaseModel):
id: int
name: 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}')
async def get_user(user_id: int): # 访问 /users/abc 自动返回 422
return {'user_id': user_id}

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

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

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

@app.post('/books/')
async def create_book(book: BookIn): # BookIn 继承自 BaseModel
return {'id': 1, **book.model_dump()}

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

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

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

from fastapi import Form, File, UploadFile

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

3.2 参数校验体系

FastAPI 的参数校验分两层：

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

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

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

from typing import Annotated
from fastapi import Query, Path

# Annotated 写法（推荐，Python 3.9+）
@app.get('/items/')
async def read_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() 承担同样的职责：

class Book(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，框架会：

class UserIn(BaseModel):
username: str
password: str # 输入时需要
email: str

class UserOut(BaseModel):
username: str
email: str # 响应时不暴露密码

@app.post('/users/', response_model=UserOut)
async def create_user(user: UserIn):
return user # password 会被自动过滤掉

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

from fastapi import status

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

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

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

from fastapi.responses import StreamingResponse

async def file_iterator(file_path: str, chunk_size: int = 8192):
"""逐块读取文件，避免内存溢出"""
with open(file_path, 'rb') as f:
while chunk := f.read(chunk_size):
yield chunk

@app.get('/stream/file')
async def stream_large_file():
return StreamingResponse(
content=file_iterator('./large_file.zip'),
media_type='application/octet-stream',
)

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

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

@app.get('/stream/sse')
async def sse_stream():
async def sse_generator():
for i in range(10):
yield f'data: 第 {i} 条消息\\n\\n'.encode('utf-8')

return StreamingResponse(
content=sse_generator(),
media_type='text/event-stream',
headers={
'Cache-Control': 'no-cache',
'Connection': 'keep-alive',
}
)

3.4 异常处理

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

from fastapi import HTTPException

@app.get('/items/{item_id}')
async def read_item(item_id: str):
if item_id not in items:
raise HTTPException(
status_code=404,
detail='Item not found',
headers={'X-Error': 'not_found'},
)
return {'item': items[item_id]}

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

from fastapi import Request
from fastapi.responses import JSONResponse

class BusinessError(Exception):
def __init__(self, code: int, message: str):
self.code = code
self.message = message

@app.exception_handler(BusinessError)
async def business_error_handler(request: Request, exc: BusinessError):
return JSONResponse(
status_code=exc.code,
content={'code': exc.code, 'message': exc.message},
)

# 使用
raise BusinessError(400, '库存不足')

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

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

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

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

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

from typing import Annotated
from fastapi import Depends

# 定义依赖函数
async def common_params(skip: int = 0, limit: int = 10):
return {'skip': skip, 'limit': limit}

# 注入到路径操作
@app.get('/items/')
async def read_items(params: Annotated[dict, Depends(common_params)]):
return params

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

def get_query(q: str | None = None):
return q

def get_query_or_empty(q: Annotated[str, Depends(get_query)]):
return q or 'empty'

@app.get('/search/')
async def search(query: Annotated[str, Depends(get_query_or_empty)]):
return {'query': query}

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

async def get_db():
db = DatabaseSession()
try:
yield db # 请求期间使用这个 db
finally:
await db.close() # 请求结束后自动关闭

@app.get('/users/')
async def read_users(db: Annotated[DatabaseSession, Depends(get_db)]):
return db.query(User).all()

依赖的作用范围：

依赖覆盖（测试用）：测试时可以替换依赖，比如用 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 可帮不了你。

# 错误做法：敏感字段输出了才过滤
class UserOut(BaseModel):
username: str
email: str

@app.post('/users/', response_model=UserOut)
async def create_user(user: UserIn):
print(user.password) # 密码已经在内存中了
return user

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

4.3 Depends 的作用范围选择

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

# 比较好的分层做法
# 1. 公开路由：不需要依赖
@app.get('/health')
async def health_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 选型

4.6 官方文档

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

5. 总结

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

FastAPI 的核心设计哲学可以用一句话概括：让类型注解替你干活。当你习惯用 Annotated、BaseModel、Depends 来表达意图时，你会发现写 API 不再是"参数搬运工"，而是真正在写业务逻辑。

下篇预告：FastAPI 进阶篇——中间件机制、安全认证（OAuth2 + JWT）、后台任务、WebSocket 实时通信与项目工程化。这些内容会让你的 FastAPI 项目从"能用"变成"能上线"。