Article
第一章:FastAPI 入门与环境搭建
1.1 FastAPI 简介与核心优势(异步、自动文档、类型提示)
| 方法/特性名称 | 语法/说明 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| 异步支持(async/await) | async def func(): / await some_async_op() | 实现非阻塞 I/O,提升高并发性能 | async def read_data(): / data = await db.fetch() / return data | 适用于数据库、网络请求等耗时操作 |
| 自动生成 API 文档 | 访问 /docs 或 /redoc | 基于 OpenAPI 和 JSON Schema 生成交互式文档 | 启动应用后访问:http://localhost:8000/docs | 无需额外配置,实时更新 |
| 类型提示(Type Hints) | param: str, return -> int | 提高代码可读性,支持自动验证与补全 | def greet(name: str) -> str: / return f"Hello {name}" | 配合 Pydantic 实现请求/响应数据校验 |
| Pydantic 集成 | 继承 BaseModel | 定义请求体、响应体结构,自动验证数据 | class User(BaseModel): / name: str / age: int | 字段类型错误会自动返回 422 响应 |
| 高性能 | 基于 Starlette 与 Pydantic | 框架本身开销小,支持高吞吐量 | 性能接近 Node.js 和 Go | 适合构建微服务和高性能 API |
1.2 安装 FastAPI 与运行服务器(Uvicorn)
| 方法/命令名称 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| 安装 FastAPI | pip install fastapi | 安装 FastAPI 核心框架 | pip install fastapi | 通常需搭配 ASGI 服务器使用 |
| 安装 Uvicorn | pip install "uvicorn[standard]" | 安装 ASGI 服务器用于运行应用 | pip install uvicorn | [standard] 包含推荐的依赖(如 uvloop) |
| 运行应用 | uvicorn app:app | 启动 Uvicorn 服务器 | uvicorn main:app --reload | app:app 表示模块名:应用实例名 |
| 查看帮助 | uvicorn --help | 查看所有运行参数选项 | uvicorn --help | 可查看 host、port、workers 等配置 |
| 安装生产依赖 | pip install "fastapi[all]" | 一次性安装所有可选依赖 | pip install "fastapi[all]" | 包括 pydantic、uvicorn、jinja2 等 |
1.3 第一个 FastAPI 应用:Hello World
| 方法/组件名称 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| 导入 FastAPI | from fastapi import FastAPI | 引入框架核心类 | from fastapi import FastAPI | 必须第一步导入 |
| 创建应用实例 | app = FastAPI() | 初始化 FastAPI 应用 | app = FastAPI() | 通常命名为 app |
| 定义 GET 路由 | @app.get("/") | 绑定函数到指定路径的 GET 请求 | @app.get("/") / def read_root(): / return {"Hello": "World"} | 装饰器语法,路径 / 表示根路径 |
| 返回字典 | return {"key": "value"} | 自动序列化为 JSON 响应 | return {"message": "OK"} | 支持 dict、list、str、int 等基本类型 |
| 启动服务器 | uvicorn.run() | 在代码中直接启动(可选) | if __name__ == "__main__": / uvicorn.run(app, host="127.0.0.1", port=8000) | 多用于开发或调试,生产环境推荐命令行启动 |
1.4 启动配置:host、port、reload、debug 模式
| 配置项/参数 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| host | --host 0.0.0.0 | 指定服务器监听的主机地址 | uvicorn main:app --host 0.0.0.0 | 0.0.0.0 表示允许外部访问 |
| port | --port 8080 | 指定服务器监听端口 | uvicorn main:app --port 8080 | 默认为 8000 |
| reload | --reload | 开启热重载,代码修改自动重启 | uvicorn main:app --reload | 仅用于开发环境,生产禁用 |
| debug 模式 | app = FastAPI(debug=True) | 启用调试模式,输出详细错误 | app = FastAPI(debug=True) | 与 --reload 配合使用更佳 |
| workers | --workers 4 | 指定工作进程数(生产) | uvicorn main:app --workers 4 | 多进程提升并发处理能力,需配合 Gunicorn |
1.5 项目结构与模块化初识
| 结构/方法名称 | 语法/布局 | 用途 | 示例结构 | 注意事项 |
|---|---|---|---|---|
| 基础项目结构 | . / ├── main.py / ├── routers/ / │ └── items.py / └── models/ / └── item.py | 组织代码,便于维护 | 推荐使用包结构管理大型项目 | 使用 __init__.py 可选(Python 3.3+) |
| APIRouter | from fastapi import APIRouter / router = APIRouter() / @router.get("/items") / def read_items(): ... | 将路由分组,实现模块化 | from fastapi import APIRouter | 在 main.py 中通过 app.include_router() 注册 |
| include_router | app.include_router(router) | 将子路由挂载到主应用 | app.include_router(router, prefix="/api/v1", tags=["items"]) | 可设置统一前缀、标签等元信息 |
| 模块导入 | from .routers import items | 相对导入子模块 | from .routers.items import router as items_router / app.include_router(items_router) | 确保 PYTHONPATH 正确或使用 -m 运行 |
| 主入口文件 | main.py | 应用启动入口 | 包含 FastAPI() 实例和路由注册 | 通常不包含具体业务逻辑 |
第二章:路由与路径操作函数
2.1 使用 @app.get()、@app.post() 等定义路由
| 方法名称 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
@app.get() | @app.get("/path") | 定义处理 GET 请求的路由 | @app.get("/") / def read_root(): / return {"message": "Hello"} | 用于获取资源,不包含请求体 |
@app.post() | @app.post("/path") | 定义处理 POST 请求的路由 | @app.post("/items") / def create_item(name: str): / return {"name": name} | 通常用于创建资源,支持请求体 |
@app.put() | @app.put("/path") | 定义处理 PUT 请求的路由 | @app.put("/items/{item_id}") / def update_item(item_id: int, name: str): / return {"item_id": item_id, "name": name} | 用于完整更新资源 |
@app.delete() | @app.delete("/path") | 定义处理 DELETE 请求的路由 | @app.delete("/items/{item_id}") / def delete_item(item_id: int): / return {"deleted": item_id} | 用于删除指定资源 |
@app.patch() | @app.patch("/path") | 定义处理 PATCH 请求的路由 | @app.patch("/items/{item_id}") / def partial_update(item_id: int, name: str = None): / return {"item_id": item_id, "name": name} | 用于部分更新资源,非幂等 |
2.2 路径参数(Path Parameters)与类型声明
| 方法名称 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| 路径参数定义 | {param_name} | 在 URL 中声明动态参数 | @app.get("/users/{user_id}") | 参数名需与函数参数一致 |
| 类型提示 | param_name: int / str / float | 自动转换路径参数类型 | def read_user(user_id: int): / return {"user_id": user_id} | 支持 int, str, float, bool 等 |
| 枚举类型 | param: Literal["value1", "value2"] | 限制路径参数取值范围 | from typing import Literal / @app.get("/cats/{name}") / def read_cat(name: Literal["garfield", "felix"]): / return {"name": name} | 使用 Literal 实现枚举约束 |
| 路径包含斜杠 | {file_path:path} | 捕获包含斜杠的完整路径 | @app.get("/files/{file_path:path}") / def read_file(file_path: str): / return {"file_path": file_path} | :path 表示匹配任意长度路径 |
2.3 查询参数(Query Parameters)与可选值处理
| 方法名称 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| 查询参数 | 函数参数未在路径中 | 从 URL 查询字符串提取参数 | @app.get("/items") / def list_items(skip: int = 0, limit: int = 10): / return {"skip": skip, "limit": limit} | 默认值决定是否为可选参数 |
| 可选参数 | param: Optional[str] = None | 显式声明可选查询参数 | from typing import Optional / def search(q: Optional[str] = None): / return {"q": q} | 必须导入 Optional |
| 必填查询参数 | param: str = ... | 声明无默认值的必填查询参数 | def read_item(item_id: str, needy: str): / return {"item_id": item_id, "needy": needy} | 缺少时返回 422 错误 |
| 多值查询参数 | param: List[str] = Query([]) | 接收多个同名查询参数 | from fastapi import Query / def tags(tags: list[str] = Query([])): / return {"tags": tags} | 需导入 Query 并设置默认值 |
2.4 多种 HTTP 方法支持(PUT、DELETE、PATCH 等)
| 方法名称 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| PUT | @app.put("/item/{id}") | 替换整个资源 | @app.put("/items/{item_id}") / def update_item(item_id: int, item: Item): / return {"item_id": item_id, **item.dict()} | 幂等操作,完全更新 |
| DELETE | @app.delete("/item/{id}") | 删除资源 | @app.delete("/items/{item_id}") / def remove_item(item_id: int): / return {"status": "deleted"} | 通常无请求体 |
| PATCH | @app.patch("/item/{id}") | 部分更新资源 | @app.patch("/items/{item_id}") / def update_name(item_id: int, name: str): / return {"item_id": item_id, "name": name} | 非幂等,只更新提供字段 |
| HEAD | @app.head("/path") | 获取响应头信息(无正文) | @app.head("/items") / def head_items(): / return {} | 自动由 GET 路由生成 |
| OPTIONS | @app.options("/path") | 获取资源支持的 HTTP 方法 | @app.options("/items") / def options_items(): / return {} | 框架自动处理,通常无需手动定义 |
2.5 路径顺序与优先级
| 概念 | 说明 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| 字面量路径优先 | 先定义具体路径 | 避免被通配路径拦截 | @app.get("/users/me") / def read_me(): ... / @app.get("/users/{user_id}") / def read_user(user_id: int): ... | /users/me 必须在 /users/{user_id} 之前 |
| 类型转换不影响优先级 | 路径字符串匹配优先 | 确保正确路由分发 | @app.get("/static/{path}") / def static_file(path: str): ... / @app.get("/static/data") / def data(): ... | /static/data 不会匹配到 path 参数 |
| 通配路径最后定义 | 使用 {path:path} 匹配剩余路径 | 实现静态文件或代理路由 | @app.get("/{full_path:path}") / def catch_all(full_path: str): / return {"path": full_path} | 必须放在所有路由最后 |
第三章:请求数据处理
3.1 使用 Pydantic 定义请求体模型(BaseModel)
| 方法名称 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| BaseModel 继承 | class Model(BaseModel): | 定义结构化数据模型 | class Item(BaseModel): / name: str / price: float | 所有字段需有类型注解 |
| 必填字段 | field: Type | 无默认值的字段为必填 | class User(BaseModel): / email: str / password: str | 请求中必须提供 |
| 可选字段 | field: Type = default | 设置默认值使字段可选 | class User(BaseModel): / is_active: bool = True | 可省略,使用默认值 |
| 高级类型 | List, Dict, Optional, Union | 使用复杂数据类型 | from typing import List, Optional / class Order(BaseModel): / items: List[str] / note: Optional[str] = None | 需从 typing 导入 |
3.2 处理 JSON 请求体(Body 参数)
| 方法名称 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| 请求体参数 | def func(item: Item) | 接收 JSON 格式的请求体 | @app.post("/items") / def create_item(item: Item): / return item | 自动解析 JSON 并验证 |
| 多个请求体 | def func(a: ModelA, b: ModelB) | 接收多个模型(不推荐) | FastAPI 不直接支持多个 Body | 需封装到一个模型中 |
| 嵌套请求体 | class Outer(BaseModel): / inner: InnerModel | 处理复杂嵌套结构 | class Item(BaseModel): / name: str / tags: list[str] = [] / @app.post("/items") / def create_item(item: Item): ... | 支持任意层级嵌套 |
| Body 函数 | Body(..., embed=True) | 强制将单个参数包装为 JSON 对象 | @app.post("/items") / def create_item(name: str = Body(..., embed=True)): / return {"name": name} | 用于单个参数也需 JSON 包装的场景 |
3.3 获取查询参数(Query 参数)
| 方法名称 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| Query 函数 | param: str = Query(default) | 增强查询参数功能 | from fastapi import Query / def search(q: str = Query(None, min_length=3)): ... | 可设置验证、别名等 |
| 参数别名 | Query(..., alias="item-query") | 使用 URL 不友好参数名 | def read_item(item_query: str = Query(..., alias="item-query")): ... | 代码中用 Python 风格命名 |
| 多值参数 | param: List[str] = Query([]) | 接收重复的查询参数 | def tags(tags: list[str] = Query([])): ... | URL: ?tags=1&tags=2 |
| 参数验证 | Query(..., min_length=3) | 对查询参数进行约束 | def search(q: str = Query(..., min_length=3, max_length=50)): ... | 错误时返回 422 |
3.4 获取路径参数(Path 参数)
| 方法名称 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| Path 函数 | param: int = Path(...) | 为路径参数添加验证 | from fastapi import Path / def get_item(item_id: int = Path(..., ge=1)): ... | ... 表示必填 |
| 数值约束 | ge, gt, le, lt | 设置数值范围 | item_id: int = Path(..., ge=1, le=100) | ge=大于等于,gt=大于 |
| 字符串约束 | min_length, max_length | 设置字符串长度 | name: str = Path(..., min_length=2) | 适用于路径中的字符串参数 |
| 参数描述 | Path(..., title="ID") | 添加参数元信息 | user_id: int = Path(..., title="用户ID") | 出现在文档中 |
3.5 获取 Cookie、Header 参数(Cookie、Header)
| 方法名称 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| Header 函数 | param = Header(...) | 从请求头中提取值 | from fastapi import Header / def get_token(x_token: str = Header(...)): ... | 自动处理 header 名称转换(如 x-token) |
| Cookie 函数 | param = Cookie(...) | 从 Cookie 中提取值 | from fastapi import Cookie / def get_session(session_id: str = Cookie(None)): ... | session_id 可能为空 |
| 多个 Header | headers: dict = Header() | 获取所有请求头 | def read_headers(headers: dict = Header(...)): ... | 返回字典形式的所有头信息 |
| 自定义 Header 名 | Header(..., convert_underscores=False) | 保留下划线(如 X_API_KEY) | x_api_key: str = Header(..., convert_underscores=False) | 默认会将 _ 转为 - |
3.6 表单数据处理(Form)
| 方法名称 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| Form 函数 | param = Form(...) | 从表单中提取字段 | from fastapi import Form / @app.post("/login") / def login(username: str = Form(...), password: str = Form(...)): ... | 需安装 python-multipart |
| 必填表单字段 | Form(...) | 声明必填表单字段 | username: str = Form(...) | 缺少时返回 422 |
| 可选表单字段 | Form(default) | 设置默认值 | note: str = Form("") | 可省略 |
| 文件与表单混合 | file: UploadFile, field: str = Form(...) | 同时处理文件和表单字段 | @app.post("/upload") / def upload(file: UploadFile, desc: str = Form("")): ... | Content-Type 为 multipart/form-data |
3.7 文件上传(File 与 UploadFile)
| 方法名称 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| UploadFile 类 | param: UploadFile | 接收上传的文件对象 | from fastapi import UploadFile / def upload(file: UploadFile): / return {"filename": file.filename} | 支持异步读取 |
| 读取文件内容 | await file.read() | 异步读取文件二进制数据 | content = await file.read() | 大文件需分块读取 |
| 文件元信息 | file.filename, file.content_type | 获取文件名和 MIME 类型 | name = file.filename / mime = file.content_type | 可用于验证或存储 |
| 多文件上传 | files: List[UploadFile] = File(...) | 接收多个文件 | from typing import List / from fastapi import File / def upload_files(files: List[UploadFile] = File(...)): ... | HTML 中 input 设置 multiple |
| File 函数 | File(..., description="...") | 增强文件参数功能 | image: UploadFile = File(..., description="上传头像") | 可设置验证、描述等 |
第四章:响应处理与模型定义
4.1 返回 JSON 响应(return dict 或 return BaseModel)
| 方法名称 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| 返回字典 | return {"key": "value"} | 自动序列化为 JSON | @app.get("/") / def home(): / return {"status": "ok"} | 支持嵌套字典和列表 |
| 返回 BaseModel | return Model(**data) | 自动转为 JSON 并校验 | class Resp(BaseModel): / msg: str / @app.get("/", response_model=Resp) / def home(): / return Resp(msg="ok") | 推荐用于响应结构化 |
| 返回列表 | return [item1, item2] | 返回 JSON 数组 | return [{"id": 1}, {"id": 2}] | 适用于集合资源 |
4.2 使用 Response 类自定义响应
| 方法名称 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| Response 基类 | return Response(...) | 自定义响应内容与类型 | from starlette.responses import Response / return Response(content="data", media_type="text/csv") | 最基础的响应类 |
| JSONResponse | return JSONResponse(...) | 显式返回 JSON 响应 | from fastapi.responses import JSONResponse / return JSONResponse(content={"error": "not found"}, status_code=404) | 可自定义状态码和头 |
| PlainTextResponse | return PlainTextResponse(...) | 返回纯文本 | from fastapi.responses import PlainTextResponse / return PlainTextResponse("Hello", status_code=200) | media_type=text/plain |
| HTMLResponse | return HTMLResponse(...) | 返回 HTML 内容 | from fastapi.responses import HTMLResponse / return HTMLResponse("<h1>OK</h1>") | media_type=text/html |
4.3 设置响应状态码(status_code 参数)
| 方法名称 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| status_code 参数 | @app.post(..., status_code=201) | 自定义成功状态码 | @app.post("/items", status_code=201) / def create_item(item: Item): ... | POST 创建常用 201 |
| 在函数中设置 | return JSONResponse(..., status_code=202) | 动态设置状态码 | return JSONResponse(content={"msg": "accepted"}, status_code=202) | 覆盖装饰器设置 |
| 重定向状态码 | status_code=302 | 实现重定向 | return RedirectResponse(url="/new", status_code=302) | 配合 Location 头使用 |
4.4 响应模型(response_model)与字段过滤
| 方法名称 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| response_model | @app.post(..., response_model=Model) | 指定响应数据结构 | @app.post("/users", response_model=UserOut) / def create_user(user: UserIn): ... | 自动过滤私密字段 |
| exclude_unset | response_model_exclude_unset=True | 排除未设置的字段 | return JSONResponse(content=user.dict(exclude_unset=True), status_code=200) | 适用于 PATCH 响应 |
| exclude_defaults | response_model_exclude_defaults=True | 排除默认值字段 | user.dict(exclude_defaults=True) | 减少响应体积 |
| exclude_none | response_model_exclude_none=True | 排除值为 None 的字段 | user.dict(exclude_none=True) | 避免返回空值 |
4.5 返回纯文本、HTML、文件等响应类型
| 方法名称 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| PlainTextResponse | response_class=PlainTextResponse | 返回纯文本 | @app.get("/", response_class=PlainTextResponse) / def home(): / return "Hello World" | 简化文本响应 |
| HTMLResponse | response_class=HTMLResponse | 返回 HTML 页面 | @app.get("/", response_class=HTMLResponse) / def home(): / return "<h1>Hi</h1>" | 可结合模板引擎 |
| FileResponse | return FileResponse("path.txt") | 返回静态文件 | from fastapi.responses import FileResponse / return FileResponse("data.csv") | 自动设置 Content-Type |
| StreamingResponse | return StreamingResponse(...) | 流式传输大文件 | 适用于视频、大 CSV 等 | 避免内存溢出 |
4.6 处理响应头与 Cookie(Response 对象方法)
| 方法名称 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| set_header() | response.headers["X-Auth"] = "val" | 设置响应头 | from fastapi import Response / def home(response: Response): / response.headers["X-Version"] = "1.0" | response 是函数参数 |
| set_cookie() | response.set_cookie(...) | 设置 Cookie | response.set_cookie(key="session_id", value="123") | 可设置过期时间、安全标志 |
| delete_cookie() | response.delete_cookie(...) | 删除 Cookie | response.delete_cookie("session_id") | 实际是设置过期 |
| CORS 头 | from fastapi.middleware.cors import CORSMiddleware | 处理跨域 | app.add_middleware(CORSMiddleware, allow_origins=["*"]) | 需中间件支持 |
第五章:数据验证与错误处理
5.1 Pydantic 模型字段验证(Field 参数)
| 方法名称 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| Field 函数 | Field(...) | 添加字段级验证和元数据 | from pydantic import Field / name: str = Field(..., min_length=2, max_length=50) | ... 表示必填 |
| 字符串验证 | min_length, max_length, regex | 约束字符串格式 | email: str = Field(..., regex=".+@.+..com") | 可组合使用 |
| 数值验证 | gt, ge, lt, le | 约束数值范围 | age: int = Field(..., ge=0, le=150) | ge=≥, gt=>, le=≤, lt=< |
| 默认工厂 | default_factory=list | 可变默认值 | tags: List[str] = Field(default_factory=list) | 避免可变对象共享 |
5.2 查询/路径参数的约束(最小值、最大值、正则等)
| 方法名称 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| Query 验证 | Query(..., min_length=3) | 验证查询参数 | q: str = Query(..., min_length=3) | 错误返回 422 |
| Path 验证 | Path(..., ge=1) | 验证路径参数 | item_id: int = Path(..., ge=1) | 路径参数也需验证 |
| 正则表达式 | Query(..., regex="^abc") | 模式匹配 | code: str = Query(..., regex="^[A-Z]{3}$") | 使用标准 Python 正则 |
| 列表长度 | Query(..., max_items=5) | 限制列表项数 | tags: List[str] = Query(..., max_items=5) | 适用于多值查询参数 |
5.3 自定义验证函数(@validator、@root_validator)
| 方法名称 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| @validator | @validator('field') | 字段级自定义验证 | from pydantic import validator / @validator('price') / def price_must_be_positive(cls, v): / if v <= 0: / raise ValueError('must be > 0') / return v | cls 是类,v 是值 |
| @root_validator | @root_validator(pre=True) | 跨字段验证 | @root_validator / def check_passwords_match(cls, values): / pw = values.get('password') / confirm = values.get('confirm') / if pw != confirm: / raise ValueError('passwords do not match') / return values | pre=True 在字段验证前执行 |
| 验证顺序 | pre=True / pre=False | 控制验证执行时机 | @root_validator(pre=False) | pre=False 是默认,后执行 |
5.4 使用 HTTPException 抛出错误
| 方法名称 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| HTTPException | raise HTTPException(status_code, detail) | 抛出标准 HTTP 错误 | from fastapi import HTTPException / raise HTTPException(status_code=404, detail="Item not found") | 自动返回 JSON 错误响应 |
| 自定义状态码 | 400, 401, 403, 404, 500 | 根据错误类型选择 | raise HTTPException(403, "Forbidden") | 遵循 HTTP 语义 |
| 错误详情 | detail="描述信息" | 提供错误原因 | detail="User not authorized" | 可为字符串或字典 |
5.5 全局异常处理器(@app.exception_handler)
| 方法名称 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| @app.exception_handler() | @app.exception_handler(ExceptionType) | 捕获特定异常 | @app.exception_handler(RequestValidationError) / def handle_validation_error(request, exc): / return JSONResponse(..., status_code=422) | 优先级高于默认处理器 |
| 处理自定义异常 | class CustomError(Exception): ... | 定义业务异常 | class InactiveUserError(Exception): pass / @app.exception_handler(InactiveUserError) / def handle_inactive(request, exc): ... | 统一处理业务错误 |
| 通用异常处理 | @app.exception_handler(Exception) | 捕获所有未处理异常 | 用于记录日志并返回 500 | 生产环境避免暴露敏感信息 |
5.6 验证错误的默认响应格式
| 特性 | 格式/说明 | 用途 | 示例 | 注意事项 |
|---|---|---|---|---|
| 默认错误状态码 | 422 Unprocessable Entity | 请求数据验证失败 | status_code=422 | 用于 Pydantic 验证错误 |
| 错误响应结构 | JSON 包含 loc, msg, type | 提供详细错误信息 | {"detail": [{"loc": ["body", "age"], "msg": "ensure this value is >= 0", "type": "value_error.number.not_ge"}]} | loc 表示错误位置 |
| 错误位置 loc | 路径如 ["body", "user", "age"] | 指明错误字段层级 | 便于前端定位问题字段 | 支持 body, query, path, header |
| 国际化支持 | 可自定义错误消息 | 适配多语言 | 在 Field 或 validator 中设置 | 需结合 i18n 库实现 |
第六章:依赖注入系统
6.1 依赖注入概念与优势
| 概念 | 说明 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| 依赖注入(DI) | 将功能逻辑解耦,由框架自动注入所需组件 | 提高代码复用性、可测试性和可维护性 | def get_db(): / return db_session / @app.get("/items", dependencies=[Depends(get_db)]) | 避免硬编码和全局变量 |
| 声明式编程 | 通过声明依赖而非手动调用 | 简化函数逻辑,专注业务处理 | 使用 Depends() 显式声明依赖 | 框架负责解析和执行 |
| 自动解析 | FastAPI 自动解析依赖层级 | 支持嵌套依赖和参数传递 | 支持函数、类、子依赖 | 无需手动管理调用顺序 |
6.2 定义和使用函数依赖
| 方法名称 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| 函数依赖 | def dependency(): ... | 定义可复用的验证或数据获取逻辑 | def require_auth(token: str = Header(...)): / if not valid(token): / raise HTTPException(401) / return token | 返回值可用于主函数参数 |
| 参数注入 | Depends(dependency_func) | 在路由中使用函数依赖 | @app.get("/secure", dependencies=[Depends(require_auth)]) / def secure_data(): ... | 可用于全局或单个路由 |
| 返回值传递 | return value | 将结果传递给主函数 | def get_user(token): / return db.get_user(token) / def route(user = Depends(get_user)): ... | 主函数可通过参数接收 |
6.3 类依赖与 call 方法
| 方法名称 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| 类依赖 | class Dependency: __call__ = ... | 实现状态化或配置化依赖 | class RateLimiter: / def __init__(self, limit): / self.limit = limit / def __call__(self, ip: str = Header(...)): / if too_many_requests(ip, self.limit): / raise HTTPException(429) | 类需实现 __call__ |
| 初始化参数 | limiter = RateLimiter(100) | 创建带配置的依赖实例 | ten_per_min = RateLimiter(10) / @app.get("/data", dependencies=[Depends(ten_per_min)]) | 支持复用不同配置 |
| 状态管理 | self.counter, self.config | 维护内部状态或配置 | 可用于缓存、计数器等场景 | 注意线程/异步安全 |
6.4 共享依赖(全局、路由、参数级)
| 方法名称 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| 全局依赖 | FastAPI(dependencies=[...]) | 为所有路由添加统一依赖 | app = FastAPI(dependencies=[Depends(check_api_key)]) | 适用于认证、日志等通用逻辑 |
| 路由级依赖 | @app.get(..., dependencies=[...]) | 为特定路由组添加依赖 | @app.get("/admin", dependencies=[Depends(is_admin)]) | 比全局依赖优先级高 |
| 参数级依赖 | param = Depends(...) | 在函数参数中直接使用 | def route(db = Depends(get_db)): ... | 最灵活,可组合多个依赖 |
6.5 子依赖与依赖嵌套
| 方法名称 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| 子依赖 | def dep_a(dep_b = Depends(...)) | 依赖之间相互调用 | def get_token(): ... / def get_user(token = Depends(get_token)): / return user / def route(user = Depends(get_user)): ... | DI 容器自动解析层级 |
| 嵌套调用 | A → B → C | 构建依赖链 | 支持任意深度嵌套 | 避免循环依赖 |
| 参数复用 | 同一依赖被多个函数使用 | 减少重复代码 | get_db 可被所有数据操作函数使用 | 推荐将通用逻辑封装为依赖 |
6.6 使用 Depends() 传递参数与复用逻辑
| 方法名称 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| 参数化依赖 | Depends(lambda: func(arg)) | 向依赖传递参数 | @app.get("/items", dependencies=[Depends(lambda: check_role("editor"))]) | 使用 lambda 包装 |
| 部分应用 | from functools import partial | 固定部分参数 | from functools import partial / editor_only = partial(check_role, role="editor") / Depends(editor_only) | 更清晰的参数传递方式 |
| 动态配置 | Depends(configured_func) | 根据环境或路径动态选择逻辑 | 根据 URL 路径切换数据库实例 | 提高灵活性和可配置性 |
第七章:安全与认证
7.1 OAuth2 + Password(含 Bearer)流程
| 概念 | 说明 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| OAuth2PasswordBearer | 用于密码模式的 Bearer token 认证 | 获取并验证 token | from fastapi.security import OAuth2PasswordBearer / oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token") | tokenUrl 是获取 token 的端点 |
| Bearer Token | Authorization: Bearer <token> | 客户端在请求头中携带 token | 请求头示例:Authorization: Bearer eyJhbGciOi... | 传输需 HTTPS |
| 密码模式 | 用户名+密码换取 token | 实现登录认证 | 通常用于内部系统或可信客户端 | 不推荐用于第三方应用 |
7.2 使用 OAuth2PasswordRequestForm 处理登录
| 方法名称 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| OAuth2PasswordRequestForm | form_data: OAuth2PasswordRequestForm = Depends() | 接收用户名密码表单 | @app.post("/token") / def login(form_data: OAuth2PasswordRequestForm = Depends()): / user = authenticate(form_data.username, form_data.password) | 自动解析 x-www-form-urlencoded |
| 表单字段 | form_data.username, form_data.password | 获取认证凭据 | 需配合密码哈希验证 | 密码不应明文存储 |
| 安全传输 | HTTPS | 防止凭据泄露 | 生产环境必须使用 HTTPS | 否则存在安全风险 |
7.3 JWT(JSON Web Token)生成与验证
| 方法名称 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| jwt.encode() | jwt.encode(payload, key, algorithm) | 生成 JWT token | import jwt / token = jwt.encode({"sub": "user123"}, "SECRET", algorithm="HS256") | 使用强密钥 |
| jwt.decode() | jwt.decode(token, key, algorithms) | 验证并解析 token | payload = jwt.decode(token, "SECRET", algorithms=["HS256"]) | 捕获 ExpiredSignatureError 等异常 |
| 载荷(Payload) | {"sub": "...", "exp": ...} | 存储用户信息和过期时间 | exp 字段表示过期时间戳 | 避免存储敏感信息 |
7.4 使用 Security 依赖保护路由
| 方法名称 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| Security() | Security(dependency, scopes=...) | 声明安全依赖和权限范围 | @app.get("/me") / def read_me(current_user = Security(get_current_user, scopes=["me"])): ... | 用于 OAuth2 作用域控制 |
| 权限校验 | raise HTTPException(403) | 在依赖中拒绝未授权访问 | if not has_scope(user, scope): / raise HTTPException(403) | 及时中断请求 |
| 作用域(Scopes) | scopes=["read", "write"] | 细粒度权限控制 | 适用于复杂权限系统 | 需在 token 中包含 scope 信息 |
7.5 API Key 认证(Header、Query、Cookie)
| 方法名称 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| APIKeyHeader | APIKeyHeader(name="X-API-Key") | 从请求头读取 API Key | from fastapi.security import APIKeyHeader / api_key_header = APIKeyHeader(name="X-API-Key") | 常见于第三方 API |
| APIKeyQuery | APIKeyQuery(name="api_key") | 从查询参数读取 API Key | api_key_q = APIKeyQuery(name="api_key") | URL 中暴露 key,安全性较低 |
| APIKeyCookie | APIKeyCookie(name="session") | 从 Cookie 读取 API Key | api_key_cookie = APIKeyCookie(name="session") | 需防范 CSRF 攻击 |
| 验证逻辑 | if key != VALID_KEY: raise ... | 校验 key 是否有效 | 可结合数据库或缓存验证 | 建议使用哈希存储 key |
7.6 权限控制与角色管理基础
| 方法名称 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| 角色检查 | if user.role != "admin": ... | 控制不同角色的访问权限 | def is_admin(user: User = Depends(get_current_user)): / if user.role != "admin": / raise HTTPException(403) / return user | 可作为依赖复用 |
| 装饰器模式 | @require_role("admin") | 简化权限检查 | 自定义装饰器封装角色逻辑 | 提高代码可读性 |
| RBAC 模型 | Role → Permissions | 实现基于角色的访问控制 | 数据库设计包含用户、角色、权限表 | 适用于复杂系统 |
第八章:数据库集成(SQLAlchemy / Tortoise ORM)
8.1 同步数据库(SQLAlchemy + Session)
| 方法名称 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| SessionLocal | session = SessionLocal() | 创建数据库会话 | from sqlalchemy.orm import sessionmaker / SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) | 每个请求应创建独立会话 |
| 依赖注入会话 | def get_db(): yield session | 将数据库会话作为依赖 | def get_db(): / db = SessionLocal() / try: / yield db / finally: / db.close() | 使用 yield 确保关闭 |
| CRUD 操作 | db.query(Model).filter(...).first() | 执行数据库操作 | user = db.query(User).filter(User.email==email).first() | 注意查询性能,避免 N+1 问题 |
8.2 异步数据库(Tortoise ORM / SQLAlchemy 2.0 async)
| 方法名称 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| async with conn | async with database.connect(): | 异步连接数据库 | async with get_db() as db: / users = await User.all() | 需使用异步 ORM |
| await 查询 | result = await query | 非阻塞执行数据库操作 | users = await db.fetch_all(users.select()) | 保持异步上下文 |
| 异步会话 | async_session = AsyncSession() | SQLAlchemy 2.0 异步支持 | from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession / engine = create_async_engine(DATABASE_URL) | 需安装 async driver(如 asyncpg) |
8.3 使用 @app.on_event(“startup”) 初始化数据库连接
| 方法名称 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| startup 事件 | @app.on_event("startup") | 应用启动时执行初始化 | @app.on_event("startup") / async def init_db(): / await database.connect() | 适用于异步初始化 |
| shutdown 事件 | @app.on_event("shutdown") | 应用关闭时清理资源 | @app.on_event("shutdown") / async def disconnect_db(): / await database.disconnect() | 释放连接、关闭文件等 |
| 同步初始化 | 同步函数直接调用 | 同步数据库连接 | create_tables() | 可在模块级执行 |
8.4 CRUD 操作封装与依赖注入
| 方法名称 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| CRUD 类 | class CRUDBase: def create(...): | 封装通用数据操作 | class CRUDUser(CRUDBase): / def get_by_email(self, db, email): ... | 提高代码复用性 |
| 依赖注入服务 | def route(service = Depends(UserService)) | 注入业务服务类 | 解耦路由与数据访问逻辑 | 便于测试和替换实现 |
| 分页查询 | skip, limit 参数 | 实现数据分页 | def get_items(db, skip=0, limit=10): / return db.query(Item).offset(skip).limit(limit).all() | 避免一次性加载大量数据 |
8.5 分页与查询优化
| 方法名称 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| OFFSET/LIMIT | .offset(skip).limit(limit) | 实现分页 | items = db.query(Item).offset(10).limit(20).all() | 大偏移量时性能下降 |
| 游标分页 | WHERE id > last_id LIMIT n | 基于游标的高效分页 | items = db.query(Item).filter(Item.id > last_id).limit(20).all() | 适用于不可变数据集 |
| 索引优化 | CREATE INDEX ... | 提高查询速度 | 为常用查询字段(如 email, status)创建索引 | 避免全表扫描 |
| 懒加载 vs 预加载 | selectinload, joinedload | 控制关联对象加载策略 | from sqlalchemy.orm import selectinload / stmt = select(User).options(selectinload(User.items)) | 防止 N+1 查询问题 |
第九章:异步编程与性能优化
9.1 async def 与 await 基础
| 方法名称 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| async def | async def func(): ... | 定义异步函数 | async def fetch_data(): / await asyncio.sleep(1) / return "data" | 只能在 async 函数中使用 await |
| await 表达式 | await coro() | 等待协程完成 | result = await fetch_data() | 释放控制权,允许其他任务运行 |
| 协程对象 | func() 返回协程 | 异步调用不立即执行 | task = fetch_data() # coroutine | 需 await 或 asyncio.create_task() |
9.2 异步视图函数与非阻塞 I/O
| 方法名称 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| 异步路由 | @app.get("/") / async def handler(): ... | 处理异步请求 | @app.get("/data") / async def get_data(): / return await db.fetch_data() | 适用于数据库、HTTP 请求等耗时操作 |
| 非阻塞调用 | await asyncio.sleep(), await db.query() | 执行非阻塞操作 | 避免使用 time.sleep(), requests.get() 等阻塞调用 | 使用 aiohttp、asyncpg 等异步库 |
9.3 异步数据库操作示例
| 方法名称 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| 异步查询 | result = await conn.fetch(...) | 获取数据 | users = await database.fetch_all(users.select()) | 需在 async 函数中 |
| 异步插入 | await conn.execute(...) | 写入数据 | await database.execute(users.insert().values(name="John")) | 注意事务管理 |
| 事务处理 | async with conn.transaction(): | 确保操作原子性 | async with database.transaction(): / await db.execute(...) | 防止数据不一致 |
9.4 并发请求处理与性能测试
| 方法名称 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| asyncio.gather() | await asyncio.gather(coro1, coro2) | 并发执行多个协程 | results = await asyncio.gather(fetch_user(1), fetch_user(2)) | 比顺序执行更快 |
| 性能测试工具 | pytest-benchmark, wrk | 测量 API 性能 | wrk -t10 -c100 -d30s http://localhost:8000/ | 评估并发处理能力 |
| 压力测试 | 模拟高并发请求 | 发现性能瓶颈 | 使用 Locust 或 k6 | 关注 CPU、内存、数据库连接数 |
9.5 使用 BackgroundTasks 执行后台任务
| 方法名称 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| BackgroundTasks | background_tasks: BackgroundTasks | 注入后台任务队列 | from fastapi import BackgroundTasks / def send_email(to: str): ... / @app.post("/notify", status_code=202) / def notify(bg: BackgroundTasks): / bg.add_task(send_email, "user@example.com") | 立即返回响应,任务后台执行 |
| add_task() | bg.add_task(func, *args, **kwargs) | 添加任务到队列 | bg.add_task(generate_report, user_id=123) | 任务函数不应有返回值依赖 |
| 适用场景 | 发送邮件、生成报告、清理缓存 | 耗时但无需即时响应的操作 | 避免在后台任务中处理关键业务逻辑 | 任务可能在应用关闭时中断 |
第十章:中间件与事件处理器
10.1 添加中间件(CORS、日志、GZip)
| 方法名称 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| CORSMiddleware | app.add_middleware(CORSMiddleware, ...) | 处理跨域请求 | from fastapi.middleware.cors import CORSMiddleware / app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"]) | 前端开发常用 |
| GZipMiddleware | app.add_middleware(GZipMiddleware) | 启用响应压缩 | from starlette.middleware.gzip import GZipMiddleware / app.add_middleware(GZipMiddleware, minimum_size=1000) | 减少传输体积,提高性能 |
| 日志中间件 | 自定义中间件记录请求信息 | 监控和调试 | 记录请求路径、方法、耗时、状态码 | 注意不要记录敏感数据 |
10.2 自定义中间件(@app.middleware(“http”))
| 方法名称 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| @app.middleware(“http”) | @app.middleware("http") / async def middleware(request, call_next): ... | 定义自定义中间件 | @app.middleware("http") / async def add_process_time_header(request: Request, call_next): / start_time = time.time() / response = await call_next(request) / response.headers["X-Process-Time"] = str(time.time() - start_time) / return response | 在请求前后执行逻辑 |
| call_next | response = await call_next(request) | 调用下一个中间件或路由函数 | 必须调用以继续处理流程 | 可修改请求或响应对象 |
10.3 启动与关闭事件(@app.on_event(“startup”) / “shutdown”)
| 方法名称 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| @app.on_event(“startup”) | @app.on_event("startup") / async def func(): ... | 应用启动时初始化资源 | 连接数据库、加载模型、创建索引 | 确保资源准备就绪 |
| @app.on_event(“shutdown”) | @app.on_event("shutdown") / async def func(): ... | 应用关闭时清理资源 | 断开数据库连接、关闭文件句柄、保存状态 | 避免资源泄漏 |
| 同步事件 | 同步函数 | 执行同步初始化 | create_tables() | 可直接定义函数 |
10.4 请求生命周期钩子
| 概念 | 说明 | 用途 | 示例 | 注意事项 |
|---|---|---|---|---|
| 请求开始 | 接收到请求 | 记录开始时间、验证身份 | 中间件中记录 start_time | 可用于性能监控 |
| 路由匹配 | 查找对应处理函数 | 确定执行哪个路径操作函数 | 框架自动处理 | 注意路径顺序 |
| 依赖注入 | 解析并执行依赖 | 获取数据库连接、用户身份等 | Depends(get_current_user) | 可能抛出异常 |
| 请求处理 | 执行主函数逻辑 | 业务逻辑处理 | CRUD 操作、计算等 | 应尽量异步 |
| 响应生成 | 序列化返回值 | 生成 JSON 或其他格式响应 | return {"data": ...} | 可通过 response_model 过滤 |
| 响应发送 | 将响应返回客户端 | 完成 HTTP 交互 | 框架自动处理 | 中间件可修改响应头 |
第十一章:API 文档与交互式界面
11.1 自动生成的 Swagger UI(/docs)
| 方法名称 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| Swagger UI | 访问 /docs | 交互式 API 文档界面 | 启动应用后访问 http://localhost:8000/docs | 基于 Swagger UI 构建 |
| 实时测试 | 在浏览器中直接调用 API | 调试和演示 API | 点击 “Try it out” 按钮 | 需处理认证(如 Bearer Token) |
| 自动更新 | 修改代码后刷新页面 | 文档与代码同步 | 无需手动维护文档 | 极大提升开发效率 |
11.2 ReDoc 文档界面(/redoc)
| 方法名称 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| ReDoc | 访问 /redoc | 另一种 API 文档展示形式 | http://localhost:8000/redoc | 基于 ReDoc 构建 |
| 阅读友好 | 更适合阅读和分享 | 技术文档、对外 API 说明 | 页面布局更清晰 | 不支持直接测试请求 |
| OpenAPI 规范 | /openapi.json | 生成标准 OpenAPI(原 Swagger)JSON | 可用于导入 Postman、生成 SDK | 标准化接口描述 |
11.3 自定义 API 元数据(标题、描述、版本)
| 方法名称 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| FastAPI 构造函数参数 | FastAPI(title=..., description=..., version=...) | 设置 API 基本信息 | app = FastAPI(title="My API", description="A sample API", version="1.0.0") | 出现在文档首页 |
| 文档 URL 自定义 | docs_url, redoc_url | 修改或禁用文档路径 | app = FastAPI(docs_url="/api-docs") / # 或 app = FastAPI(docs_url=None) 禁用 | 生产环境可禁用以提高安全 |
| 联系信息 | contact=... | 提供开发者联系方式 | contact={"name": "API Support", "email": "support@example.com"} | 便于使用者联系 |
11.4 给路由添加文档字符串(docstring)
| 方法名称 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| 函数文档字符串 | """...""" | 为路由添加详细说明 | @app.get("/items") / def read_items(): / """Retrieve a list of items. Optional query parameters: - skip: items to skip - limit: max items to return""" / return [] | 支持 Markdown 格式 |
| 参数描述 | :param param: ... | 详细描述参数 | 在 docstring 中使用标准格式 | 提高文档可读性 |
| 响应描述 | :return: ... | 描述返回值 | 文档自动生成响应模型 | 可结合 response_model 使用 |
11.5 隐藏或自定义文档路径
| 方法名称 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| 禁用文档 | docs_url=None, redoc_url=None | 关闭自动生成的文档 | app = FastAPI(docs_url=None, redoc_url=None) | 生产环境增强安全性 |
| 自定义路径 | docs_url="/apidocs" | 修改文档访问路径 | app = FastAPI(docs_url="/apidocs") | 避免被扫描发现 |
| 条件启用 | 根据环境变量决定 | 开发环境开启,生产关闭 | if settings.DEBUG: / app = FastAPI() / else: / app = FastAPI(docs_url=None) | 灵活控制 |
第十二章:测试与调试
12.1 使用 TestClient 进行单元测试
| 方法名称 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| TestClient | from fastapi.testclient import TestClient | 创建测试客户端 | client = TestClient(app) / response = client.get("/") | 同步接口,便于测试 |
| 模拟请求 | client.get(), client.post() | 发送各种 HTTP 请求 | response = client.post("/items", json={"name": "foo"}) | 可传递 json, data, headers 等参数 |
| 测试覆盖率 | pytest-cov | 测量代码测试覆盖率 | pytest --cov=app tests/ | 确保关键路径被覆盖 |
12.2 模拟请求与断言响应
| 方法名称 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| assert response.status_code == 200 | 验证状态码 | 确保请求成功 | assert response.status_code == 200 | 常见断言 |
| assert response.json() == {…} | 验证响应内容 | 检查返回数据正确性 | assert response.json() == {"msg": "ok"} | 自动解析 JSON |
| assert “required” in response.text | 验证文本内容 | 检查 HTML 或文本响应 | 适用于非 JSON 响应 | 使用 response.text |
12.3 测试依赖覆盖(app.dependency_overrides)
| 方法名称 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| dependency_overrides | app.dependency_overrides[dep] = mock | 替换依赖为模拟实现 | def override_get_db(): / return TestingSessionLocal() / app.dependency_overrides[get_db] = override_get_db | 隔离测试,避免真实数据库 |
| 清理覆盖 | app.dependency_overrides.clear() | 测试后恢复原始依赖 | 在 pytest fixture teardown 中调用 | 防止影响其他测试 |
| 模拟外部服务 | 模拟邮件发送、第三方 API | 测试时不发送真实请求 | def mock_send_email(to, body): pass / app.dependency_overrides[send_email] = mock_send_email | 提高测试速度和稳定性 |
12.4 异步测试支持(pytest-asyncio)
| 方法名称 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| @pytest.mark.asyncio | @pytest.mark.asyncio | 标记异步测试函数 | @pytest.mark.asyncio / async def test_async_route(): / response = client.get("/async") / assert response.status_code == 200 | 需安装 pytest-asyncio |
| AsyncClient | from starlette.testclient import TestClient (异步版) | 异步测试客户端 | 推荐使用 httpx.AsyncClient | 更适合测试异步逻辑 |
| await client.get() | await 发送请求 | 在异步测试中使用 | response = await async_client.get("/items") | 保持异步上下文 |
12.5 调试技巧与日志配置
| 方法名称 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| print() 调试 | print("debug info") | 快速输出变量值 | print(f"User: {user}") | 简单但有效,生产移除 |
| logging 模块 | import logging | 标准日志记录 | logging.basicConfig(level=logging.INFO) / logger = logging.getLogger(__name__) / logger.info("Request processed") | 可配置级别、格式、输出位置 |
| IDE 调试器 | 断点调试 | 逐步执行代码 | 在 PyCharm、VSCode 中设置断点 | 最强大的调试方式 |
| 环境区分 | DEBUG=True | 开发环境启用详细日志 | if settings.DEBUG: / logging.basicConfig(level=logging.DEBUG) | 生产环境使用 WARNING 或 ERROR |
第十三章:部署与生产环境配置
13.1 使用 Uvicorn 生产模式启动
| 方法名称 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| Uvicorn 命令行 | uvicorn main:app --host 0.0.0.0 --port 80 --workers 4 | 生产环境启动 | 使用多个 worker 提升性能 | 避免使用 —reload |
| workers 参数 | --workers N | 启动多个工作进程 | --workers 4 # 通常为 CPU 核心数 | 提高并发处理能力 |
| worker_class | --worker-class uvicorn.workers.UvicornWorker | 指定工作类 | 与 Gunicorn 配合使用 | 优化性能 |
13.2 部署到 Gunicorn + Uvicorn 工作进程
| 方法名称 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| Gunicorn 命令 | gunicorn -k uvicorn.workers.UvicornWorker -w 4 main:app | 使用 Gunicorn 管理 Uvicorn 进程 | 结合 Gunicorn 的进程管理与 Uvicorn 的异步能力 | 生产推荐方案 |
| 配置文件 | gunicorn.conf.py | 集中管理部署配置 | bind = "0.0.0.0:8000" / workers = 4 / worker_class = "uvicorn.workers.UvicornWorker" | 便于版本控制和复用 |
| 进程监控 | Gunicorn 自带监控 | 重启崩溃进程、负载均衡 | 自动管理 worker 生命周期 | 提高应用稳定性 |
13.3 使用 Docker 容器化部署
| 方法名称 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| Dockerfile | FROM, COPY, CMD | 定义应用镜像 | FROM python:3.9 / COPY . /app / WORKDIR /app / RUN pip install -r requirements.txt / CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "80"] | 基于官方 Python 镜像 |
| docker-compose.yml | services: app: ... | 定义多容器应用 | 包含 app、db、redis 等服务 | 便于本地开发和部署 |
| 镜像构建 | docker build -t myapp . | 构建 Docker 镜像 | tag 用于版本管理 | 推送到镜像仓库 |
13.4 环境变量管理(pydantic.BaseSettings)
| 方法名称 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| BaseSettings | class Settings(BaseSettings): ... | 管理应用配置 | class Settings(BaseSettings): / database_url: str / debug: bool = False / secret_key: str / settings = Settings() | 自动从环境变量读取 |
| .env 文件 | DATABASE_URL=sqlite:///db.sqlite3 | 本地开发配置文件 | 创建 .env 文件存放配置 | 不应提交到版本控制 |
| 字段别名 | field_name: str = Field(..., env="ENV_NAME") | 映射不同环境变量名 | api_key: str = Field(..., env="SECRET_API_KEY") | 灵活适配 |
13.5 日志配置与监控(Prometheus、Sentry)
| 方法名称 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| Structured Logging | JSON 格式日志 | 便于日志收集和分析 | {"time": "...", "level": "INFO", "message": "...", "path": "/items"} | 使用 loguru 或 structlog |
| Prometheus | /metrics 端点 | 收集应用指标 | 请求计数、响应时间、错误率 | 需集成 starlette_exporter |
| Sentry | sentry_sdk.init() | 错误追踪与告警 | import sentry_sdk / sentry_sdk.init(dsn="...") | 捕获未处理异常,便于排查 |
| 健康检查 | @app.get("/healthz") | 监控应用健康状态 | return {"status": "ok"} | 用于 Kubernetes 等编排系统 |
第十四章:高级特性与扩展
14.1 WebSockets 支持(@app.websocket)
| 方法名称 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| @app.websocket() | @app.websocket("/ws") | 定义 WebSocket 路由 | @app.websocket("/ws") / async def websocket_endpoint(websocket: WebSocket): / await websocket.accept() / while True: / data = await websocket.receive_text() / await websocket.send_text(f"Echo: {data}") | 实现双向实时通信 |
| websocket.accept() | await websocket.accept() | 接受客户端连接 | 必须在接收消息前调用 | 可指定子协议 |
| receive_text() / send_text() | await websocket.receive_text() | 收发文本消息 | await websocket.send_text("Hello") | 异步非阻塞 |
| receive_json() / send_json() | await websocket.receive_json() | 收发 JSON 消息 | data = await websocket.receive_json() | 自动序列化/反序列化 |
14.2 使用 APIRouter 模块化路由
| 方法名称 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| APIRouter() | router = APIRouter() | 创建子路由组 | from fastapi import APIRouter / router = APIRouter(prefix="/users", tags=["users"]) | 用于功能模块划分 |
| include_router() | app.include_router(router) | 将子路由挂载到主应用 | app.include_router(user_router) / app.include_router(item_router, prefix="/items") | 可设置统一前缀和标签 |
| 路由分组 | 按功能拆分文件 | 组织大型项目 | users.py, items.py, auth.py | 提高代码可维护性 |
14.3 子应用程序挂载(app.mount())
| 方法名称 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| app.mount() | app.mount("/static", StaticFiles(...)) | 挂载子应用 | from fastapi.staticfiles import StaticFiles / app.mount("/static", StaticFiles(directory="static"), name="static") | 用于静态文件服务 |
| WSGI 应用 | mount(FlaskApp()) | 集成 Flask 等 WSGI 应用 | 可逐步迁移旧系统 | 注意路径冲突 |
| 路径前缀 | mount("/sub", app) | 为子应用设置前缀 | 所有子应用路由基于该前缀 | 必须唯一 |
14.4 自定义响应类(JSONResponse、PlainTextResponse 等)
| 方法名称 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| JSONResponse | return JSONResponse(content={}) | 自定义 JSON 响应 | 可设置 headers、status_code | 最常用 |
| PlainTextResponse | return PlainTextResponse("text") | 返回纯文本 | 用于简单文本或脚本 | media_type=text/plain |
| HTMLResponse | return HTMLResponse("<html>...") | 返回 HTML 页面 | 可结合 Jinja2 模板引擎 | media_type=text/html |
| RedirectResponse | return RedirectResponse(url) | 重定向 | status_code=302 或 307 | 用于跳转 |
| StreamingResponse | return StreamingResponse(generator) | 流式传输大内容 | 适用于视频、大文件下载 | 节省内存 |
14.5 与 GraphQL 集成(Strawberry 或 Ariadne)
| 方法名称 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| Strawberry | @strawberry.type | 定义 GraphQL 类型 | import strawberry / @strawberry.type / class User: / name: str / age: int | Python 原生类型风格 |
| Ariadne | make_executable_schema | 基于 Schema First | 定义 .graphql 文件和解析器 | 适合团队协作 |
| 挂载 GraphQL 端点 | app.add_route("/graphql", GraphQLApp(...)) | 将 GraphQL 集成到 FastAPI | from strawberry.fastapi import GraphQLRouter / router = GraphQLRouter(schema) / app.include_router(router, prefix="/graphql") | 共享同一服务器 |
| 类型安全 | 自动生成类型 | 减少错误 | 编辑器自动补全 | 提高开发效率 |