【AI大模型应用开发工程师特训营】第06讲：零基础 FastAPI
目录
3.6.2 使用 File() 和 UploadFile 接收文件
4.2.3 response_model_by_alias 控制输出
5.1 认识 Depends：FastAPI 的依赖注入系统
6.2.1 @app.exception_handler 装饰器
7.2.3 TrustedHostMiddleware 限制主机名
8.4.2 使用 run_in_threadpool 转移到线程池
9.1.2 OAuth2PasswordBearer 配置 — 详细案例
9.1.3 OAuth2PasswordRequestForm 接收登录表单 — 详细案例
第 1 章 路由与请求处理
1.1 安装与环境搭建
1.1.1 创建 Python 虚拟环境
Python 虚拟环境是一个独立的目录，其中包含特定版本的 Python 解释器以及独立于系统全局或其他项目的包安装空间；它通过隔离依赖关系，让你能为每个项目自由选择不同版本的库，避免版本冲突，并随时用 venv 或 virtualenv 等工具创建、激活与管理，确保项目环境的一致性和可复现性。
生活案例：你同时为两家公司做账，一家用人民币记账，一家用美元记账。如果把账本混在一起，月底一定会乱套。虚拟环境就是给每个项目一个独立的账本和计算器，互不干扰。
实际操作：
# 创建项目文件夹
mkdir my_fastapi_project
cd my_fastapi_project
# 创建虚拟环境
python3 -m venv venv解释：
-
venv是一个文件夹，里面装了一份独立的 Python 解释器和包管理器。 -
激活后，你用
pip install安装的库全都会放在这个文件夹里，不会污染系统环境。
1.1.2 安装 FastAPI 和 Uvicorn
FastAPI 是一个现代、高性能的 Python Web 框架，专为构建 API 而设计。它基于 Python 类型提示，能够自动生成交互式 API 文档（Swagger UI 和 ReDoc），并提供数据验证、序列化和依赖注入等开箱即用的功能。由于其底层使用 Starlette 处理请求响应、Pydantic 进行数据校验，FastAPI 不仅开发效率高，而且性能接近 Node.js 和 Go 等语言编写的框架，非常适合用于生产级 RESTful 服务或异步接口开发。
Uvicorn 是一个基于 uvloop 和 httptools 构建的闪电级 ASGI 服务器，用于运行异步 Python Web 应用（如 FastAPI、Starlette 等）。它支持 HTTP/1.1 和 WebSockets，并以高并发、低资源消耗著称。Uvicorn 作为网关服务器，负责接收网络请求、调用应用程序的异步处理函数，并将响应返回给客户端，是部署 FastAPI 应用的首选服务器之一。
生活案例：FastAPI 是“前台接待员”，Uvicorn 是“接待员坐的工位（电脑+电话）”。接待员负责处理客户请求，工位负责让接待员能接到电话。
实际操作：
# 激活虚拟环境
# Windows:
venv\Scripts\activate
# macOS/Linux:
source venv/bin/activate
# 安装
pip install fastapi uvicorn1.1.3 验证安装与项目目录结构
验证：
python -c "import fastapi; print(fastapi.__version__)"如果输出类似 0.136.3，说明安装成功。
1.2 第一个 FastAPI 应用
1.2.1 编写“Hello World”接口
生活案例：
在律师事务所，前台接待员会记住“如果有人来咨询首页，就说欢迎语”。这个“记住”的过程就是定义路由。
代码：在根目录下创建 main.py（不是venv文件中）
# FastAPI 是整个框架的核心，提供了创建API应用所需的路由、请求处理、依赖注入等功能。
from fastapi import FastAPI
# 创建 FastAPI 应用实例，用于定义路由和处理请求。这里我们设置了应用的标题为 "My FastAPI Application"。
app = FastAPI(title="My FastAPI Application")
# 这是一个装饰器，定义了一个 GET 请求的路由。当用户访问根路径 "/" 时，会调用这个函数。函数返回一个 JSON 响应，包含一个键 "message" 和对应的值 "Hello, World!"。
@app.get("/")
def read_root():
# 返回一个 JSON 响应，包含一个键 "message" 和对应的值 "Hello, World!"。
return {"message": "Hello, World!"}解释：
-
@app.get("/")装饰器的意思是：当有人用 GET 方法访问/这个地址时，执行下面的函数。 -
函数返回一个字典，
FastAPI自动转成 JSON 发给用户。 -
你设置在
FastAPI实例中的title（比如"My FastAPI Application"），主要会显示在你自动生成的 API 文档页面中。
1.2.2 启动 Uvicorn 服务器
uvicorn main:app --reload-
main：文件名（main.py） -
app：变量名 -
--reload：代码修改后自动重启（开发用）
1.2.3 浏览器访问验证
打开 http://127.0.0.1:8000，看到 {"message":"Hello World"} 表示成功！
1.3 路由基础：装饰器与 HTTP 方法
装饰器 是 Python 中一种用于修改或增强函数行为的语法结构，它以 @decorator_name 的形式写在函数定义的上方，本质上是一个接收被装饰函数作为参数、并返回一个新函数的高阶函数。在 FastAPI 或 Flask 等 Web 框架中，装饰器（如 @app.get("/path")）被用来将某个函数“注册”到指定的路由上，并同时定义该路由允许的 HTTP 方法（如 GET、POST 等）。这样一来，当客户端请求该 URL 时，框架就能自动调用对应的函数，极大地简化了路由与处理函数的关联过程。
就像你去餐厅点了一份“番茄炒蛋”（这是一个普通函数），然后你跟服务员说“加一份备注：少油、不加糖”。服务员在你的点菜单上贴了一个便利贴（这就是装饰器）。这个便利贴并没有改变“番茄炒蛋”这道菜本身的烹饪步骤，但它会告诉厨师在制作时额外做点调整。在代码里，装饰器就是在不修改函数本身代码的情况下，给函数附加额外功能（比如记录日志、验证权限等）。
HTTP 方法 是 HTTP 协议中用于指定请求意图的操作动词，常见的有 GET（获取资源）、POST（提交数据）、PUT（替换资源）、DELETE（删除资源）等，每种方法都隐含了不同的语义和数据传输方式。在 RESTful API 设计中，方法决定了客户端希望服务器执行的动作，服务器端则根据不同的方法实现相应的业务逻辑。Web 框架通常通过在装饰器中显式声明方法来区分同一 URL 的不同操作，例如 @app.get("/items") 和 @app.post("/items") 会分别处理获取资源列表和创建新资源的请求，使接口设计更加清晰、符合规范。
就像你对服务员的四种不同“指令”。你说“看看菜单”就是 GET（只获取信息，不改变状态）；你说“点一份牛排”就是 POST（新增资源）；你说“把我那份牛排换成七分熟”就是 PUT（修改已有资源）；你说“把我的牛排退掉”就是 DELETE（删除资源）。服务员（相当于 Web 服务器）会根据你发出的不同指令，执行不同的操作流程，但地址（餐厅）和菜品（资源路径）可能是相同的。
1.3.1 @app.get() 的用法
在 Web 框架（如 FastAPI）的路由中，get() 是一个用于注册 HTTP GET 请求 处理函数的方法，它作为装饰器 @app.get("/path") 的核心部分，将函数绑定到指定 URL 和 HTTP 方法上；当客户端以 GET 方式请求该路径时，框架会自动调用被装饰的函数、解析查询参数，并将函数返回的数据以 JSON 等形式响应给客户端，从而让开发者只需关注业务逻辑而非底层请求处理。
生活案例：你去银行，不同的窗口办不同的事：1 号窗口“取钱”（GET），2 号窗口“存钱”（POST）。每个窗口对应一个路由，HTTP 方法就是你要办的业务类型。
代码：
@app.get("/items")
def get_items():
return ["item1", "item2"]输出如下内容：
1.3.2 @app.post() 及其他方法
在 Web 框架的路由中，post() 方法用于注册处理 HTTP POST 请求的函数，通常与创建资源或提交数据（如表单、JSON）的场景绑定；当客户端向指定 URL 发送 POST 请求时，框架会自动将请求体中的参数解析并传递给被装饰的函数，函数执行后返回响应，而 POST 与 GET 最大的区别在于它不要求幂等、可以携带大量数据，并常用于修改服务器状态（如新增数据库记录）。
生活案例：GET 是“拿来”（读），POST 是“交上去”（创建），PUT 是“全部更换”，DELETE 是“销毁”。
@app.post("/items")
def create_item():
return {"msg": "创建成功"}在 HTTP 请求中，请求方法（GET 或 POST）决定了执行哪个函数：当客户端（浏览器、curl、DBeaver 等）向 /items 发送 GET 请求时，服务器会自动匹配到 @app.get("/items") 下的 get_items 函数；发送 POST 请求时，则会匹配到 @app.post("/items") 下的 create_item 函数。你可以在命令行用 curl -X GET http://127.0.0.1:8000/items 和 curl -X POST http://127.0.0.1:8000/items 分别测试，或者用浏览器直接访问地址（浏览器默认发送 GET 请求）来观察对应的返回内容。
1.3.3 路由函数的返回值
在 FastAPI 的路由函数中，你可以直接返回 Python 的字典（dict）、列表（list）、字符串（str），甚至是后续章节会详细介绍的 Pydantic 模型对象。FastAPI 会自动将这些返回值序列化为 JSON 格式的响应体，无需手动调用 json.dumps() 或设置响应类型。
1.4 路径参数
路径参数是 URL 路径中用于动态传递数据的一部分，通常用花括号 {} 在路由定义中声明（如 /users/{user_id}），当客户端请求该 URL 时，框架会自动从路径中提取对应的值并注入到路由函数的同名参数中，从而让服务器能够根据不同的参数值返回差异化的响应内容。
1.4.1 动态 URL 段 {param}
生活案例：图书馆找书，索书号 BOOK-001，BOOK-002……不可能为每本书都单独配一个管理员，而是让同一个管理员根据索书号去取书。/books/BOOK-001 中的 BOOK-001 就是路径参数。
代码：
@app.get("/books/{book_id}")
def get_book(book_id):
return {"book_id": book_id, "title": f"书名{book_id}"}访问 http://127.0.0.1:8000/books/5，返回 {"book_id":5, "title":"书名5"}。
1.4.2 路径参数的类型标注与自动转换
@app.get("/books/{book_id}")
def get_book(book_id:int):
return {"book_id": book_id, "title": f"书名{book_id}"}加上类型 int 后，FastAPI 会自动把字符串 "5" 转成整数 5。如果传入 abc，它会在到达你的函数之前就返回错误："value is not a valid integer"。
1.4.3 枚举类型路径参数
枚举类型路径参数是指利用 Python 的 Enum 子类定义一组预置的可选值，并将其直接用作路径参数的类型声明（例如 @app.get("/items/{category}") 中 category: ItemCategory），这样 FastAPI 会自动校验请求的路径值是否属于枚举成员，若不属于则返回清晰的错误提示，同时生成的 API 文档中也会列出所有允许的取值，从而让接口更规范、自描述且不易出错。
生活案例：奶茶店只能选“大杯”“中杯”“小杯”。枚举就是限定死可选值。
from fastapi import FastAPI # 导入 FastAPI 框架
from enum import Enum # 导入枚举模块
app = FastAPI() # 创建 FastAPI 应用实例
# 定义一个枚举类，继承自 str 和 Enum，使得枚举值同时具有字符串特性
class Size(str, Enum):
small = "小杯" # 枚举成员 small，对应值 "小杯"
medium = "中杯" # 枚举成员 medium，对应值 "中杯"
large = "大杯" # 枚举成员 large，对应值 "大杯"
# 定义一个 GET 路由，路径中包含路径参数 {size}
@app.get("/drink/{size}")
def get_drink(size: Size): # 将路径参数 size 的类型声明为枚举 Size
# FastAPI 会自动校验传入的 size 是否为 "小杯"/"中杯"/"大杯"，
# 若不是则自动返回 422 错误，并提示允许的值。
# 由于枚举继承自 str，size 可以直接作为字符串使用。
return {"message": f"你选择了{size}的饮料"}1.5 查询参数
1.5.1 什么是查询字符串
查询字符串是 URL 中问号 ? 之后的部分，由多个 键=值 对组成，对之间用 & 分隔（例如 /items?page=2&limit=10），它用于向服务器传递非必须的、通常用于过滤、排序或分页的附加参数；与路径参数（用于定位资源）不同，查询字符串不影响资源在服务器上的唯一地址，且 FastAPI 会自动将其解析为路由函数中对应名称的参数，并支持默认值、可选类型和自动校验。
生活案例：百度搜索 ?wd=天气，?wd=天气 就是查询参数，告诉服务器你要搜“天气”。
1.5.2 定义必填与可选查询参数
代码：
@app.get("/search")
def search(keyword: str, page: int = 1):
return {"keyword": keyword, "page": page}-
keyword没有默认值，必填。 -
page有默认值1，可选。
访问：http://127.0.0.1:8000/search?keyword=数据治理
返回：{"keyword":"数据治理","page":1}
1.5.3 查询参数的默认值
在路由函数中，只要一个参数没有出现在路径参数中（即不在 URL 的 {} 内），它就会被 FastAPI 自动识别为查询参数；如果为该参数指定了默认值（例如 limit: int = 10），那么这个查询参数就变成可选参数—当请求未携带该参数时，框架会自动使用默认值。
下面是一个完整的案例，演示 FastAPI 中查询参数的默认值：
from fastapi import FastAPI
app = FastAPI()
# 模拟商品数据
fake_items_db = [
{"name": "鼠标", "price": 99},
{"name": "键盘", "price": 299},
{"name": "显示器", "price": 1299},
]
# 定义 GET 路由，路径为 /items
# 查询参数 skip：表示跳过前几条数据，默认值为 0（可选）
# 查询参数 limit：表示最多返回多少条数据，默认值为 10（可选）
@app.get("/items")
def list_items(skip: int = 0, limit: int = 10):
"""
获取商品列表，支持分页。
- skip: 从第几条开始（默认 0）
- limit: 返回多少条（默认 10）
"""
# 使用切片模拟分页
result = fake_items_db[skip : skip + limit]
return {"total": len(fake_items_db), "data": result}如何验证默认值效果？
如果服务未启动，使用如下命令启动服务：
uvicorn main:app --reload（1）不提供任何查询参数（使用默认值）
-
访问
http://127.0.0.1:8000/items
-
→ 相当于
skip=0, limit=10，返回所有商品（共 3 条）。
（2）只提供部分查询参数
-
访问
http://127.0.0.1:8000/items?skip=1
-
→
skip被指定为 1，limit仍使用默认值 10，返回第 2、3 个商品。
（3）覆盖默认值
-
访问
http://127.0.0.1:8000/items?skip=1&limit=1
-
→ 跳过 1 条，只取 1 条，返回
[{"name": "键盘", "price": 299}]。
关键点解释
-
参数
skip和limit未出现在路径"/items"中，所以它们自动成为查询参数。 -
通过
= 0和= 10赋予了默认值，因此它们是可选参数，客户端可以不传。 -
FastAPI会根据函数签名自动解析查询字符串并执行类型转换（int）。 -
生成的交互式文档（
/docs）中会清晰地列出这些参数及其默认值。
第 2 章 查看自动文档 — Swagger UI
2.1 访问交互式文档
Swagger UI 是一个开源的、交互式的 API 文档生成工具，它能根据 OpenAPI 规范自动将后端的 API 接口以可视化的网页形式展现出来；开发者无需手动编写文档，只需访问 /docs 路径即可看到所有路由、请求参数、响应格式，并且可以直接在页面上填写参数并发送真实请求来测试接口，极大方便了 API 的设计、调试和前后端协作。
（1）启动服务后打开 /docs
服务运行后，访问 http://127.0.0.1:8000/docs，你会看到一个漂亮的界面，这就是 Swagger UI。
（2）界面区域介绍
-
顶部：应用标题、描述、版本
-
每个路由展开后可以看到参数说明、示例
-
Try it out 按钮：可以直接在这里发请求测试
（3）理解自动生成的请求/响应示例
Swagger 会根据你的代码（类型标注、Pydantic 模型）自动生成 JSON 示例，非常直观。
2.2 使用“Try it out”测试接口
第一步：填入参数并执行
找到某个 GET 路由，点击 Try it out，填入查询参数，点击 Execute。
第二步：查看响应状态码与结果
你会看到 Code 200 和下面的响应体。
第三步：通过文档理解接口定义
你定义了什么参数、什么类型、什么必填，Swagger 全知道，还能告诉你参数对不对。
2.3 文档与代码的同步关系
（1） 修改代码后文档的自动更新
你修改路由函数、参数或模型后，刷新 /docs，文档立即变化。没有任何额外配置。
（2）利用文档调试路径参数和查询参数
Swagger 让你在正式写前端之前，就能自己测试后端，极大提升效率。
（3）Swagger UI 的实用价值
它是前后端协作的“契约”，前端工程师看文档就能知道该怎么调接口。
第 3 章 Pydantic 模型与请求体
Pydantic 模型：Pydantic 模型是一个继承自 pydantic.BaseModel 的 Python 类，通过类型注解声明数据的字段和类型，它能够自动对输入数据进行类型校验、转换和错误提示，并提供 .dict() 和 .json() 等便捷方法，常用于定义 API 请求体、响应体或配置对象的统一结构。
请求体：请求体是 HTTP 请求中携带在消息主体（Body）里的数据，通常用于 POST、PUT 等需要向服务器提交信息的请求中；在 FastAPI 中，当路由函数的参数类型被声明为一个 Pydantic 模型时，框架会自动从请求体中读取 JSON 数据、按模型校验并填充到该参数中，从而无需手动解析 request.body()。
3.1 为什么需要数据校验
3.1.1 无校验时的常见问题
如果你不用校验，用户可能把“年龄”填成 "abc"，把邮箱写成 12345，导致程序崩溃。
生活案例：银行开户，你填的表格如果生日写“2 月 30 日”，前台会立刻告诉你日期不对，这就是输入校验。
3.1.2 FastAPI 中 Pydantic 的角色
在 FastAPI 中，Pydantic 扮演着数据契约的核心角色：它通过声明式的 Python 类来定义请求体、查询参数、路径参数及响应数据的结构，自动完成类型校验、数据转换、错误提示和 JSON 序列化/反序列化，使得开发者无需编写繁琐的验证逻辑就能确保接口数据的安全性和一致性，同时为自动生成的 API 文档提供精确的字段描述。Pydantic 就是帮你自动检查数据格式的“前台小能手”，把错误挡在门外。
3.1.3 类型提示与自动校验
FastAPI + Pydantic 只需声明类型，就能自动校验。
3.2 创建请求体模型
3.2.1 BaseModel 与字段定义
from pydantic import BaseModel
class User(BaseModel):
name: str
age: int3.2.2 字段类型：str、int、bool 等
Pydantic 支持 Python 原生类型（如 str、int、bool、float、list、dict、datetime 等），还扩展了许多专用的校验类型，例如：
-
EmailStr：自动校验邮箱格式（需安装email-validator）。 -
Url：校验 URL 合法性。 -
UUID：验证 UUID 字符串。 -
FilePath、DirectoryPath：检查路径是否存在。 -
Json：接收 JSON 字符串并自动解析。 -
PositiveInt、NegativeInt：限制正负整数。 -
constr、conint等约束类型（如constr(min_length=1, max_length=10)）。
通过在模型字段中声明这些类型，Pydantic 会在赋值时自动验证并转换，不符合要求则抛出清晰的错误，极大增强了数据健壮性。
3.2.3 使用 Field() 添加校验规则
在 Pydantic 模型中，Field() 函数用于为字段附加额外的校验规则和元数据：通过指定 min_length、max_length、ge、le、regex 等参数，可以限制字符串长度、数值范围、正则匹配等，并且还能设置 title、description 等用于 API 文档的说明；当数据不符合规则时，Pydantic 会自动抛出包含详细错误信息的异常，从而在数据源头保证数据的合法性。
from pydantic import Field
class User(BaseModel):
name: str = Field(min_length=2, max_length=50)
age: int = Field(ge=0, le=120)3.3 接收 JSON 请求体
（1）@app.post() 中声明模型参数
@app.post("/users")
def create_user(user: User):
return {"name": user.name, "age": user.age}（2）请求体 JSON 的自动解析
当请求发来 {"name": "张三", "age": 25}，FastAPI 自动解析并校验。
（3）发送请求测试
在 Swagger 的 /users 接口 Try it out 中填入 JSON，点击执行。
输出如下内容：
3.4 字段校验详解
-
没有默认值的字段是必填。
-
name: str = "匿名"这样就有了默认值。
（1）字符串长度限制
title: str = Field(min_length=1, max_length=100)（2）数值范围
age: int = Field(gt=0, lt=150) # >0 且 <150（3）正则表达式校验
phone: str = Field(pattern=r"^\d{11}$")3.5 嵌套模型与复杂结构
3.5.1 模型内嵌另一个模型
生活案例：员工信息里包含“地址”，地址又包含“省、市、区”。这就是嵌套。
class Address(BaseModel):
province: str
city: str
class Employee(BaseModel):
name: str
address: Address3.5.2 列表中包含模型
class Company(BaseModel):
employees: list[Employee] # Python 3.9+ 可用 list3.5.3 字典与可选嵌套字段
from typing import Optional
metadata: Optional[dict] = None3.6 表单与文件上传
3.6.1 使用 Form() 接收表单字段
在 FastAPI 中，Form() 就像银行柜台前的纸质表格 — 当用户通过网页表单提交数据（比如登录时填写的用户名和密码）时，这些数据不是放在 URL 里，而是藏在请求体中以表单格式传来。Form() 的作用就是告诉 FastAPI：“帮我把这张表格里对应的字段取出来，并验证一下填写的内容是否合规”。例如，当你把用户名和密码填入登录框点击提交，后端用 username: str = Form() 和 password: str = Form() 就能自动提取这两个字段，省去你手动解析原始数据的麻烦。
from fastapi import Form
app = FastAPI(title="My FastAPI Application")
@app.post("/login")
def login(username: str = Form(), password: str = Form()):
return {"username": username}3.6.2 使用 File() 和 UploadFile 接收文件
就像去快递站寄包裹，你把填好的寄件单（表单字段）和要寄的文件一起交给工作人员。FastAPI 里的 File() 就像工作人员直接伸手接过你递来的薄薄信封——适合小文件，整个内容会直接存到内存里，处理简单。而 UploadFile 则像工作人员推来一辆小推车，让你把大件包裹放上去，他再推到后台慢慢处理——它专为大文件设计，会异步读写临时文件，同时还能贴心地告诉你文件名、大小和类型。无论哪种，你只需在接口里声明 file: bytes = File() 或 file: UploadFile = File()，FastAPI 就会自动帮你把客户端上传的文件“签收”下来，省去你手动解码原始数据流的麻烦。
FastAPI 在处理表单数据（Form()）和文件上传（File()）时，需要 python-multipart 这个库。你的环境里还没装。
pip install python-multipart# 1. 从 fastapi 库中导入所需的类
# File: 用于声明“这是一个文件字段”
# UploadFile: 文件对象类型，支持大文件异步读写
from fastapi import FastAPI, File, UploadFile
# 2. 创建 FastAPI 应用实例
# title 参数用于设置 API 文档的标题（显示在 Swagger 页面顶部）
app = FastAPI(title="My FastAPI Application")
# 3. 定义文件上传接口
# @app.post("/upload") 表示这个函数处理 POST 请求，路径为 /upload
@app.post("/upload")
async def upload_file(
# file: UploadFile = File()
# 参数名 file 是你在接口中使用的变量名
# UploadFile 类型告诉 FastAPI 这是一个文件对象
# File() 是声明“这个参数应该从请求体中的文件字段获取”
# 注意：File() 必须在参数默认值里调用，不能省略
file: UploadFile = File(description="要上传的文件")
):
# 3.1 读取文件内容
# file.read() 是异步方法，需要用 await 等待
# 它会返回文件的全部字节内容（bytes 类型）
content = await file.read()
# 3.2 返回文件信息
# file.filename 是上传文件时客户端提供的文件名
# len(content) 是文件内容的字节数（即文件大小）
return {
"filename": file.filename, # 原文件名
"size": len(content) # 文件大小（字节）
}async 和 await 是 Python 中实现异步编程的关键字。它们的作用可以这样理解：
生活案例：咖啡店点单
你去咖啡店点一杯拿铁：
-
同步方式（没有 async/await）：你站在柜台前，说完“一杯拿铁”后就什么也不干，干等着，直到咖啡师做好递给你，你才离开。期间你不能回消息、不能看菜单，整个店的服务效率很低。
-
异步方式（使用 async/await）：你点完单，拿到一个取餐震动器（相当于一个“等待凭证”）。你不再傻等，而是坐到旁边刷手机、回邮件。当取餐器震动时（表示咖啡好了），你再回到柜台取走咖啡。
在这段时间里，咖啡师也没闲着，他在你等待的时候已经开始为下一位客人做另一杯饮品了。
代码中的对应关系
-
async：标记一个函数是“异步函数”，表明这个函数里面可能会有需要等待的操作，并且它执行时不会阻塞别人。
async def make_coffee():
...-
await：后面跟一个“需要等待的任务”（比如读取文件、网络请求、数据库查询）。await的意思是：“我先去干别的事了，你好了叫我”。当前任务会暂停，让出控制权给其他任务，等这个操作完成后再回来继续执行。
content = await file.read() # 读取文件，不阻塞其他请求在 FastAPI 中的实际作用
FastAPI 原生支持异步，当你写 async def upload_file(...): 并用 await file.read() 时：
（1）高并发处理：当这个请求在读取文件（可能是慢速的上传过程）时，服务器不会傻等，而是立即去处理其他请求。
（2）不阻塞主线程：await 让出 CPU，等到文件读取完成再唤醒这个函数继续执行后面的代码。
（3）性能提升：相比同步方式每个请求独占线程，异步方式可以用极少的线程处理成百上千的并发连接。
一句话总结：async 定义一个可以“暂停”和“让路”的函数，await 让出控制权去处理其他事情，直到需要等待的耗时操作完成。它们共同让程序像高效的咖啡店一样，在“等待”的时候也能忙其他事。
3.6.3 混合表单与文件上传
在实际业务中，用户经常需要同时提交文字信息和文件 — 就像你去银行办业务，柜员不仅让你填一张纸质申请表（表单字段），还会让你把身份证复印件（文件）一起交上去。FastAPI 可以轻松处理这种“一单多物”的请求，你只需在同一个接口中同时声明 Form() 和 File()/UploadFile 参数即可。
生活案例：想象你要通过一个在线平台提交报销申请：你需要在网页上填写报销事由、金额等文字信息，同时上传发票的扫描件。点击“提交”按钮后，所有数据会一起发送到后端。在 FastAPI 中，这些文字信息用 Form() 接收，发票文件用 File() 或 UploadFile 接收，它们可以无缝共存于同一个接口。
代码示例
# 1. 导入 FastAPI 框架及所需组件
# FastAPI: 创建应用实例
# Form: 声明表单字段（接收用户填写的文字数据）
# File: 声明文件字段（接收上传的文件）
# UploadFile: 文件对象类型，支持异步读取，适合大文件
from fastapi import FastAPI, Form, File, UploadFile
# 2. 创建 FastAPI 应用实例
# 这个 app 是整个服务的核心，所有路由都会注册在它上面
app = FastAPI()
# 3. 定义混合表单与文件上传的接口
# @app.post("/submit-expense") 表示：
# - 这是一个 POST 接口（适合提交数据）
# - 访问路径为 /submit-expense
@app.post("/submit-expense")
async def submit_expense(
# 3.1 报销事由：文字字段
# reason: str 表示该字段的值应为字符串类型
# = Form(description="报销事由") 表示：
# - 该参数从请求体中的表单字段获取（不是 JSON）
# - description 会出现在自动生成的 Swagger 文档中
reason: str = Form(description="报销事由"),
# 3.2 报销金额：浮点数字段
# amount: float 表示该字段的值应为浮点数（如 1234.56）
# Form() 同样声明这是一个表单字段
amount: float = Form(description="报销金额"),
# 3.3 发票文件：文件字段
# invoice: UploadFile 表示该参数是一个上传的文件对象
# = File(description="发票文件") 表示：
# - 该参数从请求体中的文件字段获取
# - File() 必须写在默认值位置，不能省略
# UploadFile 提供了异步读取、文件名、MIME类型等属性
invoice: UploadFile = File(description="发票文件")
):
# 4. 异步读取发票文件内容
# invoice.read() 是异步方法，返回文件的全部字节数据（bytes 类型）
# 使用 await 关键字等待读取完成，期间不阻塞服务器处理其他请求
# 适合大文件场景，但如果文件极大（如 GB 级），应考虑分块读取
file_bytes = await invoice.read()
# 5. 返回响应 JSON
# FastAPI 会自动将字典序列化为 JSON 格式并设置响应头
# 返回内容包含：
# - message: 提交成功的提示信息
# - reason: 用户填写的报销事由
# - amount: 用户填写的报销金额
# - filename: 原始文件名（从客户端上传时获取）
# - file_size: 文件大小（字节数）
return {
"message": "报销申请已提交",
"reason": reason,
"amount": amount,
"filename": invoice.filename, # 上传时的原始文件名
"file_size": len(file_bytes) # 计算文件字节数
}输出如下内容：
关键点解释
-
同一接口，多种参数：路径函数中，
reason和amount通过Form()接收，invoice通过File()接收，三者互不干扰。 -
必须使用
python-multipart：混合使用表单和文件时，请求会以multipart/form-data格式编码，这个库是 FastAPI 解析该格式的基础。安装命令：pip install python-multipart。 -
文件可大可小：小文件用
bytes = File()直接读入内存；大文件用UploadFile，它提供异步读取、文件名、MIME 类型等属性，性能更优。 -
请求格式注意：调用这个接口时，不能用普通的 JSON 格式，而必须用
multipart/form-data方式提交。在 Swagger UI（/docs）中，它会自动生成合适的输入表单，包括文件选择器，非常方便测试。
3.7 请求体示例与文档增强
就像你买了一个多功能料理机，说明书里不仅列出了每个按钮的功能（接口定义），还贴心地配上了几道“新手菜谱” — 比如“番茄牛腩”需要哪些食材、各放多少克。在 FastAPI 中，请求体示例就是这些菜谱：你用 model_config 里的 json_schema_extra 给每个接口提前写好一个或多个请求样例，Swagger 文档就会自动展示它们，让前端或调用方一眼就知道“该怎么填”。而文档增强就是升级说明书：你给按钮加个标签、写段生动的介绍、标出哪些按钮已废弃，甚至手动修改 OpenAPI 结构，最终让 /docs 变成一份新员工看一眼就能直接上手、无需再问你的“活文档”。
以下是一个完整的 FastAPI 案例，演示如何利用 请求体示例 和 文档增强 让 API 文档更友好、更直观。我们以一个“任务管理”场景为例，就像使用一个团队协作看板，你需要创建任务，而新同事只需看文档就能知道如何填写请求。
from fastapi import FastAPI
from pydantic import BaseModel, Field
from typing import Optional
from enum import Enum
# ---------- 创建应用 ----------
app = FastAPI(
title="团队任务管理 API",
description="一个用于团队创建、查看、管理任务的接口服务。",
version="1.0.0"
)
# ---------- 辅助枚举 ----------
class Priority(str, Enum):
low = "低"
medium = "中"
high = "高"
# ---------- 请求体模型（含示例和字段说明） ----------
class CreateTaskRequest(BaseModel):
"""
创建任务的请求体
- title: 任务标题，不能为空，2~100字符
- description: 任务详细描述，可选
- priority: 任务优先级，默认中
- assignee: 负责人，默认未分配
"""
title: str = Field(
...,
min_length=2,
max_length=100,
description="任务标题，简明扼要",
examples=["编写第一季度数据治理报告"]
)
description: Optional[str] = Field(
None,
description="详细任务描述，可选",
examples=["根据公司模板，汇总各部门数据，形成最终报告。"]
)
priority: Priority = Field(
Priority.medium,
description="优先级：低、中、高",
examples=[Priority.high]
)
assignee: Optional[str] = Field(
None,
description="任务负责人，未填则为未分配",
examples=["张三"]
)
# 模型配置：提供多个请求示例给 Swagger
model_config = {
"json_schema_extra": {
"examples": [
{
"title": "编写第一季度数据治理报告",
"description": "根据公司模板，汇总各部门数据，形成最终报告。",
"priority": "高",
"assignee": "张三"
},
{
"title": "更新数据字典",
"priority": "中",
"assignee": "李四"
}
]
}
}
# ---------- 响应模型（可隐藏部分字段） ----------
class TaskResponse(BaseModel):
id: int
title: str
priority: str
assignee: Optional[str] = None
message: str
# ---------- 路由：创建任务 ----------
@app.post(
"/tasks",
response_model=TaskResponse,
tags=["任务管理"], # 在文档中对接口分组
summary="创建一个新任务", # Swagger 上的简要描述
description="传入任务标题、描述、优先级等信息，创建一个新任务。优先级可设为低/中/高。",
response_description="返回新创建任务的基本信息和状态提示"
)
async def create_task(task: CreateTaskRequest):
"""
内部详细说明（不会直接显示在 Swagger 概述中，但会出现在 OpenAPI 的描述字段）
"""
# 模拟保存到数据库并返回 ID
fake_id = 42
return TaskResponse(
id=fake_id,
title=task.title,
priority=task.priority.value,
assignee=task.assignee or "未分配",
message="任务创建成功"
)
# ---------- 路由：查看所有任务（带标签和废弃标记） ----------
@app.get(
"/tasks",
tags=["任务管理"],
summary="获取任务列表",
deprecated=True, # 标记接口已废弃，文档中会显示警告
description="此接口已废弃，请使用 /api/v2/tasks 获取任务列表。"
)
async def list_tasks():
return {"message": "此接口已废弃"}
if __name__ == "__main__":
import uvicorn
uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)运行与查看文档
（1）安装依赖（如果需要表单/文件功能才需要 python-multipart，这里不需要）：
pip install fastapi uvicorn（2）将代码保存为 main.py，运行：
uvicorn main:app --reload（3）打开 http://127.0.0.1:8000/docs 查看 Swagger UI。
你会看到什么？
-
顶部的应用标题、描述、版本（来自
FastAPI()构造函数）。 -
分组标签：“任务管理” 将两个接口归类到一起，使文档结构更清晰。
-
每个接口的摘要和详细描述：点开接口，摘要显示在标题位置，描述区域显示更长的说明。
-
请求体示例：在
POST /tasks的请求体区域，会展示你预先定义的两个示例（examples），并可以点击切换，方便测试。 -
废弃标记：
GET /tasks旁边会有“deprecated”标志，提醒调用方不要再用。
生活案例解释
就像团队新来了一位同事，你给他一本《任务操作手册》：
-
请求体示例：就是手册里预先填好的样表，告诉他“标题这样写、优先级选高”。
-
文档增强：就是在手册里给不同章节贴上标签（“任务管理”），给每页加上标题和简介，甚至在某些过时的操作上盖个“已废弃”章。这样他一翻手册就能立刻上手，不需要事事来问你。
通过以上案例，你掌握了如何用 model_config 提供示例、用 tags / summary / description / deprecated 增强文档，让你的 API 文档从“能用”变成“好用”。
第 4 章 响应模型与序列化
response_model 是一种接口契约（Contract），在服务端输出前对返回数据进行声明式过滤与类型校验，只暴露模型定义中显式声明的字段，其余内部属性（如 password_hash）自动裁剪；而序列化（通过内置的 jsonable_encoder 或响应中间件）则是将 Pydantic 模型实例、ORM 对象或 datetime 等复杂类型递归转换为可 JSON 序列化的原生 Python 字典，最终由 FastAPI 底层的 JSONResponse 渲染为合规的 JSON 字符串返回客户端 — 二者协同实现了“输出即契约，多余不泄露，复杂类型不报错”的安全与规范化响应机制。
类似于你去医院做体检，最后拿到手里的是一张体检报告单。医生在后台系统里能看到你的全部原始数据：身份证号、历次病历、甚至医生的内部备注，但交给你的报告单上只打印了血常规、尿常规、心电图这几项关键指标，其余隐私信息一概不显示。这里的“报告单模板”就是 response_model — 它是一份严格的输出契约，规定只返回哪些字段，把不该暴露的内部属性（password_hash、id_number）自动过滤掉。而报告单打印的过程就是序列化 — 医生系统里的数据可能是数据库对象、日期时间、枚举代码等复杂类型，jsonable_encoder 像一台翻译机，把它们统一转成表格里的一行行文字，最终装订成一份规范的 JSON 报告递到你手上。
4.1 使用 response_model 过滤字段
以下是详细案例，演示 response_model 过滤字段、隐藏敏感信息、以及 response_model_exclude 动态排除 的完整用法。
场景：用户管理系统
假设我们有一个用户注册和查询功能。内部模型 UserInDB 包含密码哈希、身份证号等敏感字段，但对外接口绝不能暴露它们。
4.1.1 定义内部模型与输出模型
from fastapi import FastAPI
from pydantic import BaseModel, Field
from typing import Optional
app = FastAPI(title="用户管理 API")
# ---------- 内部模型：数据库中的完整用户信息 ----------
class UserInDB(BaseModel):
id: int
username: str
email: str
password_hash: str # 密码哈希，绝不能返回
id_number: str # 身份证号，绝不能返回
age: int
is_active: bool = True
# ---------- 公开输出模型：仅暴露安全字段 ----------
class UserPublic(BaseModel):
id: int
username: str
email: str
age: int
is_active: bool
# 注意：没有 password_hash 和 id_number4.1.2 模拟数据库与注册接口
# 模拟数据库
fake_db: list[UserInDB] = []
@app.post("/users", response_model=UserPublic, status_code=201)
async def create_user(
username: str = Form(...), # Form(...) 表示该表单字段是必填的
email: str = Form(...),
password: str = Form(...),
id_number: str = Form(...),
age: int = Form(...)
):
"""
注册新用户。内部会保存密码哈希和身份证号，
但返回给客户端时只使用 UserPublic 模型，自动过滤敏感字段。
"""
# 模拟密码哈希（实际应使用 bcrypt 等）
password_hash = f"hashed_{password}"
new_user = UserInDB(
id=len(fake_db) + 1,
username=username,
email=email,
password_hash=password_hash,
id_number=id_number,
age=age
)
fake_db.append(new_user)
return new_user # FastAPI 自动按 response_model 过滤生活案例：就像你去酒店前台登记，你填写了完整的入住单（包含身份证号、银行卡号等）。前台服务员将这张单子归档到内部系统，但给你的“房卡套”上只印着你的姓名和房号，绝不会多透露你的隐私信息。UserPublic 就是那个“房卡套”。
4.1.3 查询接口：进一步动态排除字段
有时需要根据不同调用方返回不同的字段。比如内部管理后台可以看 email，而普通用户互查时连 email 也要隐藏。
# 普通用户查询（隐藏 email 和 age 之外的所有敏感字段）
@app.get("/users/{user_id}", response_model=UserPublic)
async def get_user_public(user_id: int):
"""
根据 ID 查询用户。返回的字段由 UserPublic 决定。
如果想进一步隐藏 email，可在路径装饰器上动态添加 exclude。
"""
user = next((u for u in fake_db if u.id == user_id), None)
if not user:
return {"error": "用户不存在"}
return user
# 内部管理员查询（能看到 email，但不能看身份证号）
# 使用 response_model_exclude 再过滤一层
@app.get(
"/admin/users/{user_id}",
response_model=UserPublic,
response_model_exclude={"id_number", "password_hash"} # 显式排除（尽管这两个字段本就不在 UserPublic 中，这里做双重保险）
)
async def get_user_admin(user_id: int):
"""
管理员接口：虽然使用了 UserPublic，但我们额外显式排除了 id_number，
确保即使模型定义有变动也不会泄露。
"""
user = next((u for u in fake_db if u.id == user_id), None)
if not user:
return {"error": "用户不存在"}
return userresponse_model_exclude 的实际威力：
有时你想在同一个输出模型基础上临时“遮掉”某个字段。例如一个接口平时返回 email，但在特定场景下（如被拉黑的用户），你希望连 email 也不返回，就可以这样：
@app.get("/users/{user_id}/limited", response_model=UserPublic, response_model_exclude={"email"})
async def get_user_limited(user_id: int):
"""返回用户信息，但移除邮箱字段（比如隐私保护模式）"""
user = next((u for u in fake_db if u.id == user_id), None)
return user4.1.4 完整运行测试
（1）安装依赖（需要 python-multipart 因为注册接口用了 Form）：
pip install fastapi uvicorn python-multipart（2）将以上代码保存为 main.py，启动：
uvicorn main:app --reload（3）打开 http://127.0.0.1:8000/docs，测试各接口：
-
用
POST /users注册一个新用户，观察返回的 JSON 中只有id，username，email，age，is_active，完全没有password_hash和id_number。
-
用
GET /users/{user_id}查看用户，同样不会泄露敏感信息。 -
用
GET /users/{user_id}/limited查看同一个用户，会发现连email也消失了。
4.1.5 总结
|
方式 |
作用 |
生活类比 |
|---|---|---|
|
|
声明返回的字段结构，自动过滤多余字段 |
体检报告单：只展示关键指标，隐藏原始详细数据 |
|
模型中不定义敏感字段 |
从源头杜绝泄露（ |
房卡套上根本不印身份证号 |
|
|
在路由层面动态排除特定字段 |
同一份报告单，给保险公司看时隐藏某些疾病历史 |
通过这样的设计，你的 API 既能保证数据安全，又能灵活适应不同场景的输出需求。
以下是针对 响应模型嵌套 和 字段别名 的完整案例，每个知识点都有可运行的代码和生活化类比。
4.2 响应模型嵌套与别名
场景引入：公司部门与员工查询系统
假设你要构建一个公司内部 API，需要返回员工信息，而每个员工都属于一个部门。同时，公司前端团队使用 JavaScript 开发，接口返回的字段习惯用 驼峰命名（如 userName），而后端 Python 代码习惯用 下划线命名（如 user_name）。这时就需要用到嵌套模型和字段别名
4.2.1 返回嵌套对象
生活案例：你去公司前台问“张三在哪个部门？”，前台不会只告诉你一个部门编号，而是直接告诉你“张三在技术部，技术部在 3 楼 301 室”。这里“部门信息”就是嵌套在“员工信息”里的子对象。
完整代码：
from fastapi import FastAPI
from pydantic import BaseModel
from typing import Optional
app = FastAPI(title="公司员工查询系统")
# ---------- 嵌套模型：部门信息 ----------
class Department(BaseModel):
"""部门信息模型"""
dept_id: int
dept_name: str
location: str # 部门所在楼层/房间
manager: Optional[str] = None # 部门负责人
# ---------- 外层模型：员工信息（嵌套部门） ----------
class EmployeeResponse(BaseModel):
"""员工信息输出模型，包含嵌套的部门对象"""
emp_id: int
name: str
title: str # 职位
department: Department # 嵌套整个部门对象，而非只返回部门ID
# ---------- 模拟数据库（内嵌部门数据） ----------
# 模拟的部门数据
tech_dept = Department(
dept_id=1,
dept_name="技术部",
location="3楼301室",
manager="李总"
)
# 模拟的员工数据（包含完整的部门对象）
fake_employees = [
{
"emp_id": 1001,
"name": "张三",
"title": "后端工程师",
"department": tech_dept,
"salary": 25000, # 内部敏感字段，不对外暴露
"id_number": "110101..." # 内部敏感字段
},
{
"emp_id": 1002,
"name": "李四",
"title": "前端工程师",
"department": Department(
dept_id=2,
dept_name="产品部",
location="2楼205室",
manager="王总"
),
"salary": 22000,
"id_number": "110102..."
}
]
# ---------- 查询接口 ----------
@app.get("/employees/{emp_id}", response_model=EmployeeResponse)
async def get_employee(emp_id: int):
"""
根据员工ID查询员工信息。
返回的数据中自动包含嵌套的部门对象，而非孤零零的部门ID。
同时，内部敏感字段（salary、id_number）自动被过滤。
"""
# 模拟数据库查询
employee = next((e for e in fake_employees if e["emp_id"] == emp_id), None)
if not employee:
from fastapi import HTTPException
raise HTTPException(status_code=404, detail="员工不存在")
return employee
# ---------- 启动入口 ----------
if __name__ == "__main__":
import uvicorn
uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)测试：启动服务后，访问 GET /employees/1001，返回如下 JSON：
{
"emp_id": 1001,
"name": "张三",
"title": "后端工程师",
"department": {
"dept_id": 1,
"dept_name": "技术部",
"location": "3楼301室",
"manager": "李总"
}
}注意：
-
department字段返回的是一个完整的对象，而非仅仅部门 ID。 -
内部的
salary和id_number没有出现在响应中，因为EmployeeResponse模型没有定义它们。
4.2.2 使用 alias 字段别名
生活案例：你在国内叫“王伟”，出国后护照上的拼音是“WANG WEI”。在外国酒店前台，他们只认“WANG WEI”，但你的中国朋友还是叫你“王伟”。alias 就是那个“护照上的英文名” — 对外使用，内部代码继续用中文称呼。
完整代码（继续在同一个 main.py 中追加）：
# ---------- 带别名的请求体模型 ----------
class CreateEmployeeRequest(BaseModel):
"""
创建员工的请求体。
前端传的是驼峰命名（JavaScript 习惯），但后端代码用下划线命名。
"""
emp_name: str = Field(alias="empName") # 前端传 empName，代码里用 emp_name
emp_title: str = Field(alias="empTitle") # 前端传 empTitle，代码里用 emp_title
dept_id: int = Field(alias="deptId") # 前端传 deptId，代码里用 dept_id
model_config = {
"json_schema_extra": {
"examples": [
{
"empName": "王五",
"empTitle": "测试工程师",
"deptId": 1
}
]
}
}
# ---------- 创建员工接口 ----------
@app.post("/employees", status_code=201)
async def create_employee(req: CreateEmployeeRequest):
"""
接收前端发来的驼峰命名 JSON，后端用下划线命名处理。
例如前端发：{"empName": "王五", "empTitle": "测试", "deptId": 1}
后端代码中可安全使用：req.emp_name、req.emp_title、req.dept_id
"""
return {
"message": "员工创建成功",
"emp_name": req.emp_name, # 内部使用下划线字段名
"emp_title": req.emp_title,
"dept_id": req.dept_id
}测试：在 Swagger 中，用 POST /employees 发送：
{
"empName": "王五",
"empTitle": "测试工程师",
"deptId": 1
}后端返回：
{
"message": "员工创建成功",
"emp_name": "王五",
"emp_title": "测试工程师",
"dept_id": 1
}关键点：前端用驼峰 empName，后端代码用下划线 emp_name，双方都不需要改变自己的编码习惯。
4.2.3 response_model_by_alias 控制输出
生活案例：你印了两套名片，一套中文给国内客户，一套英文给国外客户。同一个你，但名片上印的名字不同。response_model_by_alias=True 就是决定“对外发英文版名片还是中文版名片”的开关。
完整代码（继续追加）：
# ---------- 带别名的输出模型 ----------
class EmployeeWithAlias(BaseModel):
"""
员工信息输出模型，使用别名。
当 response_model_by_alias=True 时，响应中返回驼峰字段名。
当 response_model_by_alias=False（默认）时，响应中返回下划线字段名。
"""
emp_name: str = Field(alias="empName")
emp_title: str = Field(alias="empTitle")
department_name: str = Field(alias="departmentName")
model_config = {
"populate_by_name": True # ← 允许用原始字段名或别名来构造
}
# ---------- 模拟员工数据 ----------
fake_employee_with_dept = {
"emp_name": "赵六",
"emp_title": "数据工程师",
"department_name": "数据部",
"salary": 30000 # 内部字段，不暴露
}
# ---------- 接口A：默认输出（下划线命名） ----------
@app.get("/employees/{emp_id}/info-v1", response_model=EmployeeWithAlias)
async def get_employee_v1(emp_id: int):
"""
默认行为：response_model_by_alias=False（不传默认就是 False）
返回的 JSON 使用 Python 字段名（下划线格式）。
适合内部系统使用。
"""
return fake_employee_with_dept
# ---------- 接口B：使用别名输出（驼峰命名） ----------
@app.get(
"/employees/{emp_id}/info-v2",
response_model=EmployeeWithAlias,
response_model_by_alias=True # 关键参数：让输出使用别名
)
async def get_employee_v2(emp_id: int):
"""
设置 response_model_by_alias=True。
返回的 JSON 使用别名（驼峰格式），适合前端 JavaScript 直接使用。
"""
return fake_employee_with_dept测试对比：
-
访问
GET /employees/1/info-v1（默认），返回：{ "emp_name": "赵六", "emp_title": "数据工程师", "department_name": "数据部" } -
访问
GET /employees/1/info-v2（别名模式），返回：{ "empName": "赵六", "empTitle": "数据工程师", "departmentName": "数据部" }
总结：同一个 Pydantic 模型，通过 response_model_by_alias 开关，可以灵活适配不同调用方的命名规范，后端代码内部始终使用 Python 风格的下划线命名，不用做任何修改。
完整总结
|
知识点 |
作用 |
生活类比 |
|---|---|---|
|
嵌套模型 |
返回一个对象内部的完整子对象，而非孤零零的外键 ID |
问“张三在哪个部门？”直接告诉你“技术部，3 楼 301”，而不是一个部门编号让你再查一次 |
|
字段别名 |
让前端用驼峰、后端用下划线，双方各写各的 |
国内叫“王伟”，出国用“WANG WEI”，护照上写英文，回家还是中文名 |
|
|
控制最终输出的 JSON 用原始字段名还是别名 |
同一盒名片，见中国客户发中文版，见外国客户发英文版，一键切换 |
4.3 路径/查询参数的高级校验
前面的学习中，我们只给参数标个类型，FastAPI 就能自动校验。但在真实项目中，一个搜索框可能需要限制关键词长度，商品 ID 必须大于 0，或者某个参数要换个名字对外展示。这就需要用到 Query() 和 Path() 两个高级校验函数。
生活案例引入
想象你在一个电商平台工作：
-
搜索商品：搜索框里用户输入的关键词，不能太短（比如只输入“鞋”，结果太多），也不能太长（输入一篇小作文）。就像你去图书馆查书，至少要输入书名的一部分，但也不能把整本书的内容都敲进去。
-
查看商品详情：每个商品都有一个编号，这个编号必须是正整数。如果有人输入
0或-5，系统应该直接提示“商品 ID 不合法”，而不是去数据库里白白查一圈。 -
参数改名：后端的 Python 代码习惯用
search_keyword，但前端同事用的 API 文档里希望显示q（更短），甚至有时候某个参数快下线了，需要给调用方一个“废弃警告”。
FastAPI 的 Query() 和 Path() 就是专门解决这些需求的。
4.3.1 Query() 添加额外校验与元数据
作用：对查询参数（?key=value）附加长度、范围、正则等校验规则，同时提供描述信息，让 Swagger 文档更友好。
生活案例：你用一个在线翻译工具，输入框下方有一行小字：“请输入 2～200 个字符”。如果只输入一个字母，按钮会变灰，这就是前端做的长度校验。后端也得再做一次，防止有人绕过前端直接调接口。
完整代码示例：
from fastapi import FastAPI, Query
app = FastAPI(title="电商搜索服务")
@app.get("/search")
async def search_items(
q: str = Query(
default=..., # 必填
min_length=2, # 最少2个字符
max_length=50, # 最多50个字符
description="商品搜索关键词", # 在 Swagger 中显示的说明
examples=["运动鞋", "无线耳机"] # 文档中的示例值
),
page: int = Query(
default=1,
ge=1, # 页码必须 ≥1
description="页码，从1开始"
),
page_size: int = Query(
default=10,
ge=1,
le=100, # 每页最多100条
description="每页显示条数"
)
):
return {
"keyword": q,
"page": page,
"page_size": page_size,
"message": f"正在搜索与“{q}”相关的商品..."
}代码解释：
-
Query(default=...)中的...表示必填。 -
min_length、max_length控制字符串长度。 -
ge（greater than or equal）、le（less than or equal）控制数值范围。 -
description会出现在/docs文档里，方便调用方理解。 -
examples提供一组示例值，在 Swagger 中可以直接点击填充。
测试：
-
访问
http://127.0.0.1:8000/search?q=鞋→ 正常返回。 -
访问
http://127.0.0.1:8000/search?q=a→ 返回校验错误，提示字符串长度不足。 -
访问
http://127.0.0.1:8000/search?q=鞋&page=0→ 报错，页码必须 ≥1。
4.3.2 Path() 对路径参数添加校验
作用：对路径参数（/items/{item_id}）增加校验规则，本质上和 Query() 用法相同，只是专门用于路径中。
生活案例：去快递站取件，你需要提供取件码。取件码一定是 6 位数字，不可能为负数或零。如果系统发现你输入了 0 号货架，应该直接告诉你“取件码无效”，而不是去仓库里乱翻。
完整代码示例：
from fastapi import FastAPI, Path
app = FastAPI(title="商品管理")
@app.get("/items/{item_id}")
async def get_item(
item_id: int = Path(
default=...,
ge=1, # 商品ID必须 ≥1
le=1000000, # 假设最大ID为100万
description="商品唯一标识ID",
examples=[42, 100]
)
):
# 模拟从数据库查询
return {
"item_id": item_id,
"name": f"商品{item_id}",
"message": "查询成功"
}代码解释：
-
Path()里的参数和Query()几乎一样，只是它作用于路径变量。 -
ge=1确保 ID 为正整数，如果有人访问/items/0或/items/-5，FastAPI 直接返回 422 错误，根本不会进入你的函数。 -
路径参数通常必填，所以第一个参数
default=...不能省略。
测试：
-
访问
http://127.0.0.1:8000/items/100→ 正常。 -
访问
http://127.0.0.1:8000/items/0→ 返回"Input should be greater than or equal to 1"。
4.3.3 参数别名与废弃标记
作用：
-
别名（
alias）：让前端看到的参数名和后端代码里用的变量名不一样。 -
废弃标记（
deprecated）：告诉调用方“这个参数即将下线，请改用新的”。
生活案例：你公司原来的搜索接口用的参数名是 search，现在要统一改成 q。但老客户们还在用 search，直接删掉他们会报错。怎么办？可以让 search 作为别名继续工作一段时间，同时在文档里标个“已废弃”的章，引导大家迁移到 q。
完整代码示例：
from fastapi import FastAPI, Query
app = FastAPI(title="搜索服务升级")
@app.get("/search-v2")
async def search_v2(
# 新参数名（推荐使用）
q: str = Query(
default=...,
min_length=1,
description="搜索关键词（新）",
examples=["数据治理"]
),
# 老参数名，作为别名继续支持，并标记废弃
search: str = Query(
default=None,
alias="search", # 前端还可以传 ?search=xxx
deprecated=True, # 在文档中显示“已废弃”
description="搜索关键词（旧），请使用 q 参数"
)
):
# 实际以新参数 q 为准，如果老参数传了且 q 没传，则兼容处理
keyword = q if q else search
if not keyword:
from fastapi import HTTPException
raise HTTPException(status_code=400, detail="必须提供搜索关键词")
return {"keyword": keyword, "message": f"正在搜索“{keyword}”"}代码解释：
-
alias="search"允许调用方用?search=xxx传参，但代码里通过变量名search来取值。 -
deprecated=True会让 Swagger 文档中这个参数旁边出现醒目的“废弃”标记。 -
实际业务中，新旧参数可以同时存在一段时间，业务逻辑内部做兼容处理。
测试：
-
打开
http://127.0.0.1:8000/docs，你会发现search参数旁边有黄色“Deprecated”标签。 -
用
?q=你好访问 → 正常。 -
用
?search=你好访问 → 依然正常，但响应中可能已不再推荐使用。
小结
-
Query()专用于查询参数（?后面的部分），Path()专用于路径参数（/{item_id}）。 -
两者都可以设置长度、数值范围、正则等约束，并提供
description、examples丰富文档。 -
alias可以让同一参数对外有多个名字，配合deprecated=True可以平滑升级 API，避免破坏老客户。
4.4 序列化辅助工具
在实际开发中，从数据库查出来的数据往往不是纯净的 Python 字典，而是 ORM 对象（如 SQLAlchemy 返回的模型实例），里面可能嵌套着 datetime 时间、Decimal 金额等复杂类型。这些类型无法直接被 json.dumps() 序列化，硬转会报错。FastAPI 提供了一个万能工具箱 — jsonable_encoder，能将这些复杂对象递归地转换成 JSON 兼容的字典或列表，省去你手动逐层转换的麻烦。
4.4.1 jsonable_encoder 的作用
生活案例：你整理了一份员工档案，纸张上有姓名、入职日期、月薪等字段。现在公司要求你把档案录入 Excel 表格并发送邮件。纸上的日期写作“2024 年 3 月 15 日”，但 Excel 只认 2024-03-15 这种标准格式；薪水写着“两万五”，Excel 需要纯数字 25000。你手动把所有字段改成标准格式——这个过程就是 jsonable_encoder 做的事。它自动识别日期、枚举、嵌套对象等，翻译成 JSON 能懂的标准格式。
核心代码：
from fastapi.encoders import jsonable_encoder使用时只需传入任意复杂对象，即可获得可序列化的字典/列表。
4.4.2 将 ORM 对象转换为字典
场景：你从数据库查出用户记录（SQLAlchemy 模型），需要直接返回给前端，但 ORM 对象不是字典，且可能包含 datetime 等无法直接序列化的字段。
模拟 ORM 类（伪代码，模拟 SQLAlchemy 模型）：
from datetime import datetime, date
from decimal import Decimal
from fastapi import FastAPI
from fastapi.encoders import jsonable_encoder
app = FastAPI(title="序列化工具演示")
# ---------- 模拟一个 SQLAlchemy ORM 类 ----------
class FakeORMUser:
"""
此类模拟 SQLAlchemy 从数据库返回的模型实例。
实际中 SQLAlchemy 模型通过 Declarative Base 定义，但底层都是 Python 对象。
"""
def __init__(self, id, username, email, register_date, salary, tags):
self.id = id
self.username = username
self.email = email
self.register_date = register_date # datetime 类型
self.salary = salary # Decimal 类型
self.tags = tags # 列表
# ---------- 模拟数据库查询 ----------
def get_user_from_db(user_id: int) -> FakeORMUser:
"""模拟从数据库查询用户"""
return FakeORMUser(
id=user_id,
username="张三",
email="zhangsan@example.com",
register_date=datetime(2024, 3, 15, 14, 30, 0),
salary=Decimal("25000.50"),
tags=["VIP", "技术部", "年度优秀员工"]
)
# ---------- 接口：直接返回 ORM 对象（会报错） ----------
@app.get("/users/{user_id}/raw")
async def get_user_raw(user_id: int):
"""
错误示范：直接返回 ORM 对象。
如果启用 response_model 可能会因类型不匹配报错；
若不使用 response_model，FastAPI 底层会尝试 jsonable_encoder，但某些复杂类型仍需手动处理。
"""
user = get_user_from_db(user_id)
return user # 可能因 datetime 无法序列化而失败问题：datetime、Decimal 不是 JSON 原生类型，直接返回可能引发 TypeError。
正确做法：使用 jsonable_encoder 预处理。
# ---------- 正确接口：使用 jsonable_encoder 转换 ----------
@app.get("/users/{user_id}")
async def get_user(user_id: int):
"""
正确做法：使用 jsonable_encoder 将 ORM 对象递归转换为字典，
并自动处理 datetime、Decimal 等特殊类型。
"""
user = get_user_from_db(user_id) # 获取 ORM 对象
user_dict = jsonable_encoder(user) # 转换为纯字典
# 此时 user_dict 已经是普通字典，可以直接返回或进一步处理
return {"data": user_dict}返回结果（/users/1）：
{
"data": {
"id": 1,
"username": "张三",
"email": "zhangsan@example.com",
"register_date": "2024-03-15T14:30:00",
"salary": 25000.5,
"tags": ["VIP", "技术部", "年度优秀员工"]
}
}发生了什么？
-
datetime→ 转为 ISO 8601 字符串"2024-03-15T14:30:00"。 -
Decimal("25000.50")→ 转为float类型25000.5（注意精度取舍，生产环境可自定义编码器）。 -
嵌套列表保持不变。
-
整个过程递归执行，无论嵌套多深都自动处理。
4.4.3 处理日期、自定义类型
jsonable_encoder 支持众多内置类型：
|
类型 |
转换结果 |
|---|---|
|
|
ISO 8601 字符串（ |
|
|
ISO 日期字符串（ |
|
|
|
|
|
|
|
|
Base64 字符串 |
|
|
|
自定义类型支持：如果你的模型中有自己定义的类，只需实现 __jsonable_encoder__ 方法，jsonable_encoder 会自动调用它。
自定义类型案例：
from fastapi import FastAPI
from fastapi.encoders import jsonable_encoder
app = FastAPI()
# ---------- 自定义类型：员工状态 ----------
class EmployeeStatus:
def __init__(self, code: int, label: str):
self.code = code
self.label = label
# 实现此方法，告诉 jsonable_encoder 如何转换
def __jsonable_encoder__(self):
return {"code": self.code, "label": self.label}
# ---------- 模型 ----------
class EmployeeInfo:
def __init__(self, name, status):
self.name = name
self.status = status # 自定义类型
@app.get("/employee")
async def get_employee():
emp = EmployeeInfo(name="李四", status=EmployeeStatus(1, "在职"))
return {"employee": jsonable_encoder(emp)}返回结果：
{
"employee": {
"name": "李四",
"status": {
"code": 1,
"label": "在职"
}
}
}自定义类只要定义了 __jsonable_encoder__，就会被自动识别并转换。
总结
-
jsonable_encoder是 FastAPI 内置的“万能序列化器”，能将 ORM 对象、Pydantic 模型、复杂嵌套类型递归转换为 JSON 兼容的字典。 -
它自动处理
datetime→ 字符串、Decimal→ 数字、Enum→ 值等常见转换。 -
对于自定义类型，只需实现
__jsonable_encoder__方法即可优雅扩展。 -
典型使用场景：数据库查询结果直接返回、日志记录、数据导出、将复杂对象存入缓存前转换等。
生活类比：就像一位专业档案管理员，你把一堆五花八门的原始材料（日期、金额、枚举）交给他，他按照统一模板整理成规整的电子表格，保证任何人拿到都能直接使用。你无需操心底层转换细节，安心处理业务逻辑即可。
第 5 章 依赖注入系统
5.1 认识 Depends：FastAPI 的依赖注入系统
5.1.1 什么是依赖注入
生活案例：想象你是项目组的一名开发人员，每天到公司后需要开始工作。你不需要自己去：
-
打开机房空调
-
启动服务器
-
拉取最新代码
-
配置数据库连接
这些“工作前的准备工作”已经由运维团队和项目模板帮你搞定了。你只需坐在工位上，说一声“我要开始工作了”，所需的工具和环境就自动就绪，直接递到你手边。
在 FastAPI 中，依赖注入就是这样一个“后勤保障系统”：
-
路径函数 = 你（专注写业务逻辑）
-
依赖函数 = 运维团队（负责准备工作，比如验证用户身份、获取数据库会话、校验权限）
-
Depends()= 工单系统（你声明需要什么，它就自动调用对应的依赖函数，并把结果送到你手里）
核心思想：把通用逻辑从业务代码中抽离出来，形成可复用的“依赖函数”，需要时用 Depends() 一键注入，避免在每个接口里重复造轮子。
5.1.2 编写第一个依赖函数
生活案例：你开发了一个搜索接口，几乎每个接口都需要两个通用参数：搜索关键词 q 和页码 page。如果在每个接口里都写一遍校验逻辑和默认值，代码会非常冗余。更好的办法是：定义一个通用的“参数提取器”，哪个接口需要，就声明它。
依赖函数就是普通的 Python 函数，可以带参数，也可以返回任意值（字典、对象、数据库会话等）。
第一个依赖函数示例：
# 这是一个普通的异步函数，但可以被 Depends 调用
async def common_parameters(
q: str = None, # 搜索关键词，可选
page: int = 1 # 页码，默认第1页
):
"""
这是一个可复用的依赖函数。
它提取并校验查询参数，返回一个字典。
任何接口都可以通过 Depends(common_parameters) 来使用它。
"""
return {
"query": q,
"page": page,
"timestamp": "2024-01-15" # 还可以额外添加一些通用信息
}函数说明：
-
函数名随意，逻辑随意。
-
可以有自己的参数（这些参数会自动从请求的查询参数中提取）。
-
返回什么，下游的路径函数就拿什么。
5.1.3 在路径函数中声明依赖
生活案例：你写好了“参数提取器”这个通用模板，现在要在具体的接口里使用它。就像你去餐厅，不需要进厨房自己做，只要对服务员说“来一份 A 套餐”，服务员就会把准备好的饭菜端上来。Depends(common_parameters) 就是你点的“套餐名”。
完整可运行代码：
from fastapi import FastAPI, Depends
from typing import Optional
app = FastAPI(title="依赖注入演示")
# ---------- 定义依赖函数 ----------
async def common_parameters(
q: Optional[str] = None, # 搜索关键词，可选
page: int = 1 # 页码，默认1
):
"""
通用的查询参数提取器。
可以被多个接口复用。
"""
return {
"query": q,
"page": page
}
# ---------- 接口A：使用依赖 ----------
@app.get("/items")
async def read_items(
commons: dict = Depends(common_parameters) # ← 注入依赖
):
"""
commons 就是 common_parameters 函数的返回值。
这个接口无需自己处理 q 和 page 参数，直接用现成的。
"""
# commons 里已经包含了 {"query": q, "page": page}
query = commons.get("query")
page = commons.get("page")
return {
"message": f"正在搜索：{query or '全部商品'}",
"page": page,
"commons": commons # 展示完整的依赖返回值
}
# ---------- 接口B：复用同一个依赖 ----------
@app.get("/users")
async def read_users(
params: dict = Depends(common_parameters) # 变量名可以随便取
):
"""
另一个接口也可以复用同一个依赖函数，无需重新写参数提取逻辑。
"""
return {
"message": f"正在搜索用户：{params.get('query') or '全部用户'}",
"page": params.get("page")
}
# ---------- 启动入口 ----------
if __name__ == "__main__":
import uvicorn
uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)代码逐行解释：
|
代码 |
含义 |
|---|---|
|
|
告诉 FastAPI：“执行这个路径函数之前，先调用 |
|
|
自动从请求的查询参数中提取（就像直接在路径函数里声明一样） |
|
|
只是把依赖函数的返回值注入到路径函数中 |
测试：
（1）启动服务：
uvicorn main:app --reload（2）访问 http://127.0.0.1:8000/items?q=手机&page=2，返回：
{
"message": "正在搜索：手机",
"page": 2,
"commons": {
"query": "手机",
"page": 2
}
}（3）访问 http://127.0.0.1:8000/items（不传任何参数），返回：
{
"message": "正在搜索：全部商品",
"page": 1,
"commons": {
"query": null,
"page": 1
}
}关键理解：
-
依赖函数先于路径函数执行。
-
依赖函数可以有独立的参数校验、默认值逻辑。
-
一个依赖函数可以被多个接口复用，一处修改，处处生效。
-
Depends()是FastAPI依赖注入的核心，后续的数据库会话、用户认证、权限校验，全基于它。
补充：依赖注入 vs 手动调用
# ❌ 传统写法：每个接口都重复写一遍参数提取
@app.get("/items")
async def read_items(q: str = None, page: int = 1):
# 这里还得手动处理默认值逻辑
...
# ✅ 依赖注入写法：提取逻辑抽离到依赖函数，接口内干净简洁
@app.get("/items")
async def read_items(commons: dict = Depends(common_parameters)):
# 直接使用 commons，省心
...生活类比总结：
-
传统写法 = 每次去餐厅，你自己进厨房洗菜、切菜、炒菜（每个接口里写重复代码）。
-
依赖注入 = 你只需要点菜，后厨统一备料、统一烹饪，服务员端到你面前（
Depends自动调用依赖函数，把结果送到你手边）。
5.2 常用依赖场景
上一节我们学会了依赖注入的基本概念：Depends 像一个服务员，把准备好的东西端到你的路径函数里。在实际项目中，最常用的“东西”有三样：数据库连接、当前登录用户、操作权限。这三个场景几乎每个接口都会遇到，用依赖注入实现可以大幅减少重复代码。
5.2.1 数据库会话依赖
生活案例：你去图书馆借书。每借一本书，管理员都会从抽屉里拿出一张借阅登记表，记录完毕后立刻把登记表放回抽屉（而不是一直攥在手里）。如果管理员忘记放回，后面的同事就找不到登记表了。
在 FastAPI 中，每次请求处理的“借阅登记表”就是数据库会话（Session）。我们需要在请求开始时获取一个会话，请求结束时务必关闭归还。Depends 配合 yield 正好能实现这个“自动归还”机制。
代码实现（模拟数据库会话）：
from fastapi import FastAPI, Depends
from typing import List
app = FastAPI(title="数据库会话依赖演示")
# ---------- 模拟数据库会话 ----------
class FakeSession:
"""模拟一个数据库会话对象，真实项目中是 SQLAlchemy 的 Session"""
def __init__(self):
self.is_closed = False
print("数据库会话已创建")
def query(self, model):
"""模拟查询所有用户"""
print("执行数据库查询...")
return [
{"id": 1, "name": "张三"},
{"id": 2, "name": "李四"}
]
def close(self):
self.is_closed = True
print("数据库会话已关闭")
# ---------- 依赖函数：获取数据库会话 ----------
def get_db():
"""
使用 yield 的依赖函数：
- yield 之前的代码在请求开始时执行（打开会话）
- yield 之后的代码在请求结束后执行（关闭会话）
"""
db = FakeSession() # ① 开启会话
try:
yield db # ② 将会话交给路径函数使用
finally:
db.close() # ③ 无论请求成功或失败，最终都会关闭会话
# ---------- 接口：使用数据库会话 ----------
@app.get("/users")
def get_users(db: FakeSession = Depends(get_db)):
"""
路径函数中的 db 参数，就是 get_db() 中 yield 出来的那个会话对象。
请求结束后，会自动执行 finally 块中的 db.close()。
"""
users = db.query(User) # 使用会话查询数据
return {"users": users}代码关键点：
-
yield将函数一分为二：前面的代码是“准备工作”，后面的finally是“清理工作”。 -
路径函数拿到
yield出来的db，用完后框架会自动继续执行yield之后的清理代码。 -
即使路径函数内部抛出异常，
finally块仍然会执行，确保连接不泄露。
生活类比：yield 就像自动门——你进来时门打开，你出去后门自动关上。
5.2.2 用户认证依赖
生活案例：你每天进入公司大门，需要刷一下工卡。门禁系统会读取卡里的员工编号，去数据库查询你的信息（姓名、部门、职级），然后决定是否开门。FastAPI 中的认证依赖就是这个“门禁系统”：从请求头中提取 token（工卡），验证后返回当前用户对象，后续接口直接用。
JWT 是什么？
JWT（JSON Web Token） 就像你去银行办理业务时，柜员给你的一张防伪密封信封。信封正面写着你本次要办的事项（比如“转账 1000 元”），信封口盖着银行的防伪章（数字签名）。你拿着这个信封在银行内部走，任何窗口的工作人员一看防伪章完好无损，就知道这封介绍信确实是我行签发的，内容也未被篡改过，于是直接按信封里的指令办理，不需要再打电话回总行验证你的身份。JWT 由三部分组成——头部（信封规格）、载荷（你写的业务内容）、签名（防伪章），三者用点号连接，最终形成一个字符串，在客户端和服务器之间安全、无状态地传递信任。
代码实现（模拟 JWT 认证）：
from fastapi import FastAPI, Depends, Header, HTTPException
app = FastAPI(title="用户认证依赖演示")
# ---------- 模拟用户数据库 ----------
fake_users = {
"token123": {"id": 1, "username": "张三", "role": "admin"},
"token456": {"id": 2, "username": "李四", "role": "user"},
}
# ---------- 依赖函数：获取当前用户 ----------
def get_current_user(authorization: str = Header(...)):
"""
从请求头 Authorization 中提取 token，验证并返回用户对象。
如果 token 无效，直接抛 401 错误，路径函数不会执行。
"""
# 通常 token 格式为 "Bearer xxx"，这里简化为直接传 token 值
token = authorization.replace("Bearer ", "")
user = fake_users.get(token)
if not user:
raise HTTPException(status_code=401, detail="身份验证失败，请重新登录")
return user
# ---------- 接口：查看个人资料 ----------
@app.get("/me")
def read_me(current_user: dict = Depends(get_current_user)):
"""
这个接口只有登录用户才能访问。
current_user 就是依赖函数返回的用户字典。
"""
return {
"message": f"欢迎你，{current_user['username']}！",
"user": current_user
}
# ---------- 接口：查看用户列表（也需要登录） ----------
@app.get("/users")
def list_users(current_user: dict = Depends(get_current_user)):
"""任何需要登录的接口，只需声明 Depends(get_current_user) 即可"""
return {
"message": f"{current_user['username']} 正在查看用户列表",
"users": list(fake_users.values())
}测试（使用 curl 或 Swagger）：
-
不带 token 访问
/me：返回 422（缺少 Header）或 401。 -
带正确 token 访问：
curl -H "Authorization: Bearer token789" http://127.0.0.1:8000/me
curl -H "Authorization: Bearer token123" http://127.0.0.1:8000/me返回：
{"message":"欢迎你，张三！","user":{"id":1,"username":"张三","role":"admin"}}关键点：
-
认证逻辑集中在
get_current_user中，任何接口只需加一行Depends(get_current_user)即可保护。 -
依赖函数抛出
HTTPException会直接中断请求，路径函数不会执行，实现“拦住没带工卡的人”。
5.2.3 权限校验依赖
生活案例：公司里有普通员工和经理。普通员工可以进入公共会议室，但“高层战略会议室”只有经理及以上级别才能预约。权限校验依赖就像会议室门口的保安 — 他先看你的工卡（认证），再看你的职级（权限），不满足就直接拦住。
权限依赖通常会嵌套调用认证依赖，因为先知道“你是谁”，才能判断“你能做什么”。
代码实现：
from fastapi import FastAPI, Depends, HTTPException
app = FastAPI(title="权限校验依赖演示")
# ---------- 模拟用户数据（复用之前的） ----------
fake_users = {
"token123": {"id": 1, "username": "张三", "role": "admin"},
"token456": {"id": 2, "username": "李四", "role": "user"},
}
# ---------- 认证依赖（同上） ----------
def get_current_user(authorization: str = Header(...)):
token = authorization.replace("Bearer ", "")
user = fake_users.get(token)
if not user:
raise HTTPException(status_code=401, detail="请先登录")
return user
# ---------- 权限依赖：仅允许管理员 ----------
def require_admin(current_user: dict = Depends(get_current_user)):
"""
在认证的基础上，进一步校验角色。
如果角色不是 admin，直接返回 403 禁止访问。
"""
if current_user["role"] != "admin":
raise HTTPException(status_code=403, detail="权限不足，需要管理员权限")
return current_user # 通过后可以把用户信息继续向下传递
# ---------- 普通接口：登录即可访问 ----------
@app.get("/profile")
def get_profile(current_user: dict = Depends(get_current_user)):
return {"msg": f"欢迎，{current_user['username']}"}
# ---------- 管理员专用接口 ----------
@app.get("/admin/dashboard")
def admin_dashboard(admin_user: dict = Depends(require_admin)):
"""
这个接口使用了 require_admin 依赖，
它内部又依赖 get_current_user，从而形成依赖链：
① 先验证身份 → ② 再检查权限 → ③ 最后执行路径函数。
"""
return {
"msg": f"管理员 {admin_user['username']}，欢迎进入后台",
"secret_data": "所有用户的敏感信息..."
}依赖链执行顺序：
（1）require_admin 被调用。
（2）require_admin 依赖 get_current_user，于是先执行 get_current_user 获取当前用户。
（3）get_current_user 通过 token 验证，返回用户字典。
（4）require_admin 拿到用户字典，检查 role。
（5）如果角色是 admin，将用户字典传给路径函数；否则抛出 403。
测试：
-
使用
token123（管理员）访问/admin/dashboard：成功。 -
使用
token456（普通用户）访问：返回403 - 权限不足。
curl -X 'GET' 'http://127.0.0.1:8000/admin/dashboard' -H 'Authorization: Bearer token123'
curl -X 'GET' 'http://127.0.0.1:8000/admin/dashboard' -H 'Authorization: Bearer token456'
curl -X 'GET' 'http://127.0.0.1:8000/admin/dashboard' -H 'Authorization: Bearer token789'三大场景总结：
|
场景 |
依赖函数特点 |
生活类比 |
|---|---|---|
|
数据库会话 |
使用 |
图书馆登记表用完后归还原位 |
|
用户认证 |
从请求头提取 token，返回用户对象 |
公司门禁刷工卡 |
|
权限校验 |
嵌套认证依赖，加角色判断，失败抛异常 |
重要会议室门口的保安查职级 |
通过依赖注入，这些通用逻辑被封装成一个个可复用的“积木块”。任何接口想拥有这些能力，只需一行 Depends(xxx)，代码简洁且易于维护。
5.3 依赖嵌套：让依赖层层协作
你已经学会了用 Depends 把数据库会话、用户认证等通用逻辑抽成一个个独立的依赖函数。但在真实项目中，这些依赖往往不是孤立的 — 比如“获取当前用户”这个操作，它自己就需要先拿到数据库会话，然后用会话去查用户表。这就形成了依赖嵌套：一个依赖调用另一个依赖，层层递进，最终把结果传递给路径函数。
5.3.1 一个依赖调用另一个依赖
生活案例：
你去公司人事部开在职证明。整个流程是：
（1）前台先核实你的身份（认证依赖）
（2）身份确认后，她从抽屉里拿出公司内部系统登录账号（数据库会话依赖）
（3）登录系统，输入你的工号，查出你的姓名、部门、入职日期（获取用户信息）
（4）最后把信息填到在职证明模板里，盖章交给你（路径函数返回）
注意，步骤 3 依赖步骤 2 的“系统账号”，步骤 1 又是整个流程的前提。在 FastAPI 中，这就是一个依赖函数在自己的参数里用 Depends 调用另一个依赖函数。
代码实现：
from fastapi import FastAPI, Depends, Header, HTTPException
app = FastAPI(title="依赖嵌套演示")
# ---------- 模拟数据库 ----------
fake_users_db = {
1: {"id": 1, "name": "张三", "role": "admin"},
2: {"id": 2, "name": "李四", "role": "user"},
}
# ---------- 模拟令牌映射 ----------
token_to_user_id = {
"token123": 1,
"token456": 2,
}
# ---------- 底层依赖：获取数据库会话 ----------
def get_db():
"""
最底层的依赖函数。
返回一个模拟的数据库会话对象。
"""
print("→ 执行 get_db：打开数据库会话")
return {"connection": "fake_db_connection"} # 模拟数据库连接
# ---------- 上层依赖：获取当前用户（依赖 get_db） ----------
def get_current_user(
authorization: str = Header(...), # 先从请求头提取 token
db: dict = Depends(get_db) # 然后调用 get_db 拿到数据库会话
):
"""
这个依赖函数自己也需要依赖 get_db()。
FastAPI 会先执行 get_db()，再把返回值传给 db 参数。
"""
print("→ 执行 get_current_user：校验 token 并查询用户")
# 1. 清理 token
token = authorization.replace("Bearer ", "")
# 2. 根据 token 找到用户 ID
user_id = token_to_user_id.get(token)
if not user_id:
raise HTTPException(status_code=401, detail="无效的令牌")
# 3. 使用 db（来自 get_db 的返回值）模拟查询用户
print(f" 使用数据库会话 {db['connection']} 查询用户 ID={user_id}")
user = fake_users_db.get(user_id)
if not user:
raise HTTPException(status_code=401, detail="用户不存在")
return user # 返回给路径函数或下一个依赖
# ---------- 路径函数：直接使用最上层依赖 ----------
@app.get("/me")
def read_me(current_user: dict = Depends(get_current_user)):
"""
路径函数只声明 get_current_user 这一个依赖，
但 get_current_user 内部又依赖 get_db，
FastAPI 会自动解析这条依赖链，按顺序执行。
"""
print("→ 执行路径函数 read_me")
return {
"message": f"欢迎你，{current_user['name']}！",
"role": current_user['role']
}
# ---------- 启动入口 ----------
if __name__ == "__main__":
import uvicorn
uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)关键点：
-
get_current_user的db参数使用了Depends(get_db)，这就是依赖嵌套。 -
FastAPI 会自动识别：
get_current_user需要get_db的返回值，于是先执行get_db。 -
路径函数
read_me只声明了Depends(get_current_user)，但间接也得到了get_db的支持。
在终端运行如下的命令来进行测试：
curl -X 'GET' 'http://127.0.0.1:8000/me' -H 'Authorization: Bearer token123'5.3.2 依赖链的执行顺序
生活案例：多米诺骨牌 — 你推倒第一块（请求进来），它会撞倒第二块，第二块撞倒第三块，最后一块撞响铃铛（返回响应）。顺序必须是 从最外层依赖 → 逐步向内 → 最终到路径函数。
在上面的代码中，执行顺序如下：
请求到达 FastAPI
│
▼
① get_db()
│ 返回 db 对象
▼
② get_current_user(db=Depends(get_db), authorization=Header(...))
│ 拿到 db，用 token 查出用户
│ 返回 user 对象
▼
③ read_me(current_user=Depends(get_current_user))
│ 拿到 user，返回欢迎信息
▼
响应返回客户端控制台输出验证（访问 /me 时）：
→ 执行 get_db：打开数据库会话
→ 执行 get_current_user：校验 token 并查询用户
使用数据库会话 fake_db_connection 查询用户 ID=1
→ 执行路径函数 read_me规则总结：
-
从路径函数的依赖开始，逐层向下解析。
-
最底层（不再依赖别人）的函数先执行。
-
返回值逐层向上传递，最终送到路径函数。
5.3.3 传递依赖返回值
生活案例：接力赛跑。第一棒选手（get_db）把接力棒（数据库会话）传给第二棒（get_current_user），第二棒用接力棒完成自己的工作后，把自己那一段的成绩（用户对象）传给终点线的冲刺者（路径函数）。每一棒只负责自己的任务，交接时把手里的成果递出去。
关键代码回顾：
def get_db():
return {"connection": "fake_db"} # ① 返回 db 对象
def get_current_user(db: dict = Depends(get_db)): # ② 接收 get_db 的返回值
user = fake_users_db.get(user_id)
return user # ③ 返回 user 对象
@app.get("/me")
def read_me(current_user: dict = Depends(get_current_user)): # ④ 接收 get_current_user 的返回值
return {"name": current_user['name']} # ⑤ 使用 user 对象传递过程：
-
get_db()的返回值 → 自动注入到get_current_user的db参数 -
get_current_user()的返回值 → 自动注入到read_me的current_user参数
这一切都是 FastAPI 依赖注入系统自动完成的，不需要你手动调用任何函数或传递参数。
依赖嵌套 vs 手动嵌套
# ❌ 如果不用依赖注入，你需要手动写：
@app.get("/me")
def read_me(authorization: str = Header(...)):
db = get_db()
user = get_current_user(db, authorization)
...
# ✅ 用依赖注入，一切自动：
@app.get("/me")
def read_me(current_user: dict = Depends(get_current_user)):
...核心优势：
-
每个函数只关注自己的职责，不关心上游函数的调用细节。
-
更换底层实现（比如从 MySQL 换成 PostgreSQL），只需修改
get_db，其他层不受影响。 -
依赖关系清晰可见，方便测试和维护。
一句话总结：依赖嵌套就像公司审批流程——你填好申请单提交给上级，上级审核后交给他的上级，层层传递，每层只做自己该做的事，最终把审批结果反馈给你。FastAPI 的 Depends 自动帮你管理这条传递链，你只需声明“我需要什么”，不必关心“怎么拿到它”。
5.4 带资源清理的依赖
前面我们学的依赖函数返回一个值就结束了，但在实际开发中，有些资源用完后必须归还 — 比如数据库连接、文件句柄、网络连接。如果只借不还，服务器迟早会资源耗尽而崩溃。FastAPI 提供了 yield 机制，让依赖函数在完成任务后自动执行清理工作，就像“用完自动归还”。
5.4.1 yield 的用法
生活案例：
你去图书馆的自习室学习，流程是这样的：
（1）到前台：用学生证换取自习室钥匙（打开资源）
（2）使用自习室：进去学习两小时（使用资源）
（3）离开时：把钥匙还给前台（关闭资源）
如果某个同学只拿钥匙不还，后面的同学就没法用了。yield 就是这个“自动归还”机制 — 借的时候自动登记，用完了自动归还，根本不用你操心。
代码实现（模拟数据库会话的完整生命周期）：
from fastapi import FastAPI, Depends
from datetime import datetime
app = FastAPI(title="带资源清理的依赖演示")
# ---------- 模拟数据库会话类 ----------
class DatabaseSession:
"""模拟一个数据库连接会话"""
def __init__(self):
self.id = id(self) # 用内存地址作为会话ID
self.created_at = datetime.now()
print(f"🔓 会话 {self.id} 已创建（耗时操作：连接数据库）")
def query(self, sql: str):
"""模拟执行 SQL 查询"""
print(f" 📊 会话 {self.id} 执行查询：{sql}")
return [{"id": 1, "name": "张三"}]
def close(self):
"""模拟关闭会话"""
print(f"🔒 会话 {self.id} 已关闭（释放连接回连接池）")
# ---------- 不使用 yield 的错误示范 ----------
def get_db_without_yield():
"""
问题：直接返回会话，没有机会关闭它。
请求结束后，会话对象还留在内存里，没人管。
"""
db = DatabaseSession()
return db # 返回后就无法再执行清理代码了
# ---------- 使用 yield 的正确做法 ----------
def get_db():
"""
yield 将函数一分为二：
- yield 之前：资源的“借出”阶段
- yield 之后：资源的“归还”阶段
无论请求成功还是失败，yield 之后的代码都会执行。
"""
print("--- 请求开始 ---")
db = DatabaseSession() # ① 借：创建会话
try:
yield db # ② 用：交给路径函数使用
finally:
db.close() # ③ 还：无论如何都关闭会话
print("--- 请求结束 ---")
# ---------- 接口：正确使用带清理的依赖 ----------
@app.get("/users")
def get_users(db: DatabaseSession = Depends(get_db)):
"""
路径函数拿到 db 对象使用。
函数执行完毕后，get_db 中 yield 之后的代码自动运行。
"""
print(" ↳ 进入路径函数，开始处理业务逻辑")
users = db.query("SELECT * FROM users")
print(" ↳ 业务逻辑处理完毕，准备返回响应")
return {"users": users}
# ---------- 接口：模拟处理过程中发生异常 ----------
@app.get("/users/error")
def get_users_with_error(db: DatabaseSession = Depends(get_db)):
"""
即使路径函数内部抛出异常，finally 块仍然会执行，
确保数据库连接被正确关闭。
"""
print(" ↳ 进入路径函数，即将抛出异常")
db.query("SELECT * FROM users")
raise ValueError("模拟一个业务异常！")
# 这行不会执行，但 get_db 的 finally 块会执行启动后访问 /users，控制台输出：
--- 请求开始 ---
🔓 会话 123456789 已创建（耗时操作：连接数据库）
↳ 进入路径函数，开始处理业务逻辑
📊 会话 123456789 执行查询：SELECT * FROM users
↳ 业务逻辑处理完毕，准备返回响应
🔒 会话 123456789 已关闭（释放连接回连接池）
--- 请求结束 ---访问 /users/error 触发异常，控制台输出：
--- 请求开始 ---
🔓 会话 123456790 已创建（耗时操作：连接数据库）
↳ 进入路径函数，即将抛出异常
📊 会话 123456790 执行查询：SELECT * FROM users
🔒 会话 123456790 已关闭（释放连接回连接池） ← 即使异常，依然关闭！
--- 请求结束 ---
ERROR: ... ValueError: 模拟一个业务异常！可以看到，无论路径函数正常返回还是抛出异常，finally 块里的 db.close() 都执行了。
5.4.2 自动关闭数据库连接
生活案例：餐厅里的餐具回收流程。客人吃完饭（请求结束），不管吃干净还是剩了半碗（成功或异常），服务员都会把餐具收回厨房清洗。yield 的 finally 块就是那个服务员——绝不落下一副餐具。
FastAPI 的自动清理机制：
def get_db():
db = SessionLocal() # 借餐具
try:
yield db # 客人用餐
finally:
db.close() # 收回餐具（无论如何都执行）请求的生命周期：
请求到达 → 执行 yield 之前的代码（借）→ 路径函数执行（用）→ 执行 yield 之后的代码（还）→ 响应返回如果没有 yield，你只能这样手动管理：
@app.get("/users")
def get_users():
db = SessionLocal()
try:
users = db.query(User).all()
return {"users": users}
finally:
db.close()每个接口都要写 try/finally，重复且容易遗漏。有了 Depends + yield，这份清理代码只需写一次。
5.4.3 异常处理与清理
生活案例：
你去酒店入住，交了押金拿房卡。退房时有两种情况：
-
正常退房：前台检查房间无损坏，退还押金。
-
异常退房：你把电视砸了（异常），前台照扣赔偿金，但房卡你总得还回去——你不能说“我砸了电视就不还房卡了”。
try/finally 的逻辑一模一样：finally 块中的“归还房卡”操作不受前面“赔偿”逻辑影响。
异常处理与清理的完整示例：
from fastapi import FastAPI, Depends, HTTPException
app = FastAPI()
class FileManager:
"""模拟文件操作，需要手动关闭"""
def __init__(self, filename: str):
self.filename = filename
print(f"📂 打开文件：{filename}")
def read_content(self):
return f"文件 {self.filename} 的内容"
def close(self):
print(f"📂 关闭文件：{self.filename}")
def get_file_handler():
"""
模拟获取文件句柄。
无论业务逻辑成功还是失败，都要关闭文件。
"""
print("--- 请求开始 ---")
handler = FileManager("data.txt")
try:
yield handler # 正常路径
except Exception as e: # 可选的：捕获异常并处理
print(f"⚠️ 依赖层捕获到异常：{e}")
raise # 重新抛出，让全局异常处理器或客户端感知
finally:
handler.close() # 无论如何都关闭
print("--- 请求结束 ---")
@app.get("/file")
def read_file(fm: FileManager = Depends(get_file_handler)):
"""正常读取文件"""
content = fm.read_content()
return {"content": content}
@app.get("/file/error")
def read_file_error(fm: FileManager = Depends(get_file_handler)):
"""模拟读取文件时出错"""
fm.read_content()
raise HTTPException(status_code=500, detail="读取文件时发生致命错误")三种执行结果对比：
|
场景 |
|
路径函数 |
|
|---|---|---|---|
|
正常请求 |
✅ 执行 |
✅ 执行 |
✅ 执行清理 |
|
路径函数抛异常 |
✅ 执行 |
❌ 中断 |
✅ 执行清理 |
|
依赖函数抛异常 |
❌ 中断 |
❌ 不执行 |
✅ 执行清理 |
核心规则：finally 块中的代码永远执行，这是 Python 语言的保证，和 FastAPI 无关。
完整总结
|
概念 |
生活类比 |
关键代码 |
|---|---|---|
|
|
借钥匙/取餐具 |
|
|
|
把东西给你用 |
|
|
|
还钥匙/回收餐具 |
|
|
异常时 |
砸了电视也要还房卡 |
异常发生后 |
一句话记住：yield 让依赖函数拥有“善后”能力，资源借走必还，即使天塌下来（异常），finally 也能兜底。
5.5 全局与路由级依赖
前面我们学到的依赖都是在单个接口上添加的。但在真实项目中，经常会遇到这样的需求：某个模块下所有接口都需要登录，或者整个网站都要检查是否携带了特定的请求头。这时如果每个接口都单独写一行 Depends(...) 就太麻烦了。FastAPI 提供了两种“批处理”方式：路由级依赖和全局依赖。此外，有些依赖不是强制的 — 比如“如果用户登录了更好，没登录也能用”，这就涉及可选依赖。
5.5.1 路由级依赖：整个模块统一验证
生活案例：公司新租了一层写字楼，该层有会议室、茶水间、办公区三扇门。以前是每扇门都配一名保安单独检查工卡，现在改为在楼层电梯口统一设一个刷卡闸机。只要过了闸机，里面的三扇门随便进。这个“电梯口的闸机”就是路由级依赖 — 对整个模块下的所有路径统一生效。
代码实现：
from fastapi import FastAPI, APIRouter, Depends, HTTPException, Header
app = FastAPI(title="路由级依赖演示")
# ---------- 依赖函数：验证登录 ----------
def get_current_user(authorization: str = Header(...)):
"""简化认证：从 Header 提取 token，返回用户信息"""
token = authorization.replace("Bearer ", "")
# 模拟数据库
users = {"token123": {"id": 1, "name": "张三"}, "token456": {"id": 2, "name": "李四"}}
user = users.get(token)
if not user:
raise HTTPException(status_code=401, detail="未登录或 token 无效")
return user
# ---------- 创建一个路由分组（相当于“楼层”） ----------
# 在该路由上统一添加依赖：所有接口都需要登录
user_router = APIRouter(
prefix="/users", # 该组所有接口的 URL 前缀
tags=["用户模块"],
dependencies=[Depends(get_current_user)] # 关键：统一注入依赖
)
# ---------- 路由下的接口（无需再单独写 Depends） ----------
@user_router.get("/me")
def read_me(current_user: dict = Depends(get_current_user)):
"""查看个人信息"""
return {"msg": f"欢迎你，{current_user['name']}"}
@user_router.get("/settings")
def read_settings():
"""查看设置页面"""
# 注意：这里没有显式声明 Depends(get_current_user)，
# 但由于路由级依赖的存在，依然会先验证登录，未登录则直接 401
return {"theme": "dark", "language": "zh-CN"}
@user_router.post("/logout")
def logout():
return {"msg": "已退出登录"}
# 将路由注册到主应用上
app.include_router(user_router)
# ---------- 另一个路由分组：无需登录 ----------
public_router = APIRouter(prefix="/public", tags=["公开接口"])
@public_router.get("/news")
def public_news():
return {"news": "今日天气晴"}
app.include_router(public_router)效果验证：
-
访问
/users/me不带Authorization头：返回 422 校验错误（缺少 Header）。 -
访问
/users/settings不带 token：同样被拦截，返回 401。 -
访问
/public/news不带 token：正常返回，因为该路由没有添加依赖。
在终端输入如下的命令：
curl -X 'GET' 'http://127.0.0.1:8000/users/me' -H 'Authorization: Bearer token123'
curl -X 'GET' 'http://127.0.0.1:8000/users/settings' -H 'Authorization: Bearer token123'
curl -X 'GET' 'http://127.0.0.1:8000/users/logout' -H 'Authorization: Bearer token123'
curl -X 'GET' 'http://127.0.0.1:8000/public/news' -H 'Authorization: Bearer token123'关键点：
-
APIRouter的dependencies参数接受一个列表，可以传入多个依赖。 -
路由级依赖对该 Router 下所有接口生效，无需逐个添加。
-
如果某个接口还需要额外的依赖（如管理员权限），可以继续在自己的参数中声明
Depends(require_admin)，会按先路由级依赖、再接口级依赖的顺序执行
5.5.2 全局依赖 vs 中间件
有时我们需要在整个应用级别施加控制，比如为每个响应自动添加 X-Request-ID 头，或者记录所有请求的耗时日志。FastAPI 提供了两种机制：全局依赖 和 中间件。它们都能作用于“全局”，但职责不同。
|
维度 |
全局依赖 |
中间件 |
|---|---|---|
|
生效范围 |
仅对路径函数生效（接口层面） |
对所有请求生效，包括静态文件、不存在的路径 |
|
能传递数据 |
✅ 可以将依赖返回值注入路径函数的参数 |
❌ 不能直接给路径函数传参，只能修改 request/response |
|
典型用途 |
全局认证、全局数据库会话、全局配置对象 |
日志记录、计时、跨域处理、请求头修改、压缩 |
生活案例：
-
全局依赖就像公司每间办公室门禁上的工牌读卡器——你刷卡后，读卡器会把你的员工信息（姓名、部门）显示在门口的屏幕上，里面的同事看到屏幕就知道谁来了。它不但能“拦截”无卡人员，还能把身份信息传递进去。
-
中间件就像大楼入口的安检门——每个人都要过，但它只负责检查你是否携带违禁品（修改或拦截请求），并不会告诉里面的办公室你是谁，也无法传递额外数据。
全局依赖示例：为每个接口注入当前时间戳
from fastapi import FastAPI, Depends
from datetime import datetime
app = FastAPI()
# 定义一个全局依赖函数
def add_timestamp():
return datetime.now().isoformat()
# 将依赖绑定到 FastAPI 应用实例上（全局依赖）
app.dependency_overrides = {} # 确保环境干净
app.add_middleware = None # 这行无关，我们通过参数注入
# 正确的全局依赖方式（FastAPI 的 dependencie 参数）
app = FastAPI(dependencies=[Depends(add_timestamp)])
@app.get("/time")
def get_time(ts: str = Depends(add_timestamp)):
return {"current_time": ts}实际代码中更常见的是在 FastAPI() 构造函数中指定：
app = FastAPI(dependencies=[Depends(verify_api_key)])这样所有接口都会先执行 verify_api_key 函数，并且如果依赖函数有返回值，可以通过 Depends 注入到路径函数的参数中（需要显式声明参数）。
中间件示例：记录每个请求的耗时。
import time
from fastapi import Request
@app.middleware("http")
async def log_time(request: Request, call_next):
start = time.time()
response = await call_next(request)
duration = time.time() - start
response.headers["X-Process-Time"] = str(duration)
return response中间件无法把 duration 直接传给路径函数，只能操作 response。
5.5.3 可选依赖：登录了更好，不登录也行
很多场景下，我们希望“如果用户登录了，就提供个性化数据；没登录，则展示默认内容”，而不是强制登录。这就是可选依赖。
生活案例：
你访问一个购物网站，即使没登录也能浏览商品，但如果你登录了，右上角会显示你的用户名和购物车里的数量。网站不强制要求登录，但它会尝试获取你的身份信息，根据获取结果渲染不同页面。
实现方式：给依赖函数的返回值指定默认值 None，并在路径参数中使用 Depends(optional_func) 配合 default=None。
from fastapi import FastAPI, Depends, Header
from typing import Optional
app = FastAPI()
# 模拟用户数据库
fake_users = {"token123": "张三", "token456": "李四"}
# 依赖函数：可能返回用户信息，也可能返回 None
def get_current_user_optional(
authorization: Optional[str] = Header(None) # Header 本身也是可选的
) -> Optional[str]:
"""尝试从 Authorization 头提取用户名，如果未提供或无效则返回 None"""
if not authorization:
return None
token = authorization.replace("Bearer ", "")
return fake_users.get(token) # 可能返回 None
@app.get("/welcome")
async def welcome(
current_user: Optional[str] = Depends(get_current_user_optional)
):
"""
如果携带了有效的 Authorization 头，则显示个性化欢迎语；
否则显示通用的欢迎语。
"""
if current_user:
return {"message": f"欢迎回来，{current_user}！"}
else:
return {"message": "欢迎光临，游客！"}请求测试：
-
不带
Authorization头访问/welcome：返回{"message": "欢迎光临，游客！"} -
带
Authorization: Bearer token123访问：返回{"message": "欢迎回来，张三！"}
关键点：
-
依赖函数里通过参数默认值或条件判断让它不抛出异常，而是返回
None。 -
路径函数里的参数类型用
Optional[str]并且默认值Depends(...)就会接收依赖的返回值（包括None）。 -
如果依赖函数抛出异常（如
HTTPException(401)），则不再是可选，会直接中断请求。
另一种方式：使用 Depends() 的 default 参数（较少用），通常直接在依赖函数内部处理更灵活。
总结
|
场景 |
实现方法 |
生活类比 |
|---|---|---|
|
某个模块下所有接口统一验证 |
|
楼层电梯口的统一闸机 |
|
整个应用全局注入数据 |
|
每间办公室门禁读卡器（带信息传递） |
|
全局请求预处理（不传数据） |
中间件 |
大楼入口安检门（只检查，不告知是谁） |
|
有则更好，无则也可 |
依赖返回 |
购物网站“登录后显示用户名” |
通过灵活组合路由级依赖、全局依赖和可选依赖，你可以像搭积木一样构建出清晰、安全、可维护的 API 架构。
第 6 章 异常处理与统一响应
在实际项目中，接口不可能永远正常返回。用户可能传入非法参数、数据库可能宕机、外部服务可能超时……如果一遇到问题就甩一个 Python 的原始报错堆栈给客户端，不仅用户体验极差，还可能泄露内部敏感信息。FastAPI 提供了一套优雅的异常处理机制，让你能够主动抛出结构化错误、统一错误格式，甚至定义自己的业务异常。
6.1 主动抛出 HTTPException
当路径函数在执行过程中发现“情况不对”，可以主动用 HTTPException 终止处理流程，立刻向客户端返回一个结构化的错误响应。它就像一个“紧急终止按钮” — 按下去，后面的代码就不执行了。
6.1.1 抛出客户端错误（4xx）
生活案例：你去银行取号排队，如果你拿到的号码是“-5 号”（负数），前台的叫号系统会立刻提示“号码无效，请重新取号”，而不是让你白等一上午。这就是客户端错误 — 问题出在请求方，服务器无能为力，只能告诉你“你发来的东西有问题”。
代码实现：
from fastapi import FastAPI, HTTPException
app = FastAPI(title="异常处理演示")
# ---------- 模拟商品数据 ----------
fake_items = {1: "笔记本电脑", 2: "机械键盘", 3: "降噪耳机"}
@app.get("/items/{id}")
def get_item(id: int):
"""
根据 ID 查询商品。
如果 ID 不合法（≤0）或不存在，主动抛出 4xx 错误。
"""
# 场景一：参数非法
if id <= 0:
raise HTTPException(
status_code=400, # HTTP 400 Bad Request
detail=f"商品 ID 必须大于 0，你传入的是 {id}"
)
# 场景二：资源不存在
item = fake_items.get(id)
if not item:
raise HTTPException(
status_code=404, # HTTP 404 Not Found
detail=f"ID 为 {id} 的商品不存在"
)
# 一切正常，返回商品信息
return {"item_id": id, "name": item}测试结果：
-
访问
http://127.0.0.1:8000/items/0→ 返回 400{"detail": "商品 ID 必须大于 0，你传入的是 0"} -
访问
http://127.0.0.1:8000/items/999→ 返回 404{"detail": "ID 为 999 的商品不存在"} -
访问
http://127.0.0.1:8000/items/1→ 返回 200{"item_id": 1, "name": "笔记本电脑"}
关键点：
-
status_code是你主动指定的 HTTP 状态码。 -
detail是给调用方的错误说明，可以是字符串或字典。 -
一旦抛出
HTTPException，路径函数立刻停止执行，后续代码不会运行。
6.1.2 抛出服务端错误（5xx）
生活案例：你去取快递，快递员打开柜门时发现柜子里的灯坏了，里面一片漆黑无法找到你的包裹。这不是你的问题（你没有写错取件码），而是快递站自己的设备故障。快递员会说“抱歉，系统出了点问题，请稍后再来”。这就是服务端错误 — 请求本身没问题，但服务器这边出故障了。
代码实现：
import random
@app.get("/external-api")
def call_external_service():
"""
模拟调用外部服务（如第三方支付、短信网关）。
当外部服务不可用时，抛出 503 Service Unavailable。
"""
# 模拟：有 30% 的概率外部服务故障
if random.random() < 0.3:
raise HTTPException(
status_code=503, # HTTP 503 Service Unavailable
detail="外部服务暂时不可用，请稍后重试",
headers={"Retry-After": "60"} # 告诉客户端 60 秒后再试
)
return {"msg": "外部服务调用成功", "data": "第三方数据"}常见 5xx 错误及其使用场景：
|
状态码 |
含义 |
什么时候用 |
|---|---|---|
|
500 |
服务器内部错误 |
未预料的代码异常，兜底用 |
|
502 |
网关错误 |
你调了上游服务，上游返回了无效响应 |
|
503 |
服务不可用 |
服务器暂时超载或维护中 |
|
504 |
网关超时 |
上游服务响应太慢，等不及了 |
6.1.3 设置 detail 与 headers
生活案例：
-
detail就像快递站给你的小票，上面写着“包裹未找到的原因”：可能写错取件码（400）、可能包裹还在运输中（404）、可能快递柜故障（503）。 -
headers就像快递员补一句：“您别着急，预计两小时后修好，到时候您再试”（Retry-After: 7200）。这不是给你看的文字，而是给客户端程序读取的指令。
raise HTTPException(
status_code=429, # Too Many Requests
detail="请求太频繁，请慢一点",
headers={
"Retry-After": "120", # 告诉客户端 120 秒后再来
"X-Error-Code": "RATE_LIMIT" # 自定义头，用于内部排查
}
)客户端可以读取 Retry-After 头，自动设置一个定时器，时间到了再重试，而不是无脑疯狂重试加重服务器负担。
6.2 全局异常处理器
如果每个接口都自己写 try/except 来处理异常，代码会极其冗余。更可怕的是，某些未预料的异常（如代码 bug）可能绕过你的 try/except，直接暴露 Python 原始报错信息给客户端。全局异常处理器就是整个应用的“安全兜底网” — 无论哪个接口报了异常，最终都会被它拦截并统一处理。
6.2.1 @app.exception_handler 装饰器
生活案例：一家五星级酒店，无论客人在哪层楼的餐厅、健身房还是客房闹了矛盾（异常），最终都会由同一套“客户投诉处理流程”来应对：首先安抚客人情绪，然后统一登记到系统，最后给出标准化的解决方案（优惠券/升级房型）。这个标准化流程就是全局异常处理器 — 不管问题出在哪儿，最终都按统一格式反馈给客人。
代码实现：
from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse
app = FastAPI()
# ---------- 全局处理器：专门捕获 HTTPException ----------
@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
"""
当任何接口抛出 HTTPException 时，都会走这个处理器。
作用：统一 HTTP 异常的返回格式。
"""
return JSONResponse(
status_code=exc.status_code,
content={
"success": False,
"code": exc.status_code,
"message": exc.detail, # 原本的 detail 信息
"path": request.url.path # 额外附加上出错的接口路径
}
)现在，前面 get_item 接口抛出 HTTPException(status_code=404, detail="ID 为 999 的商品不存在") 时，返回的 JSON 会被全局处理器统一包装成：
{
"success": false,
"code": 404,
"message": "ID 为 999 的商品不存在",
"path": "/items/999"
}6.2.2 捕获 HTTPException 并统一格式
为什么要统一格式？
前端同事对接时，如果有的接口返回 {"detail": "错误"}，有的返回 {"error": "错误"}，有的返回 {"message": "错误"}，他的代码就要写三套解析逻辑。统一格式后，他只需要处理一种结构，大幅降低沟通成本。
6.2.3 捕获未预料的异常
HTTPException 是我们主动抛出的、可预见的错误。但代码里难免有 bug（如 1 / 0），这些异常没有对应的 raise HTTPException(...)，如果不处理，FastAPI 默认会返回原始的 Python 堆栈。在生产环境中，绝不能让客户端看到堆栈信息。
全局兜底处理器：
import traceback
import logging
logger = logging.getLogger(__name__)
@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
"""
捕获所有未被前面处理器截获的异常（如代码 bug、数据库驱动报错等）。
这个处理器必须放在最后，作为“终极兜底”。
"""
# 1. 在服务端详细记录真实错误（供开发人员排查）
logger.error(f"未捕获的异常: {exc}\n{traceback.format_exc()}")
# 2. 返回给客户端的只是模糊提示，绝不泄露内部细节
return JSONResponse(
status_code=500,
content={
"success": False,
"code": 500,
"message": "服务器内部错误，请稍后重试或联系管理员",
"path": request.url.path
}
)两种处理器的分工：
|
处理器 |
捕获对象 |
给客户端的信息 |
内部做什么 |
|---|---|---|---|
|
|
你主动抛出的业务错误 |
具体的错误原因（如“ID 不存在”） |
统一包装格式 |
|
|
代码 bug、未知错误 |
模糊提示（“服务器内部错误”） |
记录详细日志供排查 |
6.3 自定义异常与响应
有时 HTTP 状态码不足以精确表达业务错误。比如“商品不存在”和“用户不存在”都是 404，但业务含义完全不同。此时可以定义自己的异常类，并注册专门的处理器。
6.3.1 创建自定义异常类
生活案例：医院里不同颜色的床头卡代表不同的护理级别 — 红色代表危重、黄色代表需关注、绿色代表普通。护士看到颜色就知道优先级，不需要每次都翻看整本病历。
自定义异常就是给错误贴上不同的“颜色标签”，方便后端按类型分流处理。
# ---------- 自定义业务异常 ----------
class ItemNotFound(Exception):
"""商品不存在"""
def __init__(self, item_id: int):
self.item_id = item_id
self.error_code = "ITEM_NOT_FOUND"
class InsufficientStock(Exception):
"""库存不足"""
def __init__(self, item_id: int, requested: int, available: int):
self.item_id = item_id
self.requested = requested
self.available = available
self.error_code = "INSUFFICIENT_STOCK"
class ForbiddenAction(Exception):
"""权限不足"""
def __init__(self, user: str, action: str):
self.user = user
self.action = action
self.error_code = "FORBIDDEN"6.3.2 为自定义异常注册处理器
# ---------- 为 ItemNotFound 注册专用处理器 ----------
@app.exception_handler(ItemNotFound)
async def item_not_found_handler(request: Request, exc: ItemNotFound):
return JSONResponse(
status_code=404,
content={
"success": False,
"error_code": exc.error_code, # 业务错误码
"message": f"商品 {exc.item_id} 不存在"
}
)
# ---------- 为 InsufficientStock 注册专用处理器 ----------
@app.exception_handler(InsufficientStock)
async def insufficient_stock_handler(request: Request, exc: InsufficientStock):
return JSONResponse(
status_code=409, # 409 Conflict（请求冲突）
content={
"success": False,
"error_code": exc.error_code,
"message": f"商品 {exc.item_id} 库存不足：请求 {exc.requested} 件，实际可用 {exc.available} 件"
}
)在接口中使用：
@app.post("/orders")
def create_order(item_id: int, quantity: int, user: str):
if item_id not in fake_items:
raise ItemNotFound(item_id=item_id) # 自动触发专用处理器
stock = fake_stock.get(item_id, 0)
if quantity > stock:
raise InsufficientStock(
item_id=item_id,
requested=quantity,
available=stock
)
return {"msg": f"{user} 成功下单 {quantity} 件商品 {item_id}"}6.4 HTTP 状态码常量
在代码里直接写 400、404 这些裸数字虽然简单，但可读性差。FastAPI 继承了 Starlette 的 status 模块，提供了语义化的常量。
生活案例：
交通信号灯——看到红灯你立刻知道“停车”，不需要记忆“红色=停止=编号 0x0001”这种底层映射。status.HTTP_404_NOT_FOUND 就是给数字 404 起了一个一眼能看懂的名字。
常用常量速查：
from starlette import status
# 2xx：成功
status.HTTP_200_OK # 200
status.HTTP_201_CREATED # 201
# 3xx：重定向
status.HTTP_301_MOVED_PERMANENTLY # 301
# 4xx：客户端错误
status.HTTP_400_BAD_REQUEST # 400
status.HTTP_401_UNAUTHORIZED # 401
status.HTTP_403_FORBIDDEN # 403
status.HTTP_404_NOT_FOUND # 404
status.HTTP_409_CONFLICT # 409
status.HTTP_422_UNPROCESSABLE_ENTITY # 422（FastAPI 默认校验失败返回的）
# 5xx：服务端错误
status.HTTP_500_INTERNAL_SERVER_ERROR # 500
status.HTTP_502_BAD_GATEWAY # 502
status.HTTP_503_SERVICE_UNAVAILABLE # 503使用对比：
# ❌ 裸数字：两个月后你忘了 409 是什么
raise HTTPException(status_code=409, detail="库存不足")
# ✅ 语义化常量：一眼看出来是“请求冲突”
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="库存不足")完整总结
|
机制 |
用途 |
生活类比 |
|---|---|---|
|
|
主动返回标准 HTTP 错误 |
银行前台告诉你“号无效”或“系统故障” |
|
全局 |
统一所有业务错误的返回格式 |
酒店统一的客户投诉处理流程 |
|
全局 |
兜底处理未知 bug，隐藏内部细节 |
安全兜底网，意外摔下来也不会直接砸地上 |
|
自定义异常 + 处理器 |
精确表达业务错误，不同错误不同处理 |
医院不同颜色的床头卡，护士看颜色就知道轻重缓急 |
|
状态码常量 |
用可读的名字代替裸数字 |
看到红灯就知道“停”，不需要记编号 |
第 7 章 中间件
7.1 理解中间件：请求的“门卫”
在 FastAPI 中，中间件就像公司大厦入口处的智能门禁系统 — 每一个进入大楼的访客（请求）都必须先经过它，离开时（响应）也要再次经过。门卫不关心你去哪个部门办什么事（那是路径函数的事），他只负责安全检查、时间登记、放行包裹这些横切关注点。这一节我们就来拆解这道“门”到底在什么时候开、关，以及怎么用它来给所有接口装一个计时器。
7.1.1 请求-响应生命周期
生活案例：
你每天上班的流程是这样的：
（1）进门时：门卫检查你的工牌，确认你是本公司员工（身份校验），然后用体温枪测一下体温（安全检查），再在本子上记录你的到岗时间（日志记录）。
（2）进入办公区后：你坐到工位上开始处理报销单、回复邮件（路径函数做业务逻辑）。
（3）下班出门时：同一个门卫再次拦下你，记录离岗时间，检查你是否带出公司财物（响应后处理），然后放行。
在 FastAPI 中，中间件就是这个门卫。它不关心你在工位上做了什么（那是路径函数的事），但它在你进来之前和出去之后各做了一次统一检查。每一个请求都会经历这样一个完整的生命周期。
请求 → [中间件前半段] → [路径函数执行业务] → [中间件后半段] → 响应返回客户端
↑ 进门检查 ↑ 办公 ↑ 出门检查7.1.2 中间件的作用位置
从代码层面看，中间件是一个特殊的函数，它接收两个参数：
-
request: Request：当前请求对象，包含 URL、请求头、Cookie 等所有信息。 -
call_next：一个“放行”函数，调用它就会把请求转给下一个中间件或最终的路由函数。你必须调用它，否则请求就卡在门卫这里了。
中间件函数的执行顺序类似“洋葱模型” — 请求一层层剥进去，响应一层层剥出来：
中间件A 开始 → 中间件B 开始 → 路径函数 → 中间件B 结束 → 中间件A 结束与依赖注入、全局异常处理器的位置区别：
|
机制 |
生效时机 |
能获取什么 |
|---|---|---|
|
中间件 |
请求进入路由之前 / 响应离开之后 |
原始 Request、Response 对象 |
|
全局依赖 |
路由函数执行前（请求已到达路由） |
可以传数据给路径函数 |
|
全局异常处理器 |
路径函数抛出异常时 |
异常对象 |
中间件是最早拦截、最后放行的，适合做日志、计时、压缩、跨域这类与业务逻辑无关的通用处理。
7.1.3 简单计时中间件示例
生活案例：大厦物业想统计每位访客在楼内平均停留多长时间，于是在门禁系统上装了一个计时器 — 刷工牌进门时开始计时，刷工牌出门时结束计时，把停留秒数写在访客的通行小票上。这个“计时器”就是我们要写的中间件。
完整代码实现：
from fastapi import FastAPI, Request
import time
import uvicorn
app = FastAPI(title="中间件计时演示")
# ---------- 定义中间件 ----------
@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
"""
在每个请求处理前开始计时，处理结束后计算耗时，
并将耗时写入响应头 X-Process-Time 中。
"""
# ① 进门：开始计时
start_time = time.time()
print(f"🔵 请求进来：{request.method} {request.url.path}")
# ② 放行：让请求进入路由函数（以及其他中间件）
response = await call_next(request)
# ③ 出门：计算耗时，写入响应头
process_time = time.time() - start_time
response.headers["X-Process-Time"] = str(round(process_time, 4))
print(f"🟢 请求离开：{request.url.path} 耗时 {process_time:.4f} 秒")
return response
# ---------- 模拟业务接口 ----------
@app.get("/fast")
async def fast_endpoint():
"""快速接口：模拟很快返回"""
return {"msg": "这个接口很快"}
@app.get("/slow")
async def slow_endpoint():
"""慢速接口：模拟耗时操作（数据库查询、调用外部API等）"""
time.sleep(2) # 模拟 2 秒的业务处理
return {"msg": "这个接口很慢，模拟了2秒的业务逻辑"}
@app.get("/error")
async def error_endpoint():
"""异常接口：模拟业务出错"""
raise ValueError("模拟业务逻辑中的异常")
# ---------- 启动入口 ----------
if __name__ == "__main__":
uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)运行效果（使用 curl 测试，添加 -i 参数可以看到响应头）：
curl -i http://127.0.0.1:8000/slow返回：
HTTP/1.1 200 OK
x-process-time: 2.0008 ← 中间件自动注入的耗时
content-type: application/json
...
{"msg":"这个接口很慢，模拟了2秒的业务逻辑"}关键代码逐行解释：
|
代码 |
含义 |
|---|---|
|
|
告诉 FastAPI 这是一个 HTTP 中间件 |
|
|
当前请求对象，可以从中获取 URL、Header 等信息 |
|
|
一个异步函数，调用它意味着“放行，去执行下一个中间件或路由” |
|
|
暂停当前中间件，等内部全部处理完，拿到最终的响应对象 |
|
|
在发给客户端之前，给响应偷偷加一个头 |
|
|
必须返回 |
为什么中间件能处理异常情况？
即使路由函数内部抛出了异常（访问 /error），await call_next(request) 依然会返回一个包含错误信息的响应对象，中间件照常执行后续代码、记录耗时、注入响应头。控制台输出：
🔵 请求进来：GET /error
🟢 请求离开：/error 耗时 0.0012 秒总结
-
中间件是请求的“统一门卫”：所有请求进来前和离开后都要经过它，适合做日志、计时、压缩、安全校验等与业务无关的横向处理。
-
它不改变路由逻辑：中间件不关心你访问的
/fast还是/slow，它只负责“进门登记、出门计时”。 -
执行顺序是“洋葱模型”：请求从外向内穿过所有中间件到达路由函数，响应再从内向外穿回。
-
响应的头/体都可以修改：比如我们添加了
X-Process-Time，你也可以在中间件里添加 CORS 头、压缩响应体等。
7.2 内置中间件：FastAPI 的“标准安保套餐”
上一节我们学会了如何自己写一个中间件（像自己造了个门卫）。但日常开发中很多需求是通用的，比如允许前端跨域请求、压缩返回数据节省流量、防止恶意主机名攻击。FastAPI 已经把这三个最常见的需求做成了内置中间件，你只需要“安装”即可，就像买了一套标准安保套餐：跨域通行证、快递压缩打包、大楼主机名安检。
7.2.1 CORSMiddleware 解决跨域
生活案例：你在一家公司工作，公司规定员工只能在自己部门的打印机上打印文件（同源策略：协议+域名+端口都相同才能通信）。但有一天你需要打印一份合同，而彩色打印机在隔壁市场部的办公室，前台直接拒绝你进入。为了解决这个问题，行政部出台了一项新规定：只要你在申请单上注明“跨部门打印”，并且市场部的打印机被列入了允许名单，就可以使用。这个“跨部门打印许可证”就是 CORS。CORSMiddleware 就是那个帮你自动签发许可证的前台接待员。
为什么需要 CORS？
浏览器为了保护用户，默认禁止前端 JavaScript 向不同源（协议、域名、端口任一不同）的后端发送请求。比如你的前端跑在 http://localhost:3000，后端跑在 http://localhost:8000，浏览器就会拦截，提示跨域错误。要让浏览器放行，后端必须在响应头里明确告诉浏览器：“我允许来自 localhost:3000 的请求”。
代码实现：
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
app = FastAPI()
# ---------- 配置允许跨域 ----------
app.add_middleware(
CORSMiddleware,
allow_origins=["http://localhost:3000", "https://myapp.com"], # 允许这些前端域名
allow_credentials=True, # 允许携带 Cookie
allow_methods=["GET", "POST", "PUT", "DELETE"], # 允许的 HTTP 方法
allow_headers=["*"], # 允许的请求头
)关键参数解释：
|
参数 |
作用 |
生产环境建议 |
|---|---|---|
|
|
白名单：哪些前端域名可以调我 |
绝不写 |
|
|
是否允许携带 Cookie/Authorization |
如果设为 |
|
|
允许哪些 HTTP 方法 |
按需列举，如 |
|
|
允许哪些自定义请求头 |
|
测试接口（配合一个简单的前端页面或 curl 查看效果）：
@app.get("/api/data")
def get_data():
return {"message": "跨域请求成功！"}在终端输入如下命令：
curl -X GET "http://localhost:8000/api/data" \
-H "Origin: http://localhost:3000" \
-v当来自 http://localhost:3000/api/data 的前端请求这个接口时，后端响应头会自动带上：
Access-Control-Allow-Origin: http://localhost:3000
Access-Control-Allow-Credentials: true浏览器看到这些头，才会放行并把数据交给前端 JS。
7.2.2 GZipMiddleware 压缩响应
生活案例：你网购了一床羽绒被，商家发货时如果用原包装（巨大的纸箱），快递费会很贵；但商家用了真空压缩袋，把羽绒被压缩成一个小包，体积减少 70%，快递费省了一大半。到你手里后，你拆开压缩袋，羽绒被恢复原样。
GZipMiddleware 就是这个真空压缩袋 — 服务器把响应体压缩后传输，客户端收到后自动解压，既节省带宽又加快传输速度，而且用户完全无感知。
代码实现：
from fastapi import FastAPI
from fastapi.middleware.gzip import GZipMiddleware
app = FastAPI()
# ---------- 添加 GZip 压缩中间件 ----------
app.add_middleware(
GZipMiddleware,
minimum_size=1000 # 只有响应体大于 1000 字节时才压缩（小文件压缩反而浪费 CPU）
)工作原理：
（1）接口返回一个很大的 JSON（比如 50KB）。
（2）响应在离开服务器前，经过 GZipMiddleware，被压缩成约 15KB 的 gzip 数据。
（3）客户端（浏览器、curl）收到压缩数据后，根据响应头 Content-Encoding: gzip 自动解压还原。
什么时候该用？
-
适合：文本类响应（JSON、HTML、CSS、JS），压缩比高，效果显著。
-
不适合：已经压缩过的二进制文件（图片、视频、PDF），再压缩 CPU 白耗。
测试接口：
@app.get("/big-data")
def big_data():
# 模拟一个较大的响应（约 2KB）
large_list = [{"id": i, "name": f"商品{i}", "description": "这是一段很长的描述文本" * 10} for i in range(100)]
return {"data": large_list}用 curl 测试，添加 --compressed 参数（告诉 curl 支持 gzip 解压）：
curl -i --compressed http://127.0.0.1:8000/big-data在响应头中你会看到 Content-Encoding: gzip，而且传输的数据量明显减少了。
7.2.3 TrustedHostMiddleware 限制主机名
生活案例：你开了一家会员制餐厅，只允许预约过的客人用餐。有一天，一个陌生人拿着你的餐厅名片（上面印着你的店名和地址）站在门口，声称“这就是我的店”，企图冒充你进入。门卫查了一下，发现名片上的地址虽然对，但这个人并不在预约名单上，于是直接拦下。
在 HTTP 请求中，Host 头就是这张名片 — 它告诉服务器“用户想访问哪个网站”。恶意攻击者可能会伪造一个不存在的 Host 头（比如 evil.com），试图欺骗你的服务器或利用缓存投毒。TrustedHostMiddleware 就是这个门卫，它只允许放行 Host 头在可信任名单中的请求。
代码实现：
from fastapi import FastAPI
from fastapi.middleware.trustedhost import TrustedHostMiddleware
app = FastAPI()
# ---------- 添加受信任主机中间件 ----------
app.add_middleware(
TrustedHostMiddleware,
allowed_hosts=["localhost", "127.0.0.1", "myapi.com", "*.example.com"]
# 支持通配符 *，如 *.example.com 匹配 api.example.com、admin.example.com
)参数说明：
-
allowed_hosts：可信任的Host头列表，支持：-
精确匹配：
"localhost" -
IP 地址：
"127.0.0.1" -
通配符：
"*.example.com"匹配api.example.com但不匹配example.com本身
-
如果 Host 头不匹配会怎样？
中间件会直接返回 400 Bad Request，请求根本不会进入任何路由函数，从源头杜绝了 Host 头攻击。
测试：
正常请求（Host 匹配）：
curl -H "Host: localhost" http://127.0.0.1:8000/some-api
# 正常返回数据异常请求（Host 不匹配）：
curl -H "Host: evil.com" http://127.0.0.1:8000/some-api
# 返回 400 Bad Request内置中间件汇总
|
中间件 |
作用 |
生活类比 |
关键参数 |
|---|---|---|---|
|
|
允许前端跨域访问 |
跨部门打印许可证 |
|
|
|
压缩响应体，节省带宽 |
真空压缩袋快递 |
|
|
|
防御 Host 头伪造攻击 |
会员制餐厅门卫 |
|
这三个中间件是搭建生产级 API 的“三件套”——先确保安全（限制主机名），再确保互联互通（允许跨域），最后优化性能（压缩数据）。加上前面学的自定义中间件（如计时、日志），你就拥有了完整的中间件武器库。
7.3 自定义中间件
通过编写自定义中间件，开发者可以灵活地实现各种请求处理逻辑，例如：记录每个请求的详细日志（请求方法、路径、耗时）、根据客户端 IP 地址进行访问控制（黑白名单过滤）、添加统一的响应头、或对请求体进行预处理等。这种机制允许在不修改原有业务路由代码的前提下，横向切入到请求处理链路中，极大增强了应用的扩展性和维护性。
7.3.1 案例一：请求日志中间件
记录每个请求的 HTTP 方法、请求路径、处理耗时以及响应状态码。
from fastapi import FastAPI, Request
import time
import logging
app = FastAPI()
logging.basicConfig(level=logging.INFO)
@app.middleware("http")
async def log_requests(request: Request, call_next):
start_time = time.time()
# 处理请求
response = await call_next(request)
# 计算耗时
process_time = time.time() - start_time
# 记录日志
logging.info(
f"{request.method} {request.url.path} "
f"status={response.status_code} "
f"duration={process_time:.4f}s"
)
return response7.3.2 案例二：IP 黑白名单过滤中间件
根据客户端 IP 地址限制访问，只允许白名单内的 IP 访问特定路由。
from fastapi import FastAPI, Request, HTTPException
app = FastAPI()
# 配置白名单 IP 列表
WHITELIST_IPS = ["127.0.0.1", "192.168.1.100"]
@app.middleware("http")
async def ip_whitelist_middleware(request: Request, call_next):
client_ip = request.client.host
if client_ip not in WHITELIST_IPS:
raise HTTPException(status_code=403, detail="IP 未被授权访问")
response = await call_next(request)
return response7.3.3 案例三：简单限流中间件（基于内存）
限制同一 IP 在单位时间内的请求次数。
from fastapi import FastAPI, Request, HTTPException
from collections import defaultdict
import time
app = FastAPI()
# 存储 IP -> 请求时间列表
request_records = defaultdict(list)
RATE_LIMIT = 10 # 最多允许 10 次请求
TIME_WINDOW = 60 # 60 秒内
@app.middleware("http")
async def rate_limit_middleware(request: Request, call_next):
client_ip = request.client.host
now = time.time()
# 清除超过时间窗口的记录
request_records[client_ip] = [t for t in request_records[client_ip] if now - t < TIME_WINDOW]
if len(request_records[client_ip]) >= RATE_LIMIT:
raise HTTPException(status_code=429, detail="请求过于频繁，请稍后再试")
request_records[client_ip].append(now)
response = await call_next(request)
return response7.3.4 案例四：为所有响应添加自定义头
为每个响应统一添加 X-Server 和 X-Request-ID 头。
from fastapi import FastAPI, Request
import uuid
app = FastAPI()
@app.middleware("http")
async def add_custom_headers(request: Request, call_next):
response = await call_next(request)
response.headers["X-Server"] = "FastAPI-Example"
response.headers["X-Request-ID"] = str(uuid.uuid4())
return response7.3.5 案例五：请求体大小限制中间件
限制上传的请求体不超过指定大小（如 1MB）。
from fastapi import FastAPI, Request, HTTPException
app = FastAPI()
MAX_BODY_SIZE = 1024 * 1024 # 1 MB
@app.middleware("http")
async def limit_body_size(request: Request, call_next):
# 仅在 POST/PUT 等可能携带 body 的方法上检查
if request.method in ("POST", "PUT", "PATCH"):
content_length = request.headers.get("content-length")
if content_length and int(content_length) > MAX_BODY_SIZE:
raise HTTPException(status_code=413, detail="请求体过大")
response = await call_next(request)
return response这些案例展示了自定义中间件的强大能力：你可以在请求进入业务逻辑之前或之后插入任何自定义处理，且不污染路由函数的代码，非常适合实现横切关注点（cross-cutting concerns）。
第 8 章 后台任务与异步
在实际的 Web 服务中，不是所有操作都需要“立即”完成。比如用户注册后，我们要发送一封欢迎邮件，但没必要让用户盯着屏幕等邮件发出。再比如，一个接口需要调用第三方 API，对方响应很慢，如果我们的服务器傻等，其他用户请求就会堆积。FastAPI 通过异步编程和后台任务完美解决了这两个问题。本章将用生活化的案例和可运行的代码，带你彻底理解同步/异步、后台任务，以及如何处理遗留的同步阻塞代码。
8.1 同步与异步概念
8.1.1 阻塞与非阻塞的理解
生活案例：快餐店点餐
-
同步（阻塞）模式：你走进一家只有一个服务员的小店，点了一个现做的汉堡。服务员告诉你“请等一下”，然后开始做汉堡。你就站在柜台前，什么也不干，一直等，直到汉堡做好递给你，你才离开。这期间，后面排队的其他顾客也只能干等着，因为唯一的服务员正在为你服务。这就是阻塞——一个任务没完成，后续任务都不能开始。
-
异步（非阻塞）模式：你去一家更聪明的快餐店。点完餐后，服务员给你一个震动号码牌，然后说“您先坐，餐好了牌子会震”。你回到座位继续刷手机、回邮件，服务员不会闲着，立刻为下一位顾客点餐。当你的汉堡做好后，号码牌震动，你再去取餐。这期间，服务员同时服务了多位顾客，每个人也都没浪费时间。这就是异步 — 在等待某个耗时操作时，CPU 或服务器不会傻等，而是去处理其他任务。
在编程世界里，阻塞操作通常是：读写文件、网络请求、数据库查询等 I/O 操作。如果采用同步方式，一个请求在处理文件时，整个程序都会卡住。而异步方式允许程序在等待 I/O 时切换到其他任务，大大提升并发能力。
8.1.2 Python 中的 async / await
Python 从 3.5 版本开始引入了 async/await 语法，用于编写异步代码。
-
async def：定义一个协程函数（coroutine）。这个函数内部可以包含await表达式，而且它不会像普通函数一样直接执行，调用它时会返回一个协程对象，需要放入事件循环中运行。 -
await：后面跟一个可等待对象（如另一个协程、asyncio.sleep()、异步数据库查询等）。它表示：“我先暂停在这里，去做别的事，等这个可等待对象完成后，再回来继续执行”。
简单示例：
import asyncio
async def make_coffee():
print("开始制作咖啡...")
await asyncio.sleep(2) # 模拟耗时操作，如磨豆、加热
print("咖啡制作完成")
return "一杯拿铁"
async def main():
# 执行协程
coffee = await make_coffee()
print(f"拿到了：{coffee}")
# 运行事件循环
asyncio.run(main())在终端输入如下命令运行上述代码：
python main.pyasyncio.sleep(2) 模拟了一个非阻塞的等待，在此期间事件循环可以运行其他任务。
8.1.3 FastAPI 对异步的支持
FastAPI 构建在 Starlette 和 AnyIO 之上，原生支持异步。当你用 async def 定义路径函数时，FastAPI 会在一个异步事件循环中运行它，能够以极低的线程开销同时处理成百上千个并发连接。
-
如果你的路径函数内部全都是异步操作（如
await异步数据库查询、异步 HTTP 请求），那么整个处理过程都是非阻塞的。 -
如果你的路径函数内部有同步阻塞操作（如
time.sleep(5)、open()读写大文件），它会阻塞整个事件循环，导致其他请求无法被处理。后面我们会讲如何安全地处理这种遗留代码。
8.2 BackgroundTasks 基础
8.2.1 添加后台任务
生活案例：回到快餐店。你点完餐后，最关心的是“点餐成功，我已经下好单了”，至于后厨怎么炸鸡、怎么打包，你不需要在柜台前等着。服务员在收银后，把你的订单小票夹在传菜线上，然后立刻为下一位顾客服务。你的餐会由后厨完成，再由传菜员送到你桌上。
在 Web 接口中，BackgroundTasks 就是那张“订单小票” — 你可以在返回响应之前，把一些不需要立刻完成的工作（如发送邮件、写日志、推送通知）放进后台任务队列，FastAPI 会在响应发送给客户端之后，再执行这些任务。
代码示例：
# 1. 导入所需的类与模块
from fastapi import FastAPI, BackgroundTasks # FastAPI 框架及后台任务支持
import time # 用于模拟耗时操作
import uvicorn # ASGI 服务器，用于运行 FastAPI 应用
# 2. 创建 FastAPI 应用实例
# title 参数会显示在自动生成的 Swagger 文档顶部
app = FastAPI(title="后台任务演示")
# 3. 定义后台任务函数（此处模拟写日志）
def write_log(message: str):
"""
模拟一个耗时的后台操作：写入日志文件。
实际项目中可能是发送邮件、生成报表、图像处理等。
"""
time.sleep(10) # 模拟耗时操作（10 秒）
# 以追加模式打开 log.txt，将日志信息写入文件
with open("log.txt", "a", encoding="utf-8") as f:
f.write(f"{time.ctime()}: {message}\n") # 写入当前时间与消息
print(f"日志已写入：{message}") # 控制台输出提示
# 4. 定义路由：POST /order/{item}
# {item} 是路径参数，由调用方传入（例如 /order/汉堡）
@app.post("/order/{item}")
async def order(
item: str, # 路径参数：商品名称
background_tasks: BackgroundTasks # 注入后台任务对象，FastAPI 自动提供
):
"""
用户下单接口：
1. 将写日志的操作注册为后台任务（不会阻塞响应）
2. 立即向客户端返回成功信息
"""
# 注册一个后台任务：函数 write_log，参数 f"订单：{item}"
background_tasks.add_task(write_log, f"订单：{item}")
# 立即返回响应，客户端无需等待日志写入完成
return {"msg": f"订单 {item} 已接收，正在处理"}
# 5. 程序入口：启动 Uvicorn 服务器
if __name__ == "__main__":
# host="0.0.0.0" 允许外部通过 IP 访问
# port=8000 指定监听端口
uvicorn.run(app, host="0.0.0.0", port=8000)测试：用 curl 或 Swagger 发送 POST /order/汉堡，你会立刻收到 JSON 响应，而 write_log 函数会在响应返回后执行。可以观察控制台：先返回了 200，然后 2 秒后才打印“日志已写入”。
8.2.2 任务函数与参数传递
add_task 的签名是：add_task(func, *args, **kwargs)
-
第一个参数：要延迟执行的函数对象（注意不要加括号，不是
write_log()）。 -
后面的位置参数、关键字参数会原样传递给该函数。
例如：
background_tasks.add_task(send_email, recipient, subject="Welcome")
# 等价于未来某个时间调用 send_email(recipient, subject="Welcome")你可以添加多个后台任务，它们会按添加的顺序依次执行（但不会阻塞响应返回）。
8.2.3 适用场景
-
发送邮件/短信：用户注册、密码重置。
-
记录审计日志：不期望用户等待日志写入。
-
生成报表/缩略图：复杂的文件处理。
-
推送通知：WebSocket 消息、App Push。
注意：后台任务运行在同一个进程内，如果任务很重（如视频转码），建议使用任务队列（Celery、Redis Queue）。
8.3 异步路径操作
FastAPI 允许你使用 async def 定义路径函数，这样函数内部就可以使用 await 来调用异步库，避免阻塞。我们通过对比同步与异步的执行效果，加深理解。
8.3.1 同步阻塞 vs 异步非阻塞
场景：一个接口需要模拟从数据库查询数据，耗时 1 秒。我们分别用同步 time.sleep 和异步 asyncio.sleep 实现，然后并发访问。
同步版本（阻塞）：
import time
from fastapi import FastAPI
app = FastAPI()
@app.get("/sync")
def sync_endpoint():
time.sleep(5) # 阻塞整个线程
return {"msg": "同步完成"}当多个请求同时访问 /sync 时，FastAPI 默认使用线程池处理同步函数，所以仍能并发处理，但每个线程都会因 time.sleep 而阻塞，浪费线程资源。
异步版本（非阻塞）：
import asyncio
from fastapi import FastAPI
app = FastAPI()
@app.get("/async")
async def async_endpoint():
await asyncio.sleep(5) # 非阻塞等待，让出控制权
return {"msg": "异步完成"}在异步版本中，await asyncio.sleep(1) 会暂停当前协程，事件循环立即处理其他请求。这才是真正的异步优势：用非常少的线程处理海量并发。
8.3.2 实际测试对比
我们写一个更直观的例子：模拟三个并发请求，对比时间。
from fastapi import FastAPI
import asyncio
import time
app = FastAPI()
@app.get("/sync-concurrent")
def sync_concurrent():
# 模拟一个耗时 I/O（阻塞）
time.sleep(1)
return {"msg": "ok"}
@app.get("/async-concurrent")
async def async_concurrent():
await asyncio.sleep(1)
return {"msg": "ok"}你可以用 ab （ApacheBench） 或简单的 Python 脚本并发访问，会发现异步接口吞吐量远高于同步接口（在线程池有限的情况下更明显）。
8.3.3 混合使用异步与同步
如果在一个异步路径函数中调用了同步阻塞函数（如 time.sleep、旧的数据库驱动），就会阻塞整个事件循环，导致所有其他异步操作暂停。这非常危险，必须避免。解决办法就是下一节要讲的 run_in_threadpool。
8.4 运行同步阻塞函数
8.4.1 为什么不能直接调用同步阻塞函数？
假设你有一个老旧的图像处理库，它的所有函数都是同步的，处理一张图片需要 5 秒。如果你在 async def 中直接调用它会发生什么？以下是一个完整的 FastAPI 应用，用图像处理场景演示了在异步路径函数中错误地直接调用同步阻塞函数与正确地使用 run_in_threadpool 的区别。你可以直接运行并对比效果。
import time
from fastapi import FastAPI
from fastapi.concurrency import run_in_threadpool
import uvicorn
app = FastAPI(title="同步阻塞 vs 线程池示例")
# ---------- 模拟一个遗留的同步图像处理库 ----------
class LegacyImageLib:
"""
模拟一个老旧的图像处理库，所有操作都是同步阻塞的。
实际中可能是 OpenCV、Pillow 等。
"""
@staticmethod
def process(filename: str) -> str:
"""
模拟耗时 10 秒的同步图像处理操作。
例如：读取大文件、降噪、识别等。
"""
print(f" ⏳ 开始处理图片：{filename}（阻塞 5 秒）")
time.sleep(10) # 模拟 CPU/IO 密集型操作
print(f" ✅ 图片处理完成：{filename}")
return f"processed_{filename}"
legacy_image_lib = LegacyImageLib()
# ---------- 错误示范：异步函数内直接调用同步阻塞函数 ----------
@app.get("/process-image/wrong")
async def process_image_wrong():
"""
错误做法：在 async def 中直接调用同步阻塞函数。
time.sleep 会阻塞整个事件循环，导致其他请求完全无法响应。
访问这个接口时，服务会卡死 5 秒。
"""
print("→ 错误接口被调用，直接执行同步阻塞操作...")
result = legacy_image_lib.process("photo.jpg") # 阻塞 5 秒！
return {"result": result}
# ---------- 正确做法：使用 run_in_threadpool ----------
@app.get("/process-image/right")
async def process_image_right():
"""
正确做法：通过 run_in_threadpool 将同步阻塞函数丢到线程池中执行。
事件循环不会被阻塞，其他异步接口依然可以正常响应。
"""
print("→ 正确接口被调用，将任务提交到线程池...")
result = await run_in_threadpool(legacy_image_lib.process, "photo.jpg")
return {"result": result}
# ---------- 普通异步接口（用于验证事件循环未被阻塞） ----------
@app.get("/ping")
async def ping():
"""一个轻量异步接口，用于测试服务是否还能响应"""
return {"msg": "pong"}
# ---------- 启动服务 ----------
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=8000)如何验证
（1）测试错误接口：在浏览器访问 http://127.0.0.1:8000/process-image/wrong
-
注意观察：在等待的 10 秒内，立即打开另一个标签页访问
/ping或/docs，你会发现它们也卡住了，直到图片处理完成。这是因为事件循环被time.sleep完全阻塞。
（2）测试正确接口：访问 http://127.0.0.1:8000/process-image/right
-
在等待的 10 秒内，可以随时访问
/ping，会立刻得到{"msg": "pong"}。说明事件循环没有被阻塞。
原理说明
-
错误做法：
async def内使用同步阻塞调用（time.sleep），整个线程被挂起，而 Python 异步事件循环依赖该线程去调度其他协程，因此所有异步操作瘫痪。 -
正确做法：
run_in_threadpool将阻塞函数移动到独立的线程中执行，事件循环线程继续处理其他请求，等待线程完成后通过await取回结果。这既兼容了旧代码，又保持了异步并发能力。
这 5 秒钟内，整个事件循环被卡死，无法响应任何其他请求，包括健康检查、实时消息等。因为 Python 的 async/await 是协作式多任务，一旦一个协程内部没有 await，它就会一直占用 CPU，其他协程得不到运行机会。
8.4.2 使用 run_in_threadpool 转移到线程池
FastAPI 提供了 fastapi.concurrency.run_in_threadpool，它会把同步阻塞函数丢到一个独立的线程池中运行，主线程的事件循环继续处理其他请求。等线程池执行完毕，再将结果以异步方式返回。
完整案例：
import time
from fastapi import FastAPI
from fastapi.concurrency import run_in_threadpool
app = FastAPI()
# 模拟一个遗留的同步阻塞函数
def legacy_blocking_io(seconds: int):
"""模拟阻塞 I/O，比如读取大文件、调用同步 SDK"""
print(f"开始耗时 {seconds} 秒的操作（线程中）...")
time.sleep(seconds)
print(f"操作完成")
return f"睡了 {seconds} 秒"
@app.get("/blocking")
async def blocking():
"""
正确做法：把同步阻塞函数扔进线程池。
接口响应不会阻塞事件循环。
"""
result = await run_in_threadpool(legacy_blocking_io, 3)
return {"msg": "完成", "result": result}
# 为了对比，再写一个错误示范（仅演示，生产禁用）
@app.get("/blocking-wrong")
async def blocking_wrong():
# 这会阻塞整个事件循环！
result = legacy_blocking_io(3)
return {"msg": "完成", "result": result}测试：访问 /blocking，在等待 3 秒期间，你可以继续访问其他异步接口（如 GET /docs），一切正常。访问 /blocking-wrong，你会发现在这 3 秒内，整个服务都卡死了，其他请求全部超时。
8.4.3 应用场景
-
使用同步的数据库驱动（如某些老版本的 SQLAlchemy）。
-
调用同步的机器学习推理库。
-
文件 I/O 操作（
open，PIL.Image.save等）。 -
调用第三方同步 API 包装器。
最佳实践：如果你的路径函数内部大部分是 I/O 操作，请尽量使用异步库（httpx、asyncpg、aioredis）。只有在迫不得已使用遗留同步库时，再用 run_in_threadpool 隔离。
本章总结
-
同步与异步：同步像排队等餐，异步像拿号后等叫号。异步利用等待时间处理其他任务，提升并发。
-
BackgroundTasks：适合请求后立即返回，把耗时“杂活”放到后台完成。 -
异步路径函数：使用
async def和await，保持事件循环畅快，支持高并发。 -
同步阻塞函数的处理：千万不要在
async def里直接调用同步阻塞函数；用run_in_threadpool转移到线程池。
通过合理运用后台任务和异步特性，你的 FastAPI 应用将具备企业级的高并发处理能力，响应丝般顺滑。
第 9 章 安全与认证
一个开放的 API 如果没有安全机制，就像一家没有门禁的银行 — 任何人都能随意进入查看客户信息、转账取款，后果不堪设想。本章将带你从零开始，为 FastAPI 应用添加三层安全防护：OAuth2 密码流（用户登录获取令牌）、JWT（签发和验证令牌）、以及 API Key（为机器间通信提供简易凭证）。同时，我们还会复习 CORS 配置，确保前端能安全调用。
9.1 OAuth2 密码流：让用户安全登录
9.1.1 OAuth2 概念速览
生活案例：你下载了一个第三方记账 App，它需要读取你的银行流水。你不会傻到把银行卡密码直接告诉这个 App — 你用的是银行的“开放授权”功能：银行让你在自己的登录页面确认授权，然后发给 App 一个临时电子凭证（Token），App 拿着这个凭证去查询你的流水，全程看不到你的密码。这就是 OAuth2 的授权码流程，适用于第三方应用。
但如果你用的是银行自己的官方 App（你完全信任它），你会直接在 App 里输入用户名密码，App 后端立刻验证并返回一个 Token，之后用这个 Token 来识别你。这就是 OAuth2 的密码流，适用于自己公司开发的前后端分离应用。
OAuth2 密码流的步骤：
（1）用户在登录页输入 username 和 password。
（2）前端将这些信息发给后端的 /token 接口。
（3）后端验证用户名密码，如果正确，返回一个 access_token（访问令牌）。
（4）前端在后续请求的 Authorization 头中携带这个 token。
（5）后端通过 token 识别用户身份，处理业务请求。
9.1.2 OAuth2PasswordBearer 配置 — 详细案例
OAuth2PasswordBearer 是 FastAPI 安全模块的核心组件之一。它本身不做任何验证，只是一个“令牌提取器”，负责从请求头 Authorization: Bearer <token> 中取出 token 字符串，然后传递给下游的依赖函数进行真正的校验。
生活案例深化
想象一栋高科技办公大楼：
-
大楼入口有一个刷卡器（
OAuth2PasswordBearer）。 -
每位员工胸前挂着工牌（JWT token），上面印着姓名、部门、过期日期。
-
进入大楼时，你需要在刷卡器上刷一下工牌。刷卡器只负责读取工牌上的编号，它不判断工牌真假，只是把编号传给保安室。
-
保安室里的验证系统（
get_current_user依赖函数）拿到编号后，去数据库核对：这张工牌是真的吗？过期了吗？核对通过后，才放你进入。
关键理解：OAuth2PasswordBearer 就是那个刷卡器，它只负责“读取”，不负责“判断”。
以下代码演示了 OAuth2PasswordBearer 的完整用法：配置它 → 用它提取 token → 验证 token → 保护接口。
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
import uvicorn
app = FastAPI(title="OAuth2PasswordBearer 详解")
# ---------- 1. 配置 OAuth2PasswordBearer ----------
# 创建一个“令牌提取器”实例
# tokenUrl 参数的作用：
# - 告诉 Swagger 文档：“获取 token 的接口是 /api/login”
# - 当用户在 Swagger 中点击右上角绿色 "Authorize" 按钮时，
# 弹出的对话框会提示用户先去 /api/login 获取 token
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/login")
# ---------- 2. 模拟用户数据库 ----------
fake_users_db = {
"admin": {
"user_id": 1,
"username": "admin",
"token": "secret-token-admin-123"
},
"user": {
"user_id": 2,
"username": "user",
"token": "secret-token-user-456"
}
}
# ---------- 3. 模拟登录接口 ----------
@app.post("/api/login")
async def login(username: str, password: str):
"""
简化版登录接口。
实际项目中这里应该验证密码，并返回 JWT token。
这里为了聚焦 OAuth2PasswordBearer 的讲解，直接返回固定 token。
"""
# 模拟验证（实际应查数据库、校验密码哈希）
user = fake_users_db.get(username)
if not user:
raise HTTPException(status_code=401, detail="用户名不存在")
# 返回 token（格式必须包含 access_token 和 token_type）
return {
"access_token": user["token"],
"token_type": "bearer"
}
# ---------- 4. 依赖函数：验证 token 并获取用户 ----------
async def get_current_user(token: str = Depends(oauth2_scheme)):
"""
这个依赖函数接收 oauth2_scheme 提取的 token 字符串。
oauth2_scheme 自动从请求头 Authorization: Bearer <token> 中取值。
函数内部做真正的验证：查数据库，判断 token 是否有效。
"""
# 遍历模拟数据库，查找持有该 token 的用户
for username, user_data in fake_users_db.items():
if user_data["token"] == token:
return {"user_id": user_data["user_id"], "username": username}
# 如果没找到，说明 token 无效
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="无效的认证凭证",
headers={"WWW-Authenticate": "Bearer"},
)
# ---------- 5. 受保护的接口 ----------
@app.get("/users/me")
async def read_users_me(current_user: dict = Depends(get_current_user)):
"""
这个接口需要登录。
当用户访问时，流程如下：
1. oauth2_scheme 从请求头提取 token
2. get_current_user 用 token 查找用户
3. 找到用户后，用户信息传给 current_user
4. 接口返回用户信息
"""
return {
"message": f"欢迎你，{current_user['username']}！",
"user_id": current_user["user_id"]
}
@app.get("/users/settings")
async def read_settings(current_user: dict = Depends(get_current_user)):
"""另一个受保护的接口，复用同一个认证依赖"""
return {
"user": current_user["username"],
"settings": {"theme": "dark", "lang": "zh-CN"}
}
# ---------- 6. 公开接口（无需登录） ----------
@app.get("/public/news")
async def public_news():
"""公开接口，任何人都能访问，不需要 token"""
return {"news": "今日要闻：FastAPI 越来越流行了！"}
# ---------- 启动入口 ----------
if __name__ == "__main__":
uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)测试流程
1. 测试公开接口（无需 token）
curl http://127.0.0.1:8000/public/news返回：
{"news": "今日要闻：FastAPI 越来越流行了！"}2. 测试受保护接口（不带 token，应该被拒绝）
curl http://127.0.0.1:8000/users/me返回 401 Unauthorized：
{"detail": "Not authenticated"}这是因为请求头中没有 Authorization 字段，oauth2_scheme 无法提取 token，直接抛出 401。
3. 登录获取 token
curl -X POST http://127.0.0.1:8000/api/login \
-H "Content-Type: application/x-www-form-urlencoded" \
-d "username=admin&password=any"返回：
{
"access_token": "secret-token-admin-123",
"token_type": "bearer"
}4. 使用 token 访问受保护接口
curl http://127.0.0.1:8000/users/me \
-H "Authorization: Bearer secret-token-admin-123"返回：
{
"message": "欢迎你，admin！",
"user_id": 1
}5. 在 Swagger UI 中测试
（1）启动服务后，打开 http://127.0.0.1:8000/docs。
（2）你会发现 GET /users/me 和 GET /users/settings 旁边有一个锁形图标，表示它们需要认证。
（3）点击右上角绿色 Authorize 按钮。
（3）在弹出的对话框中输入 secret-token-admin-123（注意：Swagger 会自动帮你加上 Bearer 前缀，你只需输入 token 本身）。
（4）点击 Authorize，然后关闭对话框。
（5）现在点击 GET /users/me 的 Try it out → Execute，会自动携带 token，返回用户信息。
OAuth2PasswordBearer 的工作流程总结
客户端请求
↓
Authorization: Bearer secret-token-admin-123
↓
┌──────────────────────────────────┐
│ oauth2_scheme │ ← 刷卡器：读取请求头
│ (OAuth2PasswordBearer) │
│ │
│ 提取出: "secret-token-admin-123" │
│ 如果请求头缺失 → 直接返回 401 │
└──────────────┬───────────────────┘
│ token 字符串
↓
┌──────────────────────────────────┐
│ get_current_user(token) │ ← 保安室：验证 token
│ (自定义依赖函数) │
│ │
│ 1. 查数据库 │
│ 2. 判断 token 是否有效 │
│ 3. 有效 → 返回用户信息 │
│ 4. 无效 → 返回 401 │
└──────────────┬───────────────────┘
│ 用户信息字典
↓
路径函数执行
(current_user)一句话总结：OAuth2PasswordBearer 只做一件事 — 从请求头里把 Bearer 后面的 token 字符串抠出来。至于这个 token 是真还是假、有没有过期，它一概不管，全交给后面的依赖函数处理。这种“职责分离”的设计让安全逻辑清晰且易于测试。
9.1.3 OAuth2PasswordRequestForm 接收登录表单 — 详细案例
OAuth2PasswordRequestForm 是 FastAPI 安全模块提供的登录表单解析器。它专门用来接收 OAuth2 密码流中用户提交的用户名和密码，这些数据通常以表单格式（application/x-www-form-urlencoded）而非 JSON 格式发送。
生活案例深化
你去银行柜台办理业务：
（1）柜员递给你一张标准化的表格（这就是 OAuth2PasswordRequestForm），上面有两个必填栏：账号 和 密码。
（2）你不需要自己找纸手写，也不必纠结格式——表格格式是银行统一规定的。
（3）填好后你递回表格，柜员自动解析出账号和密码字段，然后转身去后台系统核对。
（4）核对通过后，柜员给你一张排队小票（JWT token），后续凭这张小票就能办理各种业务。
为什么用表单而不是 JSON？
这是 OAuth2 标准（RFC 6749）的规定。使用标准的表单格式（application/x-www-form-urlencoded），可以确保各种编程语言、各种 HTTP 客户端都能轻松地提交登录请求，不需要额外的 JSON 解析库。就连最基础的 HTML <form> 标签也能直接提交。
完整可运行案例
以下代码演示了 OAuth2PasswordRequestForm 的完整用法：接收登录表单 → 验证用户名密码 → 生成 JWT token → 返回给客户端。
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm, OAuth2PasswordBearer
from datetime import datetime, timedelta
from jose import jwt, JWTError
import uvicorn
app = FastAPI(title="OAuth2PasswordRequestForm 详解")
# ---------- 1. 配置 ----------
# JWT 密钥和算法
SECRET_KEY = "my-secret-key-for-jwt-signing-2025"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
# OAuth2 令牌提取器（用于后续受保护接口）
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/token")
# ---------- 2. 模拟用户数据库 ----------
fake_users_db = {
"admin": {
"user_id": 1,
"username": "admin",
"password": "123456", # 实际项目中应存储密码哈希
"role": "管理员",
"department": "技术部"
},
"zhangsan": {
"user_id": 2,
"username": "zhangsan",
"password": "password123",
"role": "普通用户",
"department": "市场部"
}
}
# ---------- 3. JWT 工具函数 ----------
def create_jwt_token(user_id: int, username: str, role: str) -> str:
"""根据用户信息生成 JWT token"""
now = datetime.utcnow()
payload = {
"sub": str(user_id), # subject：用户 ID
"username": username, # 用户名
"role": role, # 角色（自定义字段）
"iat": now, # 签发时间
"exp": now + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES) # 过期时间
}
token = jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)
return token
# ---------- 4. 验证 token 的依赖函数 ----------
async def get_current_user(token: str = Depends(oauth2_scheme)):
"""从 token 中解析出用户信息"""
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="无法验证凭证，请重新登录",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
user_id: str = payload.get("sub")
username: str = payload.get("username")
role: str = payload.get("role")
if user_id is None or username is None:
raise credentials_exception
return {"user_id": int(user_id), "username": username, "role": role}
except JWTError:
raise credentials_exception
# ---------- 5. 核心：登录接口 ----------
@app.post("/api/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
"""
用户登录接口。
OAuth2PasswordRequestForm 会自动解析表单中的以下字段：
- username：用户名（必填）
- password：密码（必填）
- grant_type：授权类型（可选，密码流通常为 "password"）
- scope：权限范围（可选，如 "read write"）
- client_id：客户端 ID（可选）
- client_secret：客户端密钥（可选）
在 Swagger UI 中，这个接口会自动显示为一个专用的登录表单。
"""
# 1. 从表单中获取用户名和密码
username = form_data.username
password = form_data.password
# 2. 在数据库中查找用户
user = fake_users_db.get(username)
# 3. 验证用户是否存在，密码是否正确
if not user:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="用户名不存在",
headers={"WWW-Authenticate": "Bearer"},
)
if user["password"] != password:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="密码错误",
headers={"WWW-Authenticate": "Bearer"},
)
# 4. 验证通过，生成 JWT token
token = create_jwt_token(
user_id=user["user_id"],
username=user["username"],
role=user["role"]
)
# 5. 返回 token（必须包含 access_token 和 token_type）
print(f"✅ 用户 {username} 登录成功，token 已生成")
return {
"access_token": token,
"token_type": "bearer"
}
# ---------- 6. 受保护的接口 ----------
@app.get("/users/me")
async def read_users_me(current_user: dict = Depends(get_current_user)):
"""查看当前登录用户的信息"""
return {
"message": f"欢迎回来，{current_user['username']}！",
"user_id": current_user["user_id"],
"role": current_user["role"]
}
@app.get("/users/me/role")
async def read_my_role(current_user: dict = Depends(get_current_user)):
"""查看当前用户的角色"""
return {
"username": current_user["username"],
"role": current_user["role"]
}
# ---------- 7. 公开接口 ----------
@app.get("/public/health")
async def health_check():
"""健康检查接口，无需登录"""
return {"status": "ok", "message": "服务运行正常"}
# ---------- 启动入口 ----------
if __name__ == "__main__":
uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)测试流程
方式一：在 Swagger UI 中测试（推荐）
这是最直观的方式，能直接看到 OAuth2PasswordRequestForm 生成的专用登录表单。
（1）打开 Swagger UI：访问 http://127.0.0.1:8000/docs
（2）查看登录接口：找到 POST /api/token，展开它。你会发现这个接口和之前见过的都不一样——它自动生成了一个专用的登录表单，包含 username 和 password 两个输入框，就像真正的登录页面一样。
（3）尝试登录：
-
点击 Try it out
-
在
username输入框中填写admin -
在
password输入框中填写123456 -
点击 Execute
（4）查看响应：
{
"access_token": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...",
"token_type": "bearer"
}（5）授权后续请求：
-
复制返回的
access_token值 -
点击页面右上角绿色 Authorize 按钮
-
在弹出的对话框中粘贴 token
-
点击 Authorize，然后关闭对话框
（6）访问受保护接口：
-
现在点击
GET /users/me的 Try it out → Execute -
会自动在请求头中携带 token，返回你的用户信息
方式二：使用 curl 命令行测试
1. 测试公开接口（无需 token）
curl http://127.0.0.1:8000/public/health2. 使用正确的用户名密码登录
curl -X POST http://127.0.0.1:8000/api/token \
-H "Content-Type: application/x-www-form-urlencoded" \
-d "username=admin&password=123456"返回：
{
"access_token": "eyJhbGciOiJIUzI1NiIs...",
"token_type": "bearer"
}3. 使用错误的密码登录
curl -X POST http://127.0.0.1:8000/api/token \
-H "Content-Type: application/x-www-form-urlencoded" \
-d "username=admin&password=wrongpassword"返回 401 Unauthorized：
{"detail": "密码错误"}4. 使用获取的 token 访问受保护接口
# 将 <TOKEN> 替换为上一步获取的 access_token
curl http://127.0.0.1:8000/users/me \
-H "Authorization: Bearer <TOKEN>"返回：
{
"message": "欢迎回来，admin！",
"user_id": 1,
"role": "管理员"
}OAuth2PasswordRequestForm 解析的字段详解
|
字段 |
类型 |
是否必填 |
说明 |
|---|---|---|---|
|
|
|
✅ 必填 |
用户名 |
|
|
|
✅ 必填 |
密码 |
|
|
|
❌ 可选 |
授权类型，密码流通常为 |
|
|
|
❌ 可选 |
权限范围，如 |
|
|
|
❌ 可选 |
客户端 ID |
|
|
|
❌ 可选 |
客户端密钥 |
你可以在代码中访问这些字段，例如：
print(f"授权类型: {form_data.grant_type}")
print(f"权限范围: {form_data.scope}")与普通 JSON 接口的对比
# ❌ 如果使用 JSON 接收登录数据
class LoginRequest(BaseModel):
username: str
password: str
@app.post("/api/login-json")
async def login_json(data: LoginRequest):
...
# ✅ 使用 OAuth2PasswordRequestForm（符合 OAuth2 标准）
@app.post("/api/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
...使用 OAuth2PasswordRequestForm 的好处：
-
符合国际标准：遵循 OAuth2 RFC 6749 规范。
-
Swagger 自动生成专用登录表单：用户体验更好。
-
自动解析：无需手动定义 Pydantic 模型。
-
多语言兼容：任何 HTTP 客户端都能用标准表单格式提交。
总结
-
OAuth2PasswordRequestForm是 FastAPI 为 OAuth2 密码流量身定做的表单解析器。 -
它自动从
application/x-www-form-urlencoded格式的请求体中提取username和password。 -
在 Swagger UI 中会生成一个专用登录对话框，方便测试。
-
它是 OAuth2 密码流的标准入口，通常与
OAuth2PasswordBearer配合使用：前者负责签发 token，后者负责提取 token。
9.2 JWT 认证实现：颁发电子身份证
9.2.1 JWT 的结构与原理
生活案例：你去银行办理业务，柜员给你一张盖章的凭证，上面写着“持证人：张三，账户级别：VIP，有效期至：2025 年 12 月 31 日”。以后你拿这张凭证去任何分行，工作人员一看印章完整，就知道这是本行签发、内容未被篡改过的，于是直接按凭证内容为你服务，不需要每次都调出你的原始档案核实身份。
JWT（JSON Web Token） 就是这张盖章凭证。它由三部分组成，用点号 . 连接：
Header.Payload.Signature-
Header（头部）：记录签名算法，比如
{"alg": "HS256", "typ": "JWT"}。 -
Payload（载荷）：存放“证件内容”，比如
{"sub": "用户ID", "username": "张三", "exp": 到期时间}。注意，这里的内容是 Base64 编码，不是加密，任何人都能解码查看，所以绝不能在载荷中存放密码。 -
Signature（签名）：将“头部。载荷”与服务器持有的密钥混合计算得到的一个哈希值。服务器收到 JWT 后，用同样的密钥再算一遍，如果结果一致，说明 token 确实是我签发的，且中途未被篡改。
9.2.2 生成 token
需要安装 python-jose 库来处理 JWT 的编码和解码：
pip install python-jose[cryptography]代码实现：
from datetime import datetime, timedelta
from jose import jwt
# 密钥（生产环境必须从环境变量读取，且足够长、足够随机）
SECRET_KEY = "my-very-secret-key-keep-it-safe-2025"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30 # token 有效期 30 分钟
def create_jwt_token(user_id: int, username: str) -> str:
"""
根据用户信息生成一个 JWT token。
"""
# 当前时间
now = datetime.utcnow()
# 载荷：存放需要保存在 token 中的用户信息
payload = {
"sub": str(user_id), # subject：通常放用户 ID
"username": username, # 自定义字段
"iat": now, # issued at：签发时间
"exp": now + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES) # 过期时间
}
# 使用密钥对载荷进行签名，生成最终的 token 字符串
token = jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)
return token生活类比：create_jwt_token 就像银行的制证机器 — 你把用户信息（ID、姓名）告诉它，它就用内部密码机（SECRET_KEY）盖一个防伪章，吐出一张带签名的凭证。
9.2.3 验证 token 与依赖注入
前端拿到 token 后，每次请求都把它放在 Authorization: Bearer <token> 头中。后端需要写一个依赖函数，从请求头中提取 token、验证签名、解析出用户信息。
from fastapi import Depends, HTTPException, status
from jose import jwt, JWTError
def get_current_user(token: str = Depends(oauth2_scheme)):
"""
从请求头中提取 token，验证签名和有效期，
如果验证失败（过期、伪造、缺失），直接返回 401 未授权。
如果验证通过，返回解析出的用户信息。
"""
# 自定义异常：未提供 token 或格式错误
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="无法验证凭证",
headers={"WWW-Authenticate": "Bearer"},
)
try:
# 使用密钥解码 token
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
user_id: str = payload.get("sub")
username: str = payload.get("username")
if user_id is None or username is None:
raise credentials_exception
except JWTError:
# 签名验证失败、token 过期等
raise credentials_exception
# 返回用户信息（路径函数可以直接使用）
return {"user_id": int(user_id), "username": username}保护接口：
@app.get("/users/me")
async def read_users_me(current_user: dict = Depends(get_current_user)):
"""
只有携带有效 token 的用户才能访问这个接口。
current_user 就是从 token 中解析出的用户信息。
"""
return {
"message": f"欢迎你，{current_user['username']}！",
"user_id": current_user['user_id']
}完整登录流程：
（1）用户访问受保护的 /users/me，因为没有 token，服务端返回 401。
（2）用户访问 /api/token（在 Swagger 中点击 Authorize 并输入用户名密码），获取一个 JWT 字符串。
（3）用户再次访问 /users/me，在请求头中带上 Authorization: Bearer <JWT>，服务端验证通过，返回用户信息。
9.3 API Key 校验方案
JWT 适合用户登录场景，但如果是机器与机器之间的通信（比如后端定时任务调用另一个服务的接口），让“机器”走用户登录流程就很别扭。此时可以用更简单的 API Key — 双方提前约定一个固定字符串，每次请求都带上这个字符串作为凭证。
9.3.1 从请求头读取 API Key
from fastapi import Header, HTTPException
def verify_api_key(x_api_key: str = Header(..., description="API 密钥")):
"""
从请求头 X-API-Key 中读取 API Key，
与预设的密钥比对。不匹配则返回 403 禁止访问。
"""
VALID_API_KEY = "secret-api-key-2025" # 生产环境应从环境变量读取
if x_api_key != VALID_API_KEY:
raise HTTPException(status_code=403, detail="无效的 API Key")
return x_api_key生活类比：API Key 就像大楼后门的一个密码锁。快递员不需要在前台登记身份（JWT），只需要在密码盘上输入正确的 6 位密码（API Key）就能进入。这个密码是所有快递员共用的，不区分具体是谁。
9.3.2 编写验证依赖并在路由中使用
@app.get("/api/internal-data")
async def internal_data(api_key: str = Depends(verify_api_key)):
"""
仅允许携带有效 API Key 的客户端访问。
适合内部服务间调用，无需用户登录。
"""
return {"data": "这是内部数据"}JWT vs API Key 对比：
|
特性 |
JWT |
API Key |
|---|---|---|
|
适用场景 |
用户登录、需要识别“是谁” |
服务间调用、简单认证 |
|
是否包含用户信息 |
是（载荷中可自定义字段） |
否，只是一个密钥 |
|
时效性 |
有过期时间，可设置刷新机制 |
通常长期有效，需手动轮换 |
|
安全性 |
防篡改，支持签名 |
明文传输，必须配合 HTTPS |
9.4 CORS 安全配置
前面第 7 章已详细介绍过 CORSMiddleware，这里快速回顾。CORS（跨域资源共享）并不是认证机制，而是浏览器强制的安全策略。当你的前端（如 localhost:3000）向后端（localhost:8000）发送请求时，浏览器会阻止跨域访问，除非后端在响应头中明确允许。
完整安全配置示例：
from fastapi.middleware.cors import CORSMiddleware
app.add_middleware(
CORSMiddleware,
allow_origins=["http://localhost:3000"], # 只允许信任的前端地址
allow_credentials=True,
allow_methods=["GET", "POST"], # 只允许必要的方法
allow_headers=["*"],
)本章完整运行示例
将上述所有代码整合为一个 main.py，你可以直接运行，在 Swagger UI（/docs）中体验完整的登录 → 获取 Token → 访问受保护接口流程，以及 API Key 验证接口。
from fastapi import FastAPI, Depends, HTTPException, status, Header
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from fastapi.middleware.cors import CORSMiddleware
from datetime import datetime, timedelta
from jose import jwt, JWTError
import uvicorn
app = FastAPI(title="安全认证演示")
# CORS 配置
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# JWT 配置
SECRET_KEY = "my-very-secret-key-change-in-production"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
# OAuth2 令牌提取器（指向登录接口）
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/token")
# ---------- 模拟用户数据库 ----------
fake_users_db = {
"admin": {"user_id": 1, "username": "admin", "password": "123456"},
"user": {"user_id": 2, "username": "user", "password": "password"}
}
# ---------- JWT 工具函数 ----------
def create_jwt_token(user_id: int, username: str):
payload = {
"sub": str(user_id),
"username": username,
"iat": datetime.utcnow(),
"exp": datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
}
return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)
def get_current_user(token: str = Depends(oauth2_scheme)):
exc = HTTPException(status_code=401, detail="无法验证凭证")
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
return {"user_id": int(payload["sub"]), "username": payload["username"]}
except JWTError:
raise exc
# ---------- API Key 校验 ----------
def verify_api_key(x_api_key: str = Header(...)):
if x_api_key != "secret-api-key":
raise HTTPException(status_code=403, detail="无效的 API Key")
return x_api_key
# ---------- 接口 ----------
@app.post("/api/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
user = fake_users_db.get(form_data.username)
if not user or user["password"] != form_data.password:
raise HTTPException(status_code=401, detail="用户名或密码错误")
token = create_jwt_token(user["user_id"], user["username"])
return {"access_token": token, "token_type": "bearer"}
@app.get("/users/me")
async def read_me(current_user: dict = Depends(get_current_user)):
return {"msg": f"欢迎，{current_user['username']}！"}
@app.get("/api/internal")
async def internal(api_key: str = Depends(verify_api_key)):
return {"data": "内部机密数据"}
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=8000)测试步骤：
（1）启动服务，打开 http://127.0.0.1:8000/docs
（2）展开 POST /api/token，点击 Try it out，输入用户名 admin、密码 123456，获取 token。
（3）复制返回的 access_token，点击页面右上角绿色 Authorize 按钮，粘贴 token 并确认。
（4）现在访问 GET /users/me，会自动携带 token，返回你的用户信息。
（5）测试 GET /api/internal，需要手动传入 X-API-Key: secret-api-key（在 Swagger 的 Header 输入框中添加）。
第 10 章 测试与文档：让接口质量有保障，让文档一目了然
一个没有经过测试的接口就像一辆没有经过质检的汽车 — 你永远不知道它什么时候会抛锚，把用户甩在半路上。而一份没有说明书的 API 就像没有菜谱的厨房，调用方只能摸着石头过河，沟通成本极高。FastAPI 提供了 TestClient 让你像真实用户一样测试接口，依赖覆盖让你能隔离外部服务，自动生成的 OpenAPI 文档则让你可以用几行代码就拥有专业的交互式说明书。本章将通过一个“图书管理 API”的完整案例，手把手带你掌握测试与文档的核心技能。
10.1 TestClient 基础：像浏览器一样发请求
生活案例：汽车出厂前要经过多项检测：踩油门看响应、转方向盘看转向灯、按喇叭听声音。质检员不需要真的把车开到马路上，而是在室内用一套检测设备模拟各种操作。FastAPI 的 TestClient 就是这个检测设备 — 它让你在代码里直接向你的接口发送 GET、POST 请求，检查返回的状态码和内容，而不需要真正启动一个网络服务
10.1.1 导入 TestClient
TestClient 位于 fastapi.testclient 模块中。你只需要把 FastAPI 应用实例传给它，它就会返回一个客户端对象，这个对象拥有 .get()、.post() 等和 requests 库几乎一样的接口。
from fastapi.testclient import TestClient
from main import app # 你的 FastAPI 应用实例
client = TestClient(app) # 创建测试客户端10.1.2 模拟 GET/POST 请求
我们先创建一个简单的图书管理 API，然后用 TestClient 向它发起请求。
第一步：创建被测试的应用 main.py
from fastapi import FastAPI, HTTPException
app = FastAPI(title="图书管理 API")
# 模拟数据库
books = {
1: {"id": 1, "title": "数据密集型应用系统设计", "author": "马丁·科勒普曼"},
2: {"id": 2, "title": "重构：改善既有代码的设计", "author": "马丁·福勒"}
}
@app.get("/books")
def list_books():
"""获取所有图书"""
return list(books.values())
@app.get("/books/{book_id}")
def get_book(book_id: int):
"""获取单本图书"""
book = books.get(book_id)
if not book:
raise HTTPException(status_code=404, detail="图书不存在")
return book
@app.post("/books", status_code=201)
def create_book(title: str, author: str):
"""新增图书"""
new_id = max(books.keys()) + 1
new_book = {"id": new_id, "title": title, "author": author}
books[new_id] = new_book
return new_book第二步：编写测试脚本 test_main.py
from fastapi.testclient import TestClient
from main import app
client = TestClient(app)
def test_read_main():
"""测试获取所有图书"""
response = client.get("/books")
assert response.status_code == 200
data = response.json()
assert len(data) == 2
assert data[0]["title"] == "数据密集型应用系统设计"
def test_get_book():
"""测试获取单本图书"""
response = client.get("/books/1")
assert response.status_code == 200
assert response.json()["title"] == "数据密集型应用系统设计"
def test_create_book():
"""测试新增图书"""
response = client.post("/books?title=测试书&author=测试作者")
assert response.status_code == 201
data = response.json()
assert data["title"] == "测试书"
# 确认新增后总数增加
response2 = client.get("/books")
assert len(response2.json()) == 3
if __name__ == "__main__":
test_read_main()
test_get_book()
test_create_book()
print("所有测试通过")先安装相关库：
pip install httpx2运行方式：
python test_main.py输出如下内容：
但这只是普通 Python 脚本，我们需要引入更专业的测试框架。
10.2 与 pytest 集成：更专业的测试管理
10.2.1 安装 pytest
pytest 是 Python 生态中最流行的测试框架，它自动发现以 test_ 开头的文件和函数，提供详细的测试报告。
pip install pytest10.2.2 编写测试文件与函数
按照惯例，测试文件放在 tests/ 目录下，文件名为 test_*.py，函数名以 test_ 开头。
项目结构：
project/
├── main.py
└── tests/
└── test_main.py我们重写测试代码，使其完全兼容 pytest。只需去掉 TestClient 的手动执行，让 pytest 自动运行。tests/test_main.py测试代码如下：
from fastapi.testclient import TestClient
from main import app
client = TestClient(app)
def test_list_books():
response = client.get("/books")
assert response.status_code == 200
assert isinstance(response.json(), list)
def test_get_book_found():
response = client.get("/books/1")
assert response.status_code == 200
assert response.json()["title"] == "数据密集型应用系统设计"
def test_get_book_not_found():
response = client.get("/books/999")
assert response.status_code == 404
assert "图书不存在" in response.json()["detail"]
def test_create_book():
response = client.post("/books?title=新书&author=无名")
assert response.status_code == 201
data = response.json()
assert data["title"] == "新书"10.2.3 运行测试与查看报告
在项目根目录下执行：
pytest输出类似：
============================= test session starts ==============================
collected 4 items
tests/test_main.py .... [100%]
============================== 4 passed in 0.15s ===============================如果要看到更详细的信息，可以加 -v：
pytest -v10.3 依赖覆盖：隔离外部依赖，让测试更纯粹
在实际项目中，你的接口可能依赖数据库连接、外部 API 等。测试时我们不想真的连接数据库（慢、状态不可控），也不想真的调用外部服务（花钱、不稳定）。FastAPI 的 dependency_overrides 允许你在测试中用虚假对象替换掉真实的依赖函数。
生活案例：飞行员训练时不会直接开真飞机，而是在模拟驾驶舱里练习。模拟驾驶舱的按钮和真飞机一模一样，但背后连接的不是真实引擎，而是计算机模拟的飞行参数。这样既安全又廉价。依赖覆盖就是给接口换上“模拟驾驶舱”，让测试在沙盒中运行。
10.3.1 app.dependency_overrides 用法
假设我们有一个依赖函数 get_db() 用于获取数据库会话。在测试中，我们希望它返回一个假的字典。
修改 main.py，加入依赖：
from fastapi import FastAPI, Depends, HTTPException
app = FastAPI()
# 模拟一个“沉重”的数据库依赖
def get_db():
"""真实环境：返回数据库连接，这里用字典模拟"""
return {"connection": "真实数据库连接", "books": {
1: {"id": 1, "title": "数据密集型应用系统设计", "author": "马丁·科勒普曼"}
}}
@app.get("/books")
def list_books(db: dict = Depends(get_db)):
return list(db["books"].values())测试时，我们希望 get_db 返回的 books 不干扰真实数据。
在测试文件中，我们可以这样覆盖：
import sys
from pathlib import Path
# 将项目根目录（my_fatapi_project）加入 Python 模块搜索路径
sys.path.insert(0, str(Path(__file__).parent.parent))
from fastapi.testclient import TestClient
from main import app, get_db
# 定义替换用的假依赖
def fake_get_db():
return {
"connection": "模拟连接",
"books": {
1: {"id": 1, "title": "测试书", "author": "测试作者"}
}
}
# 注册依赖覆盖
app.dependency_overrides[get_db] = fake_get_db
client = TestClient(app)
def test_list_books_with_fake_db():
response = client.get("/books")
assert response.status_code == 200
data = response.json()
assert data[0]["title"] == "测试书"10.3.2 在测试中替换数据库依赖
完整示例，我们在 main.py 中写一个更真实的场景：使用 SQLAlchemy 连接数据库（实际我们只是模拟，但依赖函数返回一个会话对象）。然后在测试中覆盖它。
为保持简洁，我们用字典模拟数据库会话，但结构类似。
main.py:
from fastapi import FastAPI, Depends, HTTPException
app = FastAPI()
class Database:
"""模拟数据库会话"""
def query_books(self):
return [
{"id": 1, "title": "真实书名", "author": "真实作者"}
]
# 真实依赖：返回数据库会话
def get_db():
return Database()
@app.get("/books")
def get_books(db: Database = Depends(get_db)):
return db.query_books()测试文件 tests/test_main.py：
import sys
from pathlib import Path
# 将项目根目录（my_fatapi_project）加入 Python 模块搜索路径
sys.path.insert(0, str(Path(__file__).parent.parent))
from fastapi.testclient import TestClient
from main import app, get_db, Database
class FakeDatabase:
"""模拟的数据库会话"""
def query_books(self):
return [
{"id": 1, "title": "模拟书名", "author": "模拟作者"}
]
def fake_get_db():
return FakeDatabase()
# 覆盖依赖
app.dependency_overrides[get_db] = fake_get_db
client = TestClient(app)
def test_get_books_with_mock_db():
response = client.get("/books")
assert response.status_code == 200
assert response.json()[0]["title"] == "模拟书名"10.3.3 隔离外部依赖（清理）
在 FastAPI 测试中，我们经常使用 app.dependency_overrides 来替换真实的依赖（比如数据库会话、认证等），以便隔离测试环境。但如果不清理这些覆盖，可能会影响后续测试用例，导致意外的行为。
pytest 的 fixture 机制可以优雅地在每个测试结束后自动执行清理。
方法：定义一个 autouse=True 的 fixture
import sys
from pathlib import Path
# 将项目根目录（my_fatapi_project）加入 Python 模块搜索路径
sys.path.insert(0, str(Path(__file__).parent.parent))
import pytest
from fastapi.testclient import TestClient
from main import app, get_db
# 假设你有一个测试用的数据库会话
def test_db_session():
return "mock_db_session"
# 在测试模块中定义 fixture
@pytest.fixture(autouse=True)
def clear_dependency_overrides():
"""每个测试执行前设置依赖覆盖，执行后自动清除"""
# 这里的代码会在每个测试函数运行前执行
app.dependency_overrides[get_db] = test_db_session
yield # 测试函数在这里运行
# 测试结束后执行清理
app.dependency_overrides.clear()
# 现在你可以写测试函数，无需手动清理
client = TestClient(app)
def test_get_books():
response = client.get("/books")
assert response.status_code == 200📌 关键点解释
-
@pytest.fixture(autouse=True)：该fixture会自动应用到模块中的每个测试函数，无需显式声明依赖。 -
yield之前的代码：在测试函数运行前执行（用于设置覆盖）。 -
yield：分割准备工作和清理工作的分界点，测试函数在这里执行。 -
yield之后的代码：在测试函数运行后执行（用于清理覆盖）。
🔧 完整示例（带依赖覆盖设置）
import pytest
from fastapi.testclient import TestClient
from main import app, get_db
# 模拟的数据库依赖
def mock_get_db():
yield "test_db_session"
# 自动清理 fixture
@pytest.fixture(autouse=True)
def setup_and_cleanup():
# 设置依赖覆盖
app.dependency_overrides[get_db] = mock_get_db
yield
# 清理覆盖
app.dependency_overrides.clear()
client = TestClient(app)
def test_read_main():
response = client.get("/books")
assert response.status_code == 200为什么需要清理？
如果多个测试文件共享同一个 app 实例（通常如此），前一个测试设置的 dependency_overrides 会一直保留，导致后续测试在不经意间使用了错误的依赖，产生难以调试的交叉污染。使用 autouse fixture 可以确保每个测试都从干净的依赖开始。
进阶：使用 fixture 的 scope 控制生命周期
如果你的测试文件只需要设置一次覆盖（比如所有测试共享同一个模拟依赖），可以将 scope="module" 或 scope="session"，并在模块结束或 session 结束时清理。但通常情况下，每个测试独立隔离更安全。
10.4 自定义文档内容：让 Swagger 说人话
FastAPI 自动生成的文档固然方便，但默认的摘要、描述往往过于简陋。你可以通过装饰器参数为接口加上标签、摘要、详细说明，甚至标记某些接口已废弃。
10.4.1 为路由添加 tags 和 summary
from fastapi import FastAPI
app = FastAPI(title="图书管理 API 增强版")
@app.get("/books", tags=["图书"], summary="获取所有图书",
description="返回当前系统收录的所有图书列表。")
def list_books():
"""内部文档（不会显示在摘要中）"""
return [...]
@app.get("/books/{book_id}", tags=["图书"], summary="获取指定图书")
def get_book(book_id: int):
...在 Swagger UI 中，tags 会将接口分组，summary 会作为标题显示，description 会显示在展开的详细信息区域。
10.4.2 描述与过时标记
@app.get("/old-books", tags=["图书"], summary="旧版接口", deprecated=True,
description="此接口已废弃，请改用 /books")
def old_list_books():
return {"msg": "请使用新接口"}在 Swagger 文档中，这个接口会显示为灰色并有“deprecated”标记，提醒调用方迁移。
完整项目代码汇总
main.py（包含被测试的应用及文档增强）
from fastapi import FastAPI, Depends, HTTPException
app = FastAPI(title="图书管理 API", description="用于管理图书的示例 API", version="0.1")
class Database:
def query_books(self):
return [
{"id": 1, "title": "数据密集型应用系统设计", "author": "马丁·科勒普曼"},
{"id": 2, "title": "重构", "author": "马丁·福勒"}
]
def get_db():
return Database()
@app.get("/books", tags=["图书"], summary="获取所有图书",
description="返回当前系统中的所有图书，不分页。")
def list_books(db: Database = Depends(get_db)):
return db.query_books()
@app.get("/books/{book_id}", tags=["图书"], summary="获取指定图书",
description="根据 ID 获取单本图书的详细信息。")
def get_book(book_id: int, db: Database = Depends(get_db)):
books = db.query_books()
for book in books:
if book["id"] == book_id:
return book
raise HTTPException(status_code=404, detail="图书不存在")
@app.get("/legacy-books", tags=["图书"], summary="旧版列表接口",
deprecated=True, description="已废弃，请使用 /books")
def legacy_list():
return {"msg": "此接口已停止服务"}tests/test_main.py
import sys
from pathlib import Path
# 将项目根目录（my_fatapi_project）加入 Python 模块搜索路径
sys.path.insert(0, str(Path(__file__).parent.parent))
from fastapi.testclient import TestClient
from main import app, get_db
# 模拟数据库
class FakeDatabase:
def query_books(self):
return [
{"id": 1, "title": "测试书", "author": "测试作者"}
]
def fake_get_db():
return FakeDatabase()
# 覆盖依赖
app.dependency_overrides[get_db] = fake_get_db
client = TestClient(app)
def test_list_books_with_fake():
response = client.get("/books")
assert response.status_code == 200
assert response.json()[0]["title"] == "测试书"
def test_get_book_not_found():
response = client.get("/books/999")
assert response.status_code == 404
def test_legacy_deprecated():
response = client.get("/legacy-books")
assert response.status_code == 200运行命令：
pytest打开 http://127.0.0.1:8000/docs 可查看自定义标签、摘要和废弃标记。
总结
-
TestClient：不用启动服务器就能测试接口，速度快、编写简单。
-
pytest：自动化测试管理，清晰的报告，便于集成到 CI/CD。
-
依赖覆盖：在测试中替换真实依赖（数据库、外部 API），保证测试隔离、快速、可重复。
-
文档增强：通过
tags、summary、description、deprecated让 API 文档更专业，降低沟通成本。
至此，你已经掌握了从接口开发、测试保障到文档输出的全流程技能，你的 FastAPI 应用不再是“裸奔”的代码，而是穿着铠甲、带着说明书的可靠产品。
第 11 章 模板与静态文件
到目前为止，我们构建的 API 都只返回冰冷的 JSON 数据。但在实际项目中，有时你需要直接返回一个网页给用户 — 比如管理后台的仪表盘、用户注册的确认页面、或者一份漂亮的报表。FastAPI 虽然主打 API，但它也提供了完整的静态文件服务和模板引擎集成，让你能轻松返回 HTML 页面。
11.1 静态文件服务
生活案例：你去一家公司办理业务，在前台填表时需要用到签字笔、计算器、饮水机。这些东西不是前台接待员临时去买的，而是公司早就买好放在公共用品柜里，任何人都能直接取用。
在 Web 应用中，CSS 样式表、JavaScript 脚本、图片、字体文件就是这些公共用品。它们不需要经过后端逻辑处理，直接“原样”交给浏览器就行。FastAPI 的 StaticFiles 就是那个公共用品柜 — 你只需指定一个本地文件夹，它就能自动把里面的文件暴露给外部访问。
为什么需要静态文件服务？
当你返回一个 HTML 页面时，页面里通常会引用 CSS 和 JS 文件：
<link rel="stylesheet" href="/static/style.css">
<script src="/static/app.js"></script>
<img src="/static/logo.png">浏览器看到这些标签后，会再次向服务器发起请求，分别获取 style.css、app.js 和 logo.png。如果没有静态文件服务，这些请求会返回 404，页面就会失去样式和交互功能。
代码实现
from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles
import uvicorn
app = FastAPI(title="静态文件服务演示")
# ---------- 挂载静态文件目录 ----------
# 将本地 "static" 文件夹映射到 URL "/static" 路径
# 例如：static/logo.png → http://127.0.0.1:8000/static/logo.png
app.mount("/static", StaticFiles(directory="static"), name="static")
# ---------- 普通接口 ----------
@app.get("/")
def read_root():
return {"msg": "访问 /static/文件名 查看静态资源"}
if __name__ == "__main__":
uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)项目目录结构
project/
├── main.py # 上面的代码
└── static/ # 静态文件目录
├── style.css
├── app.js
└── logo.png创建测试文件：
static/style.css:
body {
font-family: Arial, sans-serif;
background-color: #f0f8ff;
text-align: center;
padding-top: 50px;
}
h1 { color: #0066cc; }static/app.js:
console.log("静态文件加载成功！");
document.addEventListener("DOMContentLoaded", () => {
document.getElementById("msg").innerText = "JavaScript 已生效";
});测试
启动服务后访问：
-
http://127.0.0.1:8000/static/style.css→ 看到 CSS 内容 -
http://127.0.0.1:8000/static/app.js→ 看到 JS 内容 -
http://127.0.0.1:8000/static/logo.png→ 看到图片（如果放了）
关键点
-
app.mount()不是路由装饰器，而是将另一个 ASGI 应用“挂载”到指定路径。StaticFiles本身就是一个专门处理静态文件的 ASGI 应用。 -
directory="static"是本地文件夹路径，相对于main.py所在目录。 -
name="static"是内部标识名，不能与已有的路由名称冲突。
11.2 Jinja2 模板引擎
生活案例：你收到了一封银行的“信用卡账单”邮件。这封邮件的格式是统一的 — 银行的 Logo 在最上方，中间是你的姓名、消费明细、应还金额，底部是客服电话。模板就是那个空白表格，银行每天要做的事只是把你的姓名、卡号、金额等数据填入表格，然后发出去，而不是为每位客户重新设计一整封邮件。
Jinja2 就是 FastAPI 默认使用的模板引擎。它让你写出一个包含占位符的 HTML 文件（模板），然后后端把真实数据填入占位符，生成最终的 HTML 页面返回给浏览器。
安装依赖
pip install jinja2项目目录结构：
project/
├── main.py
├── static/ # 静态文件
│ └── style.css
└── templates/ # 模板文件
└── index.htmlmain.py:
from fastapi import FastAPI, Request
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
import uvicorn
app = FastAPI(title="Jinja2 模板演示")
# 挂载静态文件（让模板中的 CSS/JS 能正常加载）
app.mount("/static", StaticFiles(directory="static"), name="static")
# 配置模板引擎，指定模板文件存放的文件夹
templates = Jinja2Templates(directory="templates")
@app.get("/")
def read_root():
return {"msg": "访问 /page 查看网页"}
@app.get("/page")
def read_page(request: Request):
"""
返回一个渲染后的 HTML 页面。
request 参数必须传入，Jinja2 需要它来生成 URL。
"""
# 模板中需要的数据
context = {
"request": request, # 必须传 request
"title": "AI 数据治理助手", # 页面标题
"heading": "欢迎使用智能问答系统",
"features": [
{"name": "数据血缘分析", "desc": "追踪数据的来龙去脉"},
{"name": "元数据管理", "desc": "统一管理数据资产"},
{"name": "质量监控", "desc": "实时检测数据异常"},
],
"user_count": 1280,
"show_beta": True
}
return templates.TemplateResponse("index.html", context)
if __name__ == "__main__":
uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)templates/index.html:
<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<title>{{ title }}</title>
<!-- 引用静态文件 -->
<link rel="stylesheet" href="/static/style.css">
</head>
<body>
<h1>{{ heading }}</h1>
<p id="msg">正在加载...</p>
<h2>功能列表</h2>
<ul>
<!-- 循环遍历 features 列表 -->
{% for feature in features %}
<li>
<strong>{{ feature.name }}</strong>：{{ feature.desc }}
</li>
{% endfor %}
</ul>
<p>当前注册用户：<b>{{ user_count }}</b> 人</p>
<!-- 条件判断 -->
{% if show_beta %}
<p style="color: orange;">🚧 部分功能处于 Beta 测试阶段</p>
{% endif %}
<hr>
<footer>
<p>请求路径：{{ request.url.path }}</p>
</footer>
<script src="/static/app.js"></script>
</body>
</html>测试
启动服务后访问 http://127.0.0.1:8000/page，你会看到一个完整的网页，其中：
-
{{ title }}被替换成了"AI 数据治理助手" -
{{ heading }}被替换成了"欢迎使用智能问答系统" -
{% for feature in features %}循环生成了三个功能列表项 -
{% if show_beta %}因为值为True，显示 Beta 提示 -
{{ request.url.path }}输出了当前路径/page
Jinja2 核心语法速查
|
语法 |
作用 |
示例 |
|---|---|---|
|
|
输出变量的值 |
|
|
|
循环 |
|
|
|
条件判断 |
|
|
|
模板继承 |
见下文高级用法 |
|
|
模板注释，不会出现在 HTML 中 |
|
高级用法：模板继承
当多个页面共享相同的头部、导航栏、底部时，可以定义一个基础模板，让其他页面继承它。
templates/base.html:
<!DOCTYPE html>
<html>
<head>
<title>{% block title %}默认标题{% endblock %}</title>
<link rel="stylesheet" href="/static/style.css">
</head>
<body>
<nav>导航栏：<a href="/page">首页</a> | <a href="/about">关于</a></nav>
<hr>
{% block content %}{% endblock %}
<hr>
<footer>© 2025 AI 数据治理助手</footer>
</body>
</html>templates/about.html:
{% extends "base.html" %}
{% block title %}关于我们{% endblock %}
{% block content %}
<h1>关于我们</h1>
<p>这是一个演示 Jinja2 模板继承的页面。</p>
{% endblock %}@app.get("/about")
def about(request: Request):
return templates.TemplateResponse("about.html", {"request": request})总结
-
静态文件：CSS、JS、图片等不需要后端处理的文件，通过
app.mount("/static", StaticFiles(...))直接暴露。 -
Jinja2 模板：让你用 HTML + 占位符 + 循环/条件语法生成动态网页，核心是
Jinja2Templates(directory="templates")和TemplateResponse。 -
两者结合：模板负责页面结构，静态文件负责样式和交互，共同构成完整的 Web 页面。
通过本章，你的 FastAPI 应用不再只是“数据接口”，而是一个能直接返回漂亮网页的全栈 Web 服务。
第 12 章 使用 APIRouter 拆分项目
当你的 FastAPI 应用逐渐变大，把所有路由都塞在一个 main.py 里会变成一场灾难 — 几十个接口堆在一起，难以维护，团队协作时频繁冲突。FastAPI 提供了 APIRouter、子应用挂载、生命周期管理和配置管理四大工具，帮你把“大泥球”拆解成清晰、可复用的模块。本章将用生活化的案例和完整可运行的代码，让你彻底掌握大型项目的组织方法。
12.1 创建路由模块
生活案例：一家大型公司不可能所有人都挤在一间办公室里。公司会把员工分到不同部门 — 技术部、市场部、人事部 — 每个部门有自己独立的工位、独立的职责、独立的门牌号（比如“3 楼 301 技术部”）。前台接待员（主应用）根据访客要找的部门，直接引导到对应楼层和房间号。
在 FastAPI 中，APIRouter 就是这些“部门”。你可以把一组相关的接口（比如所有与用户相关的操作）放在一个单独的 Router 文件中，给它们统一的前缀（prefix）和标签（tags），最后在主应用中“登记”这个部门。
为什么要拆分？
-
代码组织清晰：每个
Router只负责一个功能模块（用户、订单、商品等）。 -
复用性强：同一个
Router可以挂载到多个应用。 -
避免冲突：不同模块可以有相同的路径，通过前缀区分（如
/users/me和/orders/me）。 -
团队协作友好：不同的人负责不同 Router，互不干扰。
项目结构示例
project/
├── main.py # 主应用
├── routers/
│ ├── __init__.py
│ ├── users.py # 用户模块
│ └── items.py # 商品模块
└── models/
└── ...代码实现
步骤一：创建路由模块 routers/users.py
from fastapi import APIRouter, HTTPException
# 创建 Router 实例
# prefix="/users" → 所有路由都会自动加上 /users 前缀
# tags=["users"] → 在 Swagger 文档中归入“users”分组
router = APIRouter(prefix="/users", tags=["用户模块"])
# 模拟数据库
fake_users = {
1: {"id": 1, "name": "张三", "email": "zhangsan@example.com"},
2: {"id": 2, "name": "李四", "email": "lisi@example.com"}
}
@router.get("/")
def list_users():
"""获取所有用户列表"""
return list(fake_users.values())
@router.get("/{user_id}")
def get_user(user_id: int):
"""根据 ID 获取单个用户"""
user = fake_users.get(user_id)
if not user:
raise HTTPException(status_code=404, detail="用户不存在")
return user
@router.post("/", status_code=201)
def create_user(name: str, email: str):
"""创建新用户"""
new_id = max(fake_users.keys()) + 1 if fake_users else 1
new_user = {"id": new_id, "name": name, "email": email}
fake_users[new_id] = new_user
return new_user步骤二：创建另一个路由模块 routers/items.py
from fastapi import APIRouter
router = APIRouter(prefix="/items", tags=["商品模块"])
fake_items = [
{"id": 1, "name": "机械键盘", "price": 399},
{"id": 2, "name": "降噪耳机", "price": 899}
]
@router.get("/")
def list_items():
return fake_items
@router.get("/{item_id}")
def get_item(item_id: int):
return next((item for item in fake_items if item["id"] == item_id), {"error": "商品不存在"})步骤三：主应用 main.py 中注册路由
from fastapi import FastAPI
from routers import users, items # 导入路由模块
import uvicorn
app = FastAPI(title="大型项目拆分演示")
# 登记各个“部门”，告诉前台接待员这些模块的存在
app.include_router(users.router) # 所有 /users/* 的请求交给 users 模块处理
app.include_router(items.router) # 所有 /items/* 的请求交给 items 模块处理
@app.get("/")
def root():
return {"message": "欢迎来到主应用！访问 /docs 查看接口"}
if __name__ == "__main__":
uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)效果验证：
-
启动服务后访问
http://127.0.0.1:8000/docs，你会看到两组接口分别列在“用户模块”和“商品模块”下。 -
访问
http://127.0.0.1:8000/users/→ 返回用户列表 -
访问
http://127.0.0.1:8000/items/→ 返回商品列表
关键点：
-
prefix让所有路由自动添加前缀，无需在每个@router.get("/{id}")里写/users/{id}。 -
tags控制 Swagger 文档分组。 -
include_router的顺序不影响路由匹配，但可以在一个主应用里加载多个路由器。
12.2 挂载子应用
生活案例：一座大型购物中心，一楼是超市，二楼是电影院，三楼是健身房。它们虽然在同一栋楼里，但各自有独立的收银系统、独立的会员卡、独立的宣传页。购物中心的大厅（主应用）只负责引导顾客到不同楼层（子应用），而不干预它们的内部运作。
FastAPI 的 app.mount() 允许你将另一个独立的 FastAPI（或其它 ASGI 应用）作为子应用挂载到主应用的某个路径下。子应用拥有自己完全独立的路由、文档、中间件，就像购物中心里独立运营的店铺。
适用场景
-
将一个大型项目拆分为多个微服务，但对外暴露统一入口。
-
集成第三方
ASGI应用（如静态文件、Jupyter Notebook 等）。 -
多个团队各自开发独立的子应用，最后组装。
代码实现
主应用 main.py
from fastapi import FastAPI
import uvicorn
app = FastAPI(title="购物中心（主应用）")
@app.get("/")
def main_home():
return {"msg": "欢迎来到购物中心！"}
# 创建一个独立的子应用（超市）
supermarket_app = FastAPI(title="超市子系统")
@supermarket_app.get("/")
def supermarket_home():
return {"msg": "欢迎来到超市"}
@supermarket_app.get("/products")
def supermarket_products():
return ["牛奶", "面包", "鸡蛋"]
# 创建另一个子应用（电影院）
cinema_app = FastAPI(title="电影院子系统")
@cinema_app.get("/")
def cinema_home():
return {"msg": "欢迎来到电影院"}
@cinema_app.get("/movies")
def cinema_movies():
return ["流浪地球3", "哪吒2"]
# 挂载子应用到主应用的路径上
app.mount("/supermarket", supermarket_app) # 访问 /supermarket 进入超市子应用
app.mount("/cinema", cinema_app) # 访问 /cinema 进入电影院子应用
if __name__ == "__main__":
uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)效果验证：
-
访问
http://127.0.0.1:8000/→ 主应用首页 -
访问
http://127.0.0.1:8000/supermarket/→ 超市子应用首页 -
访问
http://127.0.0.1:8000/supermarket/products→ 超市的商品列表 -
访问
http://127.0.0.1:8000/cinema/movies→ 电影院的电影列表
注意：
-
子应用自带独立的 OpenAPI 文档，可在
http://127.0.0.1:8000/supermarket/docs看到超市自己的文档。 -
主应用的
/docs不会显示子应用的路由，因为它们是完全隔离的。 -
app.mount()中的路径会作为前缀，子应用内部的/products映射到外部就是/supermarket/products。
12.3 应用生命周期管理
生活案例：开一家餐厅，早上开门前需要：打开炉灶、预热烤箱、准备食材。晚上关门后需要：清理厨房、关闭燃气、锁门。这些操作不是在客人的点单过程中进行的，而是在“营业开始”和“营业结束”两个特定时刻统一完成的。
FastAPI 提供了 lifespan 上下文管理器，让你在应用启动时执行初始化操作（如建立数据库连接池、加载模型），在应用关闭时执行清理操作（如释放连接、保存日志）。它替代了旧版的 @app.on_event("startup") 和 @app.on_event("shutdown")，使用更现代的 @asynccontextmanager 语法。
代码实现
from fastapi import FastAPI
from contextlib import asynccontextmanager
import uvicorn
# 模拟初始化资源
async def init_database():
print("🗄️ 正在创建数据库连接池...")
# 实际项目中会创建 SQLAlchemy 引擎等
return {"connection": "pool_ready"}
async def close_database(db):
print("🗄️ 正在关闭数据库连接池...")
# 实际项目中会释放连接
@asynccontextmanager
async def lifespan(app: FastAPI):
"""
应用生命周期管理器：
- yield 之前的代码在应用启动时执行（只执行一次）
- yield 之后的代码在应用关闭时执行（只执行一次）
"""
# 启动阶段：加载资源
print("🚀 应用正在启动...")
db = await init_database()
# 可以将资源存储到 app.state 中，供所有请求使用
app.state.db = db
print("✅ 应用启动完成，开始接收请求\n")
yield # 应用运行期间在此暂停
# 关闭阶段：清理资源
print("\n🛑 应用正在关闭...")
await close_database(app.state.db)
print("👋 应用已安全关闭")
# 创建 FastAPI 应用时传入 lifespan
app = FastAPI(lifespan=lifespan, title="生命周期管理演示")
@app.get("/")
def read_root():
# 可以从 app.state 中访问启动时加载的资源
db_status = "已连接" if app.state.db else "未连接"
return {"message": f"数据库状态: {db_status}"}
if __name__ == "__main__":
uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)运行效果（观察终端输出）：
🚀 应用正在启动...
🗄️ 正在创建数据库连接池...
✅ 应用启动完成，开始接收请求
（此时你可以发送请求）
🛑 应用正在关闭... （按 Ctrl+C 时）
🗄️ 正在关闭数据库连接池...
👋 应用已安全关闭关键点：
-
@asynccontextmanager必须用在异步函数上，且必须包含yield。 -
yield之前的代码在应用启动时执行（类似旧版的startup），之后的代码在应用关闭时执行（类似旧版的shutdown）。 -
可以通过
app.state在整个应用生命周期内共享对象（如数据库连接池、配置等），它类似于 Flask 的g对象，但生命周期更长。 -
推荐使用 lifespan 替代旧的
on_event，因为它能更好地支持测试和依赖注入。
12.4 配置管理
生活案例：你有一套衣服，但春夏秋冬要穿的厚度不同。你不会把羽绒服焊在身上，而是根据天气预报（环境变量）选择当天穿什么。同样，一个 Web 应用在开发环境（本地）可能使用 SQLite 数据库和调试模式，而在生产环境（服务器）则需要 PostgreSQL、更高的安全等级和日志级别。把这些差异硬编码在代码里是噩梦。
FastAPI 推荐使用 Pydantic-settings 的 BaseSettings 从环境变量（或 .env 文件）中自动加载配置，实现“一套代码，多套配置”。
安装依赖
pip install pydantic-settings代码实现
步骤一：创建配置文件 config.py
from pydantic_settings import BaseSettings
from functools import lru_cache
class Settings(BaseSettings):
"""应用配置类，自动从环境变量或 .env 文件加载"""
# 应用基础配置
app_name: str = "我的 FastAPI 应用" # 默认值，可被环境变量覆盖
debug: bool = False
# 数据库配置
database_url: str = "sqlite:///./default.db"
database_pool_size: int = 5
# JWT 配置
secret_key: str = "default-secret-change-me"
algorithm: str = "HS256"
access_token_expire_minutes: int = 30
# CORS 配置
allowed_origins: list = ["http://localhost:3000"]
class Config:
# 指定 .env 文件路径，如果存在则读取
env_file = ".env"
env_file_encoding = "utf-8"
# 使用 lru_cache 缓存配置，避免每次请求都重新读取
@lru_cache()
def get_settings() -> Settings:
return Settings()步骤二：创建 .env 文件（与 main.py 同级目录）
# 开发环境配置示例
APP_NAME=数据治理助手-开发版
DEBUG=true
DATABASE_URL=sqlite:///./dev.db
SECRET_KEY=dev-secret-12345
ALLOWED_ORIGINS=["http://localhost:3000", "http://localhost:8080"]步骤三：在主应用中使用配置
from fastapi import FastAPI, Depends
from config import get_settings, Settings
import uvicorn
app = FastAPI(title="配置管理演示")
# 依赖注入：让路径函数可以获取配置对象
@app.get("/info")
def get_app_info(settings: Settings = Depends(get_settings)):
return {
"app_name": settings.app_name,
"debug": settings.debug,
"database": settings.database_url,
"token_expire_minutes": settings.access_token_expire_minutes
}
@app.get("/cors")
def get_cors(settings: Settings = Depends(get_settings)):
return {"allowed_origins": settings.allowed_origins}
if __name__ == "__main__":
import os
# 可以手动设置环境变量（覆盖 .env 文件中的值）
# os.environ["APP_NAME"] = "临时测试环境"
uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)效果验证：
-
启动服务后访问
http://127.0.0.1:8000/info，返回的app_name将是.env中配置的"数据治理助手-开发版"。 -
修改
.env中的DEBUG=true，重启服务，debug字段变为true。 -
生产环境可以通过设置真实的环境变量（而非
.env文件）来覆盖，例如：
export APP_NAME="生产环境" && python main.py关键点：
-
BaseSettings会自动读取同名的环境变量（不区分大小写），例如APP_NAME对应类中的app_name字段。 -
支持类型转换：字段声明为
int则自动将字符串转为整数，list类型可以自动解析 JSON 格式的环境变量（如["a","b"]）。 -
.env文件适合开发环境，生产环境建议使用真实的系统环境变量或配置中心（如 Kubernetes ConfigMap、HashiCorp Vault）。 -
通过
Depends(get_settings)可以在任何接口中获取配置，无需全局变量。
12.5 综合案例：整合所有拆分技术
最后，我们将上述四个技术整合到一个项目中，展示一个生产级的项目结构。
项目结构：
project/
├── main.py # 主应用，组装一切
├── config.py # 配置管理
├── routers/
│ ├── __init__.py
│ ├── users.py
│ └── items.py
├── subapps/ # 子应用
│ └── admin.py
└── .env # 环境配置文件config.py
from pydantic_settings import BaseSettings
from functools import lru_cache
class Settings(BaseSettings):
"""应用配置类，自动从环境变量或 .env 文件加载"""
# 应用基础配置
app_name: str = "我的 FastAPI 应用" # 默认值，可被环境变量覆盖
debug: bool = False
# 数据库配置
database_url: str = "sqlite:///./default.db"
database_pool_size: int = 5
# JWT 配置
secret_key: str = "default-secret-change-me"
algorithm: str = "HS256"
access_token_expire_minutes: int = 30
# CORS 配置
allowed_origins: list = ["http://localhost:3000"]
class Config:
# 指定 .env 文件路径，如果存在则读取
env_file = ".env"
env_file_encoding = "utf-8"
# 使用 lru_cache 缓存配置，避免每次请求都重新读取
@lru_cache()
def get_settings() -> Settings:
return Settings()routers/users.py
from fastapi import APIRouter, HTTPException
# 创建 Router 实例
# prefix="/users" → 所有路由都会自动加上 /users 前缀
# tags=["users"] → 在 Swagger 文档中归入“users”分组
router = APIRouter(prefix="/users", tags=["用户模块"])
# 模拟数据库
fake_users = {
1: {"id": 1, "name": "张三", "email": "zhangsan@example.com"},
2: {"id": 2, "name": "李四", "email": "lisi@example.com"}
}
@router.get("/")
def list_users():
"""获取所有用户列表"""
return list(fake_users.values())
@router.get("/{user_id}")
def get_user(user_id: int):
"""根据 ID 获取单个用户"""
user = fake_users.get(user_id)
if not user:
raise HTTPException(status_code=404, detail="用户不存在")
return user
@router.post("/", status_code=201)
def create_user(name: str, email: str):
"""创建新用户"""
new_id = max(fake_users.keys()) + 1 if fake_users else 1
new_user = {"id": new_id, "name": name, "email": email}
fake_users[new_id] = new_user
return new_userrouters/items.py
from fastapi import APIRouter
router = APIRouter(prefix="/items", tags=["商品模块"])
fake_items = [
{"id": 1, "name": "机械键盘", "price": 399},
{"id": 2, "name": "降噪耳机", "price": 899}
]
@router.get("/")
def list_items():
return fake_items
@router.get("/{item_id}")
def get_item(item_id: int):
return next((item for item in fake_items if item["id"] == item_id), {"error": "商品不存在"})subapps/admin.py（一个简单的后台子应用）
from fastapi import FastAPI
admin_app = FastAPI(title="后台管理子系统")
@admin_app.get("/")
def admin_home():
return {"msg": "后台管理首页"}
@admin_app.get("/stats")
def admin_stats():
return {"users": 128, "items": 56}main.py
from fastapi import FastAPI, Depends
from contextlib import asynccontextmanager
from routers import users, items
from subapps.admin import admin_app
from config import get_settings, Settings
import uvicorn
@asynccontextmanager
async def lifespan(app: FastAPI):
# 启动时从配置读取数据库 URL 等信息，建立连接池
settings = get_settings()
print(f"🚀 应用启动，数据库: {settings.database_url}")
yield
print("🛑 应用关闭")
app = FastAPI(lifespan=lifespan, title="整合案例")
# 注册路由模块
app.include_router(users.router)
app.include_router(items.router)
# 挂载子应用
app.mount("/admin", admin_app)
@app.get("/config-test")
def test_config(settings: Settings = Depends(get_settings)):
return {"app_name": settings.app_name, "debug": settings.debug}
if __name__ == "__main__":
uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)现在，你的应用已经具备：
-
✅ 模块化的路由拆分（用户、商品）
-
✅ 独立的后台子应用
-
✅ 启动/关闭生命周期管理
-
✅ 灵活的多环境配置
通过本章的学习，你已经从单文件脚本迈入了真正的工程化开发阶段，可以自信地构建和维护中大型 FastAPI 项目了。
第 13 章 WebSocket 与其他常用特性
本章涵盖 FastAPI 中几个重要的扩展能力：WebSocket 实时双向通信、连接管理与广播、Cookie 与 Header 操作、数据库集成模式，以及一些实用的补充特性。掌握这些内容后，你将具备构建实时应用、处理客户端状态、集成数据库和灵活控制响应格式的能力。
13.1 WebSocket 基础
生活案例：HTTP 请求就像你给朋友写信：你寄出一封信，等待对方回信，一次一问一答。而 WebSocket 就像你给朋友打了一个电话 — 拨通后，双方可以随时说话，无需等待对方说完才能开口，而且通话一直保持，直到主动挂断。
在 Web 应用中，WebSocket 通常用于：
-
在线聊天室
-
实时股票行情推送
-
多人协作编辑
-
游戏实时对战
FastAPI 原生支持 WebSocket，你只需用 @app.websocket 装饰器定义一个 WebSocket 路由。
代码实现
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
import uvicorn
app = FastAPI(title="WebSocket 基础演示")
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
"""
一个简单的 WebSocket 回声服务：
接收客户端发来的消息，原样返回。
"""
# 1. 接受连接（握手）
await websocket.accept()
print("🔗 客户端已连接")
try:
while True:
# 2. 接收文本消息
data = await websocket.receive_text()
print(f"📩 收到消息: {data}")
# 3. 发送回复
await websocket.send_text(f"你说: {data}")
except WebSocketDisconnect:
print("🔌 客户端断开连接")
except Exception as e:
print(f"⚠️ 异常: {e}")
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=8000)测试 WebSocket
你可以用 FastAPI 自带的 /docs 无法测试 WebSocket，需要用以下方式：
pip install websockets方法一：在浏览器控制台中使用 JavaScript
打开 http://127.0.0.1:8000/（可以是任何页面），按 F12 打开控制台，粘贴以下代码：
const ws = new WebSocket("ws://127.0.0.1:8000/ws");
ws.onopen = () => {
console.log("连接成功");
ws.send("你好，服务器！");
};
ws.onmessage = (event) => {
console.log("收到服务器消息:", event.data);
};
ws.onclose = () => {
console.log("连接关闭");
};
// 发送多条消息
setTimeout(() => ws.send("第二条消息"), 1000);方法二：编写 Python 客户端
安装 websockets 库：
pip install websockets创建 ws_client.py：
import asyncio
import websockets
async def test():
async with websockets.connect("ws://127.0.0.1:8000/ws") as ws:
await ws.send("Hello")
response = await ws.recv()
print(f"收到回复: {response}")
asyncio.run(test())运行：
python ws_client.py注意：在运行上述代码时，要确保后端服务处于运行状态。
期望输出：
收到回复: 你说: Hello13.2 连接管理
生活案例：你加入一个群聊，你发的消息所有群成员都能看到。服务器需要维护一个“在线成员列表”，当有人发消息时，遍历列表，把消息推送给每个人。
代码实现
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from typing import List
import uvicorn
app = FastAPI(title="WebSocket 聊天室")
# 维护所有活跃的 WebSocket 连接
connected_clients: List[WebSocket] = []
@app.websocket("/ws/chat")
async def chat_endpoint(websocket: WebSocket):
await websocket.accept()
# 新用户加入，添加到列表
connected_clients.append(websocket)
client_id = id(websocket)
print(f"🔗 用户 {client_id} 加入，当前在线 {len(connected_clients)} 人")
# 广播加入消息
for client in connected_clients:
if client != websocket:
await client.send_text(f"系统消息: 用户 {client_id} 加入了聊天室")
try:
while True:
data = await websocket.receive_text()
# 广播给所有客户端（包括自己）
for client in connected_clients:
await client.send_text(f"用户 {client_id} 说: {data}")
except WebSocketDisconnect:
# 用户离开，从列表中移除
connected_clients.remove(websocket)
print(f"🔌 用户 {client_id} 离开，当前在线 {len(connected_clients)} 人")
# 广播离开消息
for client in connected_clients:
await client.send_text(f"系统消息: 用户 {client_id} 离开了聊天室")
except Exception as e:
print(f"⚠️ 异常: {e}")
if websocket in connected_clients:
connected_clients.remove(websocket)
@app.get("/status")
def get_status():
return {"online_users": len(connected_clients)}
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=8000)测试：
-
打开多个浏览器标签页，在每个控制台中连接
ws://127.0.0.1:8000/ws/chat，发送消息。 -
可以看到所有标签页都能收到广播消息。
-
关闭某个标签页，其他标签页会收到“用户离开”提示。
13.3 Cookie 与 Header 操作
生活案例
-
Cookie：你去图书馆，管理员在你手上盖了一个章（Cookie），之后进出不用重复登记。浏览器每次请求都会自动带上 Cookie，服务器根据它识别你的身份或偏好。
-
Header：你在快递单上填的“寄件人备注”就是自定义请求头，供快递公司内部系统读取，不影响包裹本身的内容。
FastAPI 提供了 Cookie() 和 Header() 辅助函数，可以像路径参数一样直接提取这些值。
代码实现
from fastapi import FastAPI, Cookie, Header, Response
from typing import Optional
import uvicorn
app = FastAPI(title="Cookie 与 Header 操作演示")
# ---------- 读取 Cookie ----------
@app.get("/read-cookie")
def read_cookie(session_id: Optional[str] = Cookie(None)):
"""
从请求的 Cookie 中读取 session_id。
如果不存在，则返回提示。
"""
if session_id:
return {"session_id": session_id}
else:
return {"msg": "未找到 session_id Cookie"}
# ---------- 设置 Cookie ----------
@app.get("/set-cookie")
def set_cookie(response: Response):
"""在响应中设置一个 Cookie"""
response.set_cookie(
key="session_id",
value="abc-123-xyz",
max_age=3600, # 有效期 1 小时（秒）
httponly=True, # 禁止 JavaScript 读取，防 XSS
samesite="lax" # 防 CSRF
)
return {"msg": "Cookie 已设置"}
# ---------- 读取自定义请求头 ----------
@app.get("/read-header")
def read_header(
x_request_id: Optional[str] = Header(None),
user_agent: Optional[str] = Header(None)
):
"""
读取自定义请求头 X-Request-Id 和标准头 User-Agent。
"""
return {
"x_request_id": x_request_id or "未提供",
"user_agent": user_agent or "未提供"
}
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=8000)测试 Cookie：
-
访问
http://127.0.0.1:8000/set-cookie，浏览器会收到一个Set-Cookie响应头。 -
再访问
http://127.0.0.1:8000/read-cookie，浏览器会自动带上刚才设置的 Cookie。
测试 Header：
curl -H "X-Request-Id: req-999" http://127.0.0.1:8000/read-header13.4 数据库集成模式
FastAPI 不内置 ORM，但可以灵活集成任何异步或同步数据库库。本节展示使用 SQLAlchemy 异步引擎 + 依赖注入 的标准模式。
安装依赖
pip install sqlalchemy[asyncio] asyncpg如果使用 SQLite（零配置，方便本地测试）：
pip install aiosqlite完整代码（使用 SQLite 异步驱动）
from fastapi import FastAPI, Depends
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy import select
import uvicorn
from contextlib import asynccontextmanager
# ---------- 数据库配置 ----------
DATABASE_URL = "sqlite+aiosqlite:///./test.db"
engine = create_async_engine(DATABASE_URL, echo=True)
async_session = async_sessionmaker(engine, expire_on_commit=False)
# ---------- ORM 基类 ----------
class Base(DeclarativeBase):
pass
# ---------- 定义模型 ----------
class User(Base):
__tablename__ = "users"
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str]
email: Mapped[str]
# ---------- 生命周期管理 ----------
@asynccontextmanager
async def lifespan(app: FastAPI):
# 启动时创建表
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
yield
# 关闭时释放数据库连接（可选）
await engine.dispose()
# ---------- FastAPI 应用 ----------
app = FastAPI(title="数据库集成演示", lifespan=lifespan) # 直接传入 lifespan
# ---------- 依赖：获取数据库会话 ----------
async def get_db():
async with async_session() as session:
yield session
# ---------- 创建用户接口 ----------
@app.post("/users")
async def create_user(name: str, email: str, db: AsyncSession = Depends(get_db)):
user = User(name=name, email=email)
db.add(user)
await db.commit()
await db.refresh(user)
return {"id": user.id, "name": user.name}
# ---------- 查询所有用户 ----------
@app.get("/users")
async def list_users(db: AsyncSession = Depends(get_db)):
result = await db.execute(select(User))
users = result.scalars().all()
return [{"id": u.id, "name": u.name, "email": u.email} for u in users]
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=8000)关键点：
-
异步引擎：
create_async_engine支持异步 I/O。 -
依赖注入：
get_db使用yield自动在请求结束后关闭会话。 -
数据库初始化：
@app.on_event("startup")确保服务启动时自动建表（生产环境建议使用 Alembic 迁移工具）。
13.5 其他补充特性
13.5.1 对标准库 dataclasses 的支持
如果不想使用 Pydantic，FastAPI 也支持 Python 内置的 dataclass：
from dataclasses import dataclass
from fastapi import FastAPI
app = FastAPI()
@dataclass
class Item:
name: str
price: float
@app.post("/items")
def create_item(item: Item):
return {"name": item.name, "price": item.price}局限性：dataclass 没有内置校验，建议 Pydantic 优先。
13.5.2 自定义状态码与响应描述
你可以在路径装饰器中直接指定默认状态码和响应描述，让文档更清晰：
@app.post(
"/products",
status_code=201, # 默认返回 201 Created
response_description="返回新创建的商品信息"
)
def create_product():
return {"id": 1}13.5.3 使用 response_class 返回不同格式
FastAPI 默认返回 JSON，但你可以指定返回纯文本、HTML 或其他格式：
from fastapi.responses import PlainTextResponse, HTMLResponse
@app.get("/text", response_class=PlainTextResponse)
def get_text():
return "这是一段纯文本"
@app.get("/html", response_class=HTMLResponse)
def get_html():
return "<h1>这是 HTML 标题</h1>"总结
-
WebSocket：让你的应用支持实时双向通信，适合聊天、推送等场景。
-
连接管理：通过列表维护活跃连接，实现广播和在线统计。
-
Cookie 与 Header：
Cookie()和Header()让你轻松读取客户端信息，Response.set_cookie设置 Cookie。 -
数据库集成：使用 SQLAlchemy 异步引擎 +
yield依赖注入实现安全、高效的数据库操作。 -
补充特性：
dataclass支持、自定义状态码、response_class等，让你在各种场景下都能灵活应对。
结束语
至此，你已系统掌握了 FastAPI 从基础路由到大型项目管理的全部核心知识。每学完一个阶段，建议立即动手实现一个小项目（如待办事项 API、聊天室、用户登录系统）。理论结合实践，才是最快的进阶之路。
AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念，把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起，为开发者提供从开发、训练到部署的一站式体验。
更多推荐
10
0- 0
已为社区贡献25条内容
所有评论(0)