FastAPI 基础知识

最后更新:2026-03-26

前置知识:Pydantic详解

1)Pydantic 是什么?

Pydantic 是 Python 生态中最常用的数据验证与序列化库:你用类型注解描述数据结构,Pydantic 负责在运行时做验证、类型转换、错误报告与 JSON Schema 生成。

2)为什么使用 Pydantic?

  • 类型驱动:直接复用 Python 类型提示,和 IDE、mypy/pyright 协同良好。
  • 高性能:核心验证引擎 pydantic-core 用 Rust 实现,速度在同类库中很有竞争力。
  • 序列化能力强:统一支持导出 Python 对象、JSON 兼容对象与 JSON 字符串。
  • JSON Schema 生成:模型可直接生成 JSON Schema(2020-12),天然适配 OpenAPI 3.1。
  • 严格/宽松双模式:可做智能转换(默认)也可强校验(strict=True)。
  • 可扩展:验证器、序列化器、TypeAdapter、自定义类型协议都很完善。
  • 生态成熟:FastAPI、SQLModel、Django Ninja、LangChain 等大量项目依赖。

3)Pydantic 解决了哪些工程痛点?

  • 外部输入不可信:自动拦截脏数据并定位具体字段错误。
  • 类型漂移:把字符串数字、日期字符串等按规则转换为目标类型。
  • 接口契约不稳定:通过模型统一入参/出参,降低前后端沟通成本。
  • 文档同步困难:模型即文档,Schema 自动生成。

5)Pydantic v2 必须掌握的 API

  • model_validate(obj)输入 Python 对象(如 dict、ORM 对象),执行验证和类型转换后,返回模型实例。适合服务内已经拿到 Python 数据的场景。
  • model_validate_json(json_data)输入 JSON 字符串/字节串,一步完成“JSON 解析 + 校验 + 转换”。常用于高性能入口(HTTP 原始报文、消息队列消息体)。
  • model_dump(...):导出为 Python 字典。常用参数:exclude_unset(只导出已设置字段)、exclude_defaults(排除默认值字段)、exclude_none(排除 None 字段)、mode="json"(导出 JSON 兼容值)。
  • model_dump_json(...):直接导出为 JSON 字符串,适合写日志、发消息、直接返回文本化载荷。
  • model_json_schema():生成模型对应的 JSON Schema,是 OpenAPI 文档、前端自动表单和契约校验工具链的基础。

6)序列化三种常见形态

from datetime import datetime
from pydantic import BaseModel


class Meeting(BaseModel):
    when: datetime
    where: bytes
    why: str = "No idea"


m = Meeting(when="2020-01-01T12:00", where="home")

# 形态 1:Python 对象字典(保留 Python 原生类型)
# exclude_unset=True:只导出“显式传入/被设置过”的字段
# 例如 datetime 仍是 datetime 对象、bytes 仍是 bytes
print(m.model_dump(exclude_unset=True))

# 形态 2:JSON 兼容字典(仍是 dict,但值已可 JSON 化)
# mode="json":把 datetime/bytes/UUID 等转换为 JSON 友好值
# exclude={"where"}:排除指定字段
print(m.model_dump(mode="json", exclude={"where"}))

# 形态 3:JSON 字符串(可直接落日志/消息队列/HTTP)
# exclude_defaults=True:排除“等于默认值”的字段,减少冗余
print(m.model_dump_json(exclude_defaults=True))

7)严格模式 vs 宽松模式

  • 宽松模式(默认):更偏业务友好,允许合理类型转换。
  • 严格模式:更偏安全与一致性,不做隐式转换。

实践建议:外部入口(支付、风控、权限)可偏严格;内部服务交互可按业务选择宽松/严格策略。

8)除了 BaseModel,还有哪些建模方式?

  • BaseModel:最常用,功能最完整。
  • pydantic.dataclasses.dataclass:给 dataclass 增加验证能力。
  • TypeAdapter:对任意类型(如 TypedDictlist[int])做验证/序列化。
  • validate_call:为函数调用参数做运行时验证。

9)TypeAdapter 的典型场景

from typing_extensions import TypedDict
from pydantic import TypeAdapter


class Query(TypedDict):
    page: int
    size: int


adapter = TypeAdapter(Query)
data = adapter.validate_python({"page": "1", "size": "20"})
print(data)
print(adapter.json_schema())

当你不想额外定义 BaseModel,但仍要验证类型时,TypeAdapter 很高效。

10)自定义验证与序列化

先记住一句话:验证器(Validator)用于“输入阶段”,序列化器(Serializer)用于“输出阶段”。

  • field_validator:字段级规则,适合单字段校验/修正(如用户名格式)。
  • model_validator:模型级规则,适合跨字段一致性(如密码与确认密码一致)。
  • field_serializer:输出阶段格式化/脱敏(如时间格式化、手机号打码)。
from datetime import datetime, timezone

from pydantic import BaseModel, field_serializer, field_validator, model_validator


class UserProfile(BaseModel):
    username: str
    password: str
    confirm_password: str
    created_at: datetime
    phone: str

    # 1) 字段级验证器:校验并修正单字段输入
    @field_validator("username")
    @classmethod
    def normalize_username(cls, value: str) -> str:
        if " " in value:
            raise ValueError("用户名不能包含空格")
        return value.strip().title()

    # “now”转时间 + 时区修正到 UTC
    @field_validator("created_at", mode="before")
    @classmethod
    def parse_created_at(cls, value: object) -> datetime:
        if isinstance(value, str) and value.lower() == "now":
            return datetime.now(timezone.utc)

        if isinstance(value, str):
            # Python 3.13 推荐:fromisoformat 解析 ISO8601 字符串
            value = datetime.fromisoformat(value)

        if isinstance(value, datetime):
            # 统一输出为 UTC 感知时间
            if value.tzinfo is None:
                return value.replace(tzinfo=timezone.utc)
            return value.astimezone(timezone.utc)

        raise TypeError("created_at 必须是 datetime 或 ISO8601 字符串")

    # 2) 模型级验证器:处理跨字段业务规则
    @model_validator(mode="after")
    def check_passwords_match(self) -> "UserProfile":
        if self.password != self.confirm_password:
            raise ValueError("两次密码输入不一致")
        return self

    # 3) 字段级序列化器:控制输出给前端的格式
    @field_serializer("created_at")
    def serialize_created_at(self, dt: datetime) -> str:
        return dt.strftime("%Y-%m-%d %H:%M:%S %Z")

    @field_serializer("phone")
    def serialize_phone(self, value: str) -> str:
        if len(value) >= 7:
            return f"{value[:3]}****{value[-4:]}"
        return "****"


payload = {
    "username": "alice",
    "password": "secret123",
    "confirm_password": "secret123",
    "created_at": "now",
    "phone": "13812345678",
}

user = UserProfile.model_validate(payload)
print(user.model_dump())
print(user.model_dump_json())

面试回答可总结:field_validator 管单字段输入质量,model_validator 管跨字段业务一致性,field_serializer 管输出格式与脱敏。

11)Pydantic 与 FastAPI 的关系

  • 请求体校验:非法输入自动 422。
  • 响应模型过滤:避免敏感字段泄漏。
  • OpenAPI 生成:模型直接驱动文档。

面试可总结:FastAPI 的“自动化体验”很大程度来自 Pydantic。

0. FastAPI 高频面试题

基础

Q1:FastAPI 是什么?核心优势是什么?

FastAPI 是基于 ASGI 的现代 Python API 框架,围绕“类型注解驱动开发”。面试可概括为四点:高性能自动校验自动文档异步友好。底层依赖 Starlette(Web 能力)+ Pydantic(数据校验)。

Q2:FastAPI、Flask、Django 各适合什么场景?

  • Flask:轻量灵活,适合小型服务或需要高度自定义的项目。
  • Django:全家桶,适合后台管理、传统 Web 站点。
  • FastAPI:API-first、前后端分离、微服务、AI 推理服务最常见。

面试话术:如果主要目标是高质量 API 交付速度 + 类型安全,我优先 FastAPI。

Q3:最小可运行 FastAPI 应用?

from fastapi import FastAPI

app = FastAPI()


@app.get("/")
async def root() -> dict[str, str]:
    return {"message": "ok"}

Q4:路径参数 vs 查询参数怎么区分?

  • 路径参数:资源定位(/users/{user_id})。
  • 查询参数:过滤/分页/排序(?page=1&size=20)。
from typing import Annotated
from fastapi import FastAPI, Path, Query

app = FastAPI()


@app.get("/users/{user_id}")
async def read_user(
    user_id: Annotated[int, Path(gt=0)],
    keyword: Annotated[str | None, Query(max_length=50)] = None,
):
    return {"user_id": user_id, "keyword": keyword}

Q5:Pydantic 在 FastAPI 里做了什么?

三件事:验证输入约束输出生成 Schema。你写模型,FastAPI 自动完成校验与文档,错误返回标准 422 结构。

Q6:什么时候用 async def,什么时候用 def

规则很实用:涉及数据库/HTTP/文件等 I/O,用 async def;纯 CPU 密集计算通常用 def(必要时丢到任务队列或进程池)。避免在 async def 里直接跑阻塞 I/O。

Q7:文档为什么“自动出现”?可否定制?

因为 FastAPI 将路由、参数、模型自动汇总为 OpenAPI。默认 /docs/redoc,可通过标题、描述、tags、response_model 等持续增强可读性。

进阶

Q8:依赖注入(Depends)有什么价值?

复用(鉴权、分页、会话)、解耦(业务与基础设施分离)、可测试(可覆盖依赖)。生产代码中,依赖往往形成“应用级→路由级→路径级”分层。

Q9:数据库连接的官方推荐实践?

使用依赖提供会话(每请求一个会话),使用 lifespan 管理应用级资源,数据库结构变更使用 Alembic 迁移。

Q11:错误处理该怎么设计?

业务错误用 HTTPException;系统级异常可全局处理器统一封装。推荐返回稳定错误码结构(如 code/message/request_id),便于前端和日志追踪。

Q12:BackgroundTasks 适合做什么?

结论:适合轻量后置任务(审计日志、通知、事件上报),不适合 CPU 重计算或强可靠任务。

“同进程”是什么意思? 就是后台任务和你的 FastAPI 请求处理跑在同一个进程里:响应先返回,随后由同一进程继续执行任务。

  • 优点:零额外基础设施、开发成本低、可直接复用当前进程上下文。
  • 风险 1(资源争抢):CPU 密集任务会抢占同进程 CPU,拖慢后续请求。
  • 风险 2(隔离性弱):任务异常可能影响主服务稳定性。
  • 风险 3(可靠性有限):服务重启时,进行中的同进程任务可能中断且丢失。

面试回答建议BackgroundTasks 用于“轻、短、可丢一点”的任务;需要重试、持久化、失败恢复、水平扩展时,使用 ARQ/Celery 这类异进程任务队列(通常基于 Redis/RabbitMQ)。

对比项 BackgroundTasks(同进程) ARQ / Celery(异进程)
任务重量 轻量后置任务 重任务 / 批处理 / 长任务
隔离性 弱(共享同一进程资源) 强(主服务与 Worker 隔离)
可靠性 一般(重启可能丢失) 高(支持持久化、重试、恢复)
落地复杂度 低(开箱即用) 中高(需消息中间件与 Worker)

Q13:CORS 与中间件常见坑?

allow_credentials=True,就不能配 allow_origins=["*"];要显式白名单。中间件顺序是“洋葱模型”,后加的在外层。

Q14:认证授权怎么落地(OAuth2/JWT)?

  1. 登录:校验用户名/密码,成功后签发 access_token(JWT)
  2. 带 Token 访问:客户端在请求头传 Authorization: Bearer <token>
  3. 解析 Token(当前用户依赖):在依赖里验证签名、过期时间、issuer/audience,拿到用户身份(如 sub)。
  4. 授权(权限依赖):基于角色/权限(RBAC)判断是否允许访问当前接口。

工程实践:把“当前用户依赖”和“权限依赖”分开。前者负责“身份正确性”,后者负责“权限策略”,这样可组合、可复用、可单测。

面试一句话版本:先登录签发 JWT,再通过“当前用户依赖”完成认证,通过“权限依赖”完成授权,二者解耦以保证可维护性与可测试性。

高级

Q16:FastAPI 为什么“快”?

快来自异步 I/O + ASGI 栈:Uvicorn 负责网络与事件循环,Starlette 负责 Web 基础设施,FastAPI 负责类型驱动的参数解析和文档生成。高并发 I/O 场景优势明显。

Q18:lifespan 到底是什么?

一句话lifespan 是应用级“启动/关闭钩子”,用于启动时初始化、关闭时清理。它把生命周期逻辑集中到一个异步上下文中,语义更完整,适合 DB 引擎、缓存客户端、模型句柄等资源管理。

1. FastAPI 并发与 async / await

Q1:为什么在 FastAPI 中,即使函数内部没有 await,也常建议优先使用 async def

通常在 Python 中,如果你不打算在函数内部使用 await,我们会倾向于定义普通的 def。但在 FastAPI 的实践规范中,如果你确定这个函数不需要与数据库、API 或文件系统进行耗时的通信,那么即使没有 await,官方依然建议优先使用 async def

1. 线程开销与性能

  • def:FastAPI 会将其放入线程池执行,避免阻塞主流程,但会引入线程调度开销。
  • async def:在事件循环中直接运行;若无耗时 I/O,通常执行更轻量。

2. 框架互操作性

  • FastAPI 的中间件和依赖体系大量基于异步模型。
  • 统一使用 async def,未来需要加 await 时无需改函数签名。

Q2:asyncio、Uvicorn、Starlette 三者是如何协作完成 FastAPI 并发执行流程的?

1. 核心角色定义

  • asyncio:它是 Python 标准库提供的“心脏”。它负责管理一个事件循环(Event Loop),像一个永不停歇的调度员,决定现在该处理哪个任务。
  • Uvicorn:它是一个 ASGI 服务器。它的工作是监听网络端口(如 8000),接收原始的 HTTP 请求,并将其包装成一种标准格式(ASGI 协议),交给上面的框架处理。
  • Starlette:它是 FastAPI 底层的 Web 工具包。它负责具体的业务逻辑:路由匹配、异常处理、Session 管理等。FastAPI 其实是在 Starlette 之上加了一层能力(如 Pydantic 校验和自动文档)。

2. 底层原理:

当一个请求到达 FastAPI 应用时,它们的串联流程如下:

步骤 参与者 动作内容
1. 监听UvicornUvicorn 启动并运行在 asyncio 的事件循环中,等待网络连接。
2. 接收Uvicorn浏览器发来请求。Uvicorn 解析 HTTP 协议,将其转化为一个 ASGI Scope(包含请求信息的字典)。
3. 调度asyncioUvicorn 告诉 asyncio:有一个新任务要处理,请加入执行队列。
4. 路由Starletteasyncio 执行该任务并调用 Starlette。Starlette 查看路由表,找到对应的 async def 处理函数。
5. 执行FastAPIFastAPI 在 Starlette 运行期间介入,利用 Pydantic 检查参数,最后运行业务代码。
6. 挂起asyncio如果代码中有 await(如查数据库),asyncio 立刻挂起当前任务,转去处理下一个请求。
7. 返回全链路I/O 返回后,asyncio 恢复任务,Starlette 生成响应,Uvicorn 发回浏览器。

1. 谁是真正的“指挥官”?

  • 在异步编程中,asyncio.run() 才是调度入口。
  • Uvicorn 启动后会接入事件循环,把监听、读取数据包、解析协议等动作转成循环中的任务。

2. 协作模型

  • 启动阶段:uvicorn main:app 本质是启动一个 Python 进程。
  • 接管阶段:Uvicorn 初始化并运行 asyncio 事件循环。
  • 循环阶段:监听任务常驻;有请求到来时,被事件循环唤醒处理。
  • 分发阶段:Uvicorn 解析后把请求交给 FastAPI/Starlette,并继续由同一事件循环调度。

3. 单进程下,你的代码与 Uvicorn 通常就在同一线程、同一事件循环里协作运行。

2. 路径参数

Q1:FastAPI 路由匹配优先级

在 FastAPI 中,谁先声明,谁先被尝试匹配;一旦命中就停止继续向下匹配。

Q2:为什么“通用路径”要放在“具体路径”后面?

  • 静态路径(如 /users/me/users/admin)是明确字符串,匹配精确。
  • 动态路径(如 /users/{user_id})是“贪婪匹配”,会尝试接收几乎所有同结构片段。
  • 如果把动态路径放前面,就可能像“黑洞”一样抢先吞掉后面的静态路径请求。

Q3:顺序写反时会发生什么?

@app.get("/users/{user_id}")
async def read_user(user_id: str):
    return {"user_id": user_id}

@app.get("/users/me")
async def read_user_me():
    return {"user_id": "the current user"}

访问 /users/me 时,"me" 会先被当成 user_id 命中第一个路由,导致第二个静态路由无法触发。

Q4:正确写法是什么?

# 1) 先声明更具体的静态路径
@app.get("/users/me")
async def read_user_me():
    return {"user_id": "the current user"}

# 2) 再声明更通用的动态路径
@app.get("/users/{user_id}")
async def read_user(user_id: str):
    return {"user_id": user_id}

Q5:类型提示会影响匹配吗?

会。路径参数类型可作为“第二层过滤器”。例如把 user_id 限制为 int

@app.get("/users/{user_id}")
async def read_user(user_id: int):
    return {"user_id": user_id}

@app.get("/users/me")
async def read_user_me():
    return {"user_id": "the current user"}

当访问 /users/me 时,若前面的动态路由无法把 "me" 解析为 int,会继续寻找后续可匹配路由,从而命中静态路径。

Q6:如何用 Enum 限制路径参数的可选值?

如果路径参数只允许“预定义值集合”,推荐使用标准库 Enum。在 FastAPI 中,通常让枚举类继承 strEnum,这样文档会按字符串枚举正确渲染,校验也更直观。

  • 为什么继承 str:让 OpenAPI 文档将其识别为字符串枚举,而不是通用对象。
  • 收益:参数值自动校验、接口文档下拉可选、非法值自动返回 422。
from enum import Enum

from fastapi import FastAPI


class ModelName(str, Enum):
    alexnet = "alexnet"
    resnet = "resnet"
    lenet = "lenet"


app = FastAPI()


@app.get("/models/{model_name}")
async def get_model(model_name: ModelName):
    if model_name is ModelName.alexnet:
        return {"model_name": model_name, "message": "Deep Learning FTW!"}

    if model_name.value == "lenet":
        return {"model_name": model_name, "message": "LeCNN all the images"}

    return {"model_name": model_name, "message": "Have some residuals"}

Q7:如何在路径参数中包含“路径本身”?

假设你的接口是 /files/{file_path},但你希望 file_path 里还能包含斜杠(/),例如 home/johndoe/myfile.txt。这时需要使用 Starlette 的路径转换器:

/files/{file_path:path}

其中参数名仍是 file_path,结尾的 :path 表示该参数可以匹配完整路径片段。

OpenAPI 支持说明

OpenAPI 规范本身不支持直接声明“包含路径的路径参数”,因此文档不会额外标注该参数是路径类型;但 FastAPI 仍可正常工作。

示例代码

from fastapi import FastAPI

app = FastAPI()


@app.get("/files/{file_path:path}")
async def read_file(file_path: str):
    return {"file_path": file_path}

前导斜杠提示

如果你希望参数值本身以 / 开头,例如 /home/johndoe/myfile.txt,请求 URL 需要写成:

/files//home/johndoe/myfile.txt

注意 fileshome 之间是双斜杠 //

3. 请求体

Q1:在一次完整请求-响应中,请求体和响应体在 JSON、Pydantic、dict 之间是如何转换的?

FastAPI:请求体 / 响应体格式转换全流程 ① 客户端发送请求 Content-Type: application/json Body = JSON 字符串 ② FastAPI 解析请求体(反序列化) JSON → Python 基础类型(通常是一个字典) ③ 参数类型决定注入形态 item: Item(BaseModel) → Pydantic 对象 item: dict / list[...] → 原生 Python 对象 ④ 进入路径函数 若参数是 Pydantic:可用 item.name / item.price 等属性 需要 dict 时:Pydantic v2 用 item.model_dump() 你可以返回:dict / list / Pydantic 模型 / dataclass / ORM 对象 ⑤ FastAPI 响应序列化阶段 默认:自动将返回值转为 JSON 可编码数据 ⑥ 返回给客户端 HTTP Response Body = JSON(文本) 重点:网络上传输始终是 JSON;Pydantic / dict 是服务端 Python 世界中的中间形态 若你直接返回 JSONResponse,需自己保证内容 JSON 可序列化;复杂对象先用 jsonable_encoder 转换

官方实践要点

  • 请求体建模优先用 Pydantic,获得验证、类型转换和文档能力。
  • 需要把模型转字典时,Pydantic v2 使用 model_dump()
  • 当你手动构造 JSONResponse 且内容含复杂对象时,先用 jsonable_encoder()
  • response_model 控制“响应体最终格式”,避免把不该暴露字段返回给前端。
4. 查询参数与字符串验证

Q1:查询参数基础写法是什么?可选参数如何声明?

Python 3.13 推荐直接使用联合类型语法:str | None。当默认值是 None 时,参数就是可选。

from fastapi import FastAPI

app = FastAPI()


@app.get("/items/")
async def read_items(q: str | None = None):
    result = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
    if q:
        result["q"] = q
    return result

Q2:为什么推荐 Annotated + Query()

官方推荐把校验与元数据放进 Annotated[..., Query(...)]:语义更清晰、类型提示更稳定、函数默认值仍是“真实 Python 默认值”。

from typing import Annotated
from fastapi import FastAPI, Query

app = FastAPI()


@app.get("/items/")
async def read_items(
    q: Annotated[str | None, Query(max_length=50)] = None,
):
    return {"q": q}
  • 旧写法(仍可用,但不再是首选):q: str | None = Query(default=None, max_length=50)
  • 使用 Annotated 时,默认值应写在函数参数位置,而不是 Query(default=...) ,避免“双默认值冲突”

Q4:字符串校验可以做哪些?

可直接在 Query() 中声明:

  • min_length
  • max_length
  • pattern(正则)

Q6:如何声明“一个查询参数可传多个值”?

使用列表类型并显式 Query(),例如 ?q=foo&q=bar

from typing import Annotated
from fastapi import FastAPI, Query

app = FastAPI()


@app.get("/items/")
async def read_items(
    q: Annotated[list[str] | None, Query()] = None,
):
    return {"q": q}

带默认列表:

from typing import Annotated
from fastapi import FastAPI, Query

app = FastAPI()


@app.get("/items/default-list")
async def read_items_default(
    q: Annotated[list[str], Query()] = ["foo", "bar"],
):
    return {"q": q}

Q7:除了验证,还能给查询参数加哪些文档元数据?

常见元数据:

  • title(字段的简短名称)
  • description(详细说明文字)
  • alias(字段别名)
  • deprecated(标记弃用)
  • include_in_schema(是否出现在 OpenAPI)
5. 路径参数和数字验证

Q1:路径参数也能像查询参数那样做元数据与验证吗?

可以。FastAPI 使用 Path() 给路径参数声明元数据与约束,能力与 Query() 同源、风格一致。官方推荐写法是 Annotated[T, Path(...)]

from typing import Annotated
from fastapi import FastAPI, Path, Query

app = FastAPI()


@app.get("/items/{item_id}")
async def read_items(
    item_id: Annotated[int, Path(title="要获取的 item ID")],
    q: Annotated[str | None, Query(alias="item-query")] = None,
):
    result = {"item_id": item_id}
    if q:
        result["q"] = q
    return result

Q2:路径参数能否设置默认值或设为可选?

不能。路径参数始终是必填,因为它是 URL 路径的一部分。即使你写了默认值或 None,在路由匹配层面它仍然必须存在。

Q3:Annotated 为什么是当前最佳实践?

  • 把“类型”与“参数元数据/验证”放在一起,语义清晰。
  • 避免旧写法中“函数参数默认值顺序”造成的 Python 语法限制。
  • 函数默认值保持纯 Python 语义,可读性和可复用性更好。

Q8:数字校验关键字?

  • gt:greater than(>)
  • ge:greater than or equal(>=)
  • lt:less than(<)
  • le:less than or equal(<=)
6. 查询参数模型

Q1:什么是“查询参数模型”?

当多个查询参数属于同一组筛选条件时,可以用一个 Pydantic 模型统一声明。这样可复用、可集中校验、可自动生成更清晰的文档。

  • 适用场景:分页、排序、过滤标签等参数经常一起出现。
  • 收益:避免函数签名参数过长;规则集中在一个模型里维护。

Q2:官方推荐写法是什么?

推荐:Annotated[Model, Query()]。这是当前官方最佳实践风格。

  • Field() 来自 Pydantic,用于给模型字段加“约束 + 元数据”。(注:在底层,FastAPI 的 Query/Path/Body 与 Pydantic 的 Field 都可追溯到同一“家族”(FieldInfo);因此它们共享大量通用参数,如 titledescriptiondefaultalias 等。)
  • Field 的第一个位置参数通常是默认值
  • Literal 来自 typing,用于把值限制在固定字面量集合内。
  • 例如 Literal["created_at", "updated_at"] 表示只能二选一。
from typing import Annotated, Literal

from fastapi import FastAPI, Query
from pydantic import BaseModel, Field

app = FastAPI()


class FilterParams(BaseModel):
    limit: int = Field(100, gt=0, le=100)
    offset: int = Field(0, ge=0)
    order_by: Literal["created_at", "updated_at"] = "created_at"
    tags: list[str] = []


@app.get("/items/")
async def read_items(filter_query: Annotated[FilterParams, Query()]):
    return filter_query

Q5:如何禁止“未声明的额外查询参数”?

使用 Pydantic v2 的模型配置:model_config = {"extra": "forbid"}

from typing import Annotated, Literal

from fastapi import FastAPI, Query
from pydantic import BaseModel, Field

app = FastAPI()


class StrictFilterParams(BaseModel):
    model_config = {"extra": "forbid"}

    limit: int = Field(100, gt=0, le=100)
    offset: int = Field(0, ge=0)
    order_by: Literal["created_at", "updated_at"] = "created_at"
    tags: list[str] = []


@app.get("/strict-items/")
async def read_strict_items(filters: Annotated[StrictFilterParams, Query()]):
    return filters

此时若请求包含未声明参数(例如 tool=plumbus),会返回校验错误(extra_forbidden)。

7. 请求体 - 多个参数

Q1:当路径函数里既有路径参数、查询参数,又有请求体参数时,FastAPI 如何区分?

  • 在路径模板里出现过的参数名(如 /items/{item_id})→ 路径参数。
  • “简单类型”参数(如 intstrbool)且不在路径中 → 查询参数。
  • Pydantic 模型参数(BaseModel)→ 请求体。
from typing import Annotated

from fastapi import Body, FastAPI
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):
    name: str
    description: str | None = None
    price: float
    tax: float | None = None


@app.put("/items/{item_id}")
async def update_item(
    item_id: int,              # path
    item: Item,                # body
    q: str | None = None,      # query
):
    return {"item_id": item_id, "item": item, "q": q}

Q2:可以在一个接口中接收多个请求体模型吗?

可以。多个模型参数会被组合到同一个 JSON Body 的不同 key 下,key 默认就是参数名。

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):
    name: str
    price: float


class User(BaseModel):
    username: str
    full_name: str | None = None


@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item, user: User):
    return {"item_id": item_id, "item": item, "user": user}

对应请求体形状示意:

{
  "item": {"name": "Foo", "price": 42.0},
  "user": {"username": "dave", "full_name": "Dave Grohl"}
}

Q3:同一个接口里能否再加“单独的 body 字段”(非模型)?

可以。使用 Body() 显式声明它来自请求体,而不是查询参数。

from typing import Annotated

from fastapi import Body, FastAPI
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):
    name: str
    price: float


class User(BaseModel):
    username: str


@app.put("/items/{item_id}")
async def update_item(
    item_id: int,
    item: Item,
    user: User,
    importance: Annotated[int, Body(gt=0)],
):
    return {
        "item_id": item_id,
        "item": item,
        "user": user,
        "importance": importance,
    }

此时请求体会包含 importance 顶层字段。

Q4:为什么简单类型参数有时会“跑到查询参数”里?

这是 FastAPI 的内置参数解析优先级决定的:

  • 若参数类型是 BaseModel(Pydantic 模型),默认按 Body 解析。
  • 若参数是简单类型(int / str / float / bool),且不在路径模板中,默认按 Query 解析。
  • 所以简单类型若想进入请求体,必须显式使用 Body() 改变默认行为。
from typing import Annotated

from fastapi import Body, FastAPI

app = FastAPI()


# ❌ 不显式 Body:age 会被当成查询参数(/update-age?age=25)
@app.post("/update-age-wrong")
async def update_age_wrong(age: int):
    return {"age": age}


# ✅ Python 3.13 + FastAPI 推荐:使用 Annotated 显式声明 Body
# 期待 JSON Body:{"age": 25}
@app.post("/update-age")
async def update_age(age: Annotated[int, Body()]):
    return {"age": age}

Q5:请求体只有一个模型参数时,如何把它“嵌入键名”下?

默认不加 embed=True 时,FastAPI 会把“整个请求体”直接当作这个模型(结构最扁平、最简洁)。

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):
    name: str
    description: str | None = None
    price: float
    tax: float | None = None


@app.post("/items/")
async def create_item(item: Item):
    return item

此时期待的 JSON 形状:

{
  "name": "Foo",
  "description": "The pretender",
  "price": 42.0,
  "tax": 3.2
}

使用 embed=True 时,FastAPI 会去请求体里找与参数同名的键(这里是 item),再把该键下的数据映射到模型。

from typing import Annotated

from fastapi import Body, FastAPI
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):
    name: str
    description: str | None = None
    price: float
    tax: float | None = None


@app.post("/items/embedded")
async def create_item_embedded(item: Annotated[Item, Body(embed=True)]):
    return item

此时期待的 JSON 形状:

{
  "item": {
    "name": "Foo",
    "description": "The pretender",
    "price": 42.0,
    "tax": 3.2
  }
}

什么时候用 embed=True 当你希望结构更具扩展性(例如未来同一接口还会同时接收 itemusermetadata),提前使用嵌入键名会让 API 结构更一致、更易演进。

Q6:Body() 能加哪些信息?和 Query()/Path() 一致吗?

基本一致:可添加校验与文档元数据(如边界约束、标题、描述、示例等)。FastAPI 在 OpenAPI 中会自动反映这些信息。

8. 请求体 - 嵌套模型

Q1:什么是“嵌套模型”?

嵌套模型就是:一个 Pydantic 模型里包含另一个模型(或模型列表)。在真实业务里非常常见,例如商品里有图片、标签、作者信息等结构化子对象。

Q2:基础嵌套怎么写?

from typing import Annotated

from fastapi import Body, FastAPI
from pydantic import BaseModel, HttpUrl

app = FastAPI()


class Image(BaseModel):
    url: HttpUrl
    name: str


class Item(BaseModel):
    name: str
    description: str | None = None
    price: float
    tax: float | None = None
    image: Image | None = None


@app.put("/items/{item_id}")
async def update_item(
    item_id: int,
    item: Annotated[Item, Body()],
):
    return {"item_id": item_id, "item": item}

这里 item.image 就是一个嵌套对象;HttpUrl 会自动验证 URL 合法性。

Q3:列表里的每一项也能是模型吗?

可以,这是最常见模式之一:list[Image]

from typing import Annotated

from fastapi import Body, FastAPI
from pydantic import BaseModel, HttpUrl

app = FastAPI()


class Image(BaseModel):
    url: HttpUrl
    name: str


class Item(BaseModel):
    name: str
    images: list[Image] = []


@app.post("/items/")
async def create_item(item: Annotated[Item, Body()]):
    return item

请求体里的 images 应是对象数组,FastAPI 会逐项校验。

Q4:深层嵌套怎么写?

可直接使用 Python 3.13 类型注解,如 set[str]dict[str, float],并与模型组合。

from typing import Annotated

from fastapi import Body, FastAPI
from pydantic import BaseModel, HttpUrl

app = FastAPI()


class Image(BaseModel):
    url: HttpUrl
    name: str


class Item(BaseModel):
    name: str
    tags: set[str] = set()
    images: list[Image] = []
    metadata: dict[str, float] = {}


@app.post("/items/complex")
async def create_complex_item(item: Annotated[Item, Body()]):
    return item
  • set[str] 会去重(输入重复值时输出为集合语义)。
  • dict[str, float] 会校验 value 必须可解析为浮点数。

Q5:如果请求体本身就是“模型列表”而不是对象怎么办?

路径函数参数直接声明为 list[Item] 即可。

from typing import Annotated

from fastapi import Body, FastAPI
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):
    name: str
    price: float


@app.post("/items/batch")
async def create_items(items: Annotated[list[Item], Body()]):
    return {"count": len(items), "items": items}

这时请求体根节点是 JSON 数组,而不是 JSON 对象。

9. 声明请求示例数据

Q1:什么是 schema-extra-example?

它本质是在 JSON Schema / OpenAPI 里给请求体声明示例。Pydantic v2 中推荐使用 model_config = {"json_schema_extra": {...}}

Q2:如何给模型声明“单个/多个请求示例”?

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):
    name: str
    description: str | None = None
    price: float
    tax: float | None = None

    model_config = {
        "json_schema_extra": {
            "examples": [
                {
                    "name": "Foo",
                    "description": "A very nice Item",
                    "price": 35.4,
                    "tax": 3.2,
                },
                {
                    "name": "Bar",
                    "price": 12.5,
                },
            ]
        }
    }


@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item):
    return {"item_id": item_id, "item": item}

Q3:如何同一接口多示例?

在参数层使用 Body(openapi_examples=...)

from typing import Annotated

from fastapi import Body, FastAPI
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):
    name: str
    description: str | None = None
    price: float
    tax: float | None = None


@app.put("/items/{item_id}/examples")
async def update_item_with_examples(
    item_id: int,
    item: Annotated[
        Item,
        Body(
            openapi_examples={
                "normal": {
                    "summary": "标准场景",
                    "description": "包含可选字段 tax",
                    "value": {
                        "name": "Foo",
                        "description": "A very nice Item",
                        "price": 35.4,
                        "tax": 3.2,
                    },
                },
                "minimal": {
                    "summary": "最小请求",
                    "description": "只传必填字段",
                    "value": {
                        "name": "Bar",
                        "price": 12.5,
                    },
                },
                "discounted": {
                    "summary": "促销商品",
                    "description": "用于折扣业务测试",
                    "value": {
                        "name": "Baz",
                        "description": "Discounted item",
                        "price": 9.9,
                        "tax": 0.8,
                    },
                },
            }
        ),
    ],
):
    return {"item_id": item_id, "item": item}

Q4:example / examples / openapi_examples 如何选?

  • example:单示例,最简。
  • examples:Schema 层多示例,常放在模型 json_schema_extra
  • openapi_examples:Request Body 层多示例,文档展示最友好,适合“示例数 > 1”的面试场景。
10. Cookie 与 Header 参数

Q1:Cookie/Header 参数和 Query/Path 的声明模式一致吗?

一致。四者都属于 FastAPI 的参数声明体系,推荐统一使用 Annotated[T, Param()] 风格

  • Cookie 用 Cookie()
  • Header 用 Header()
  • 如果不用对应的参数类,FastAPI 会把它当作查询参数处理。
from typing import Annotated

from fastapi import Cookie, FastAPI, Header

app = FastAPI()


@app.get("/client-context")
async def read_client_context(
    session_id: Annotated[str | None, Cookie()] = None,
    user_agent: Annotated[str | None, Header()] = None,
):
    return {
        "session_id": session_id,
        "user_agent": user_agent,
    }

Q2:Header 为什么可以写 user_agent 却匹配 User-Agent

因为 Header() 默认会把参数名中的下划线 _ 自动转换为连字符 -,并且 Header 名大小写不敏感。

如需关闭自动转换,可设置 convert_underscores=False

Q3:如何处理“重复 Header”(同名多值)?

把 Header 参数声明成列表类型即可,FastAPI 会收集同名 Header 的所有值。

from typing import Annotated

from fastapi import FastAPI, Header

app = FastAPI()


@app.get("/headers/multi")
async def read_multi_header(
    x_token: Annotated[list[str] | None, Header()] = None,
):
    return {"x_token_values": x_token}

例如请求里发两次 X-Token,响应里会得到一个字符串列表。

Q4:什么时候使用 Cookie/Header“参数模型”?

当你有一组相关参数要复用(例如统一客户端上下文、灰度标记、链路追踪头)时,推荐建 Pydantic 模型统一声明。

Q5:如何禁止“未声明的额外 Cookie/Header”?

在模型里设置 model_config = {"extra": "forbid"},即可拒绝未声明字段,返回 extra_forbidden

11. 响应模型

Q1:为什么要声明响应模型?

响应模型会在运行时参与 校验、过滤、序列化,并生成 OpenAPI Schema。核心价值是:保证返回结构稳定、避免敏感字段泄漏、让前后端契约可验证。

Q2:用“返回类型注解”还是 response_model

场景一:直接使用返回类型注解(类型一致)

当函数直接返回 Pydantic 模型实例时,优先使用返回类型注解,这是现代且清晰的写法。

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


class User(BaseModel):
    name: str
    age: int


@app.get("/users/{user_id}")
async def get_user(user_id: int) -> User:
    return User(name="Alice", age=25)

这种方式的优势是类型信息直接体现在函数签名里,编辑器和类型检查工具能更准确地提示返回值是否正确。

场景二:使用 response_model(类型不一致 / 需要转换过滤)

当函数返回的是 ORM 对象、原始字典,或包含更多内部字段的数据时,使用 response_model 让 FastAPI 按指定模型进行转换与过滤。

from typing import Any

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


class UserOut(BaseModel):
    name: str


@app.get("/users/{user_id}", response_model=UserOut)
async def get_user_public(user_id: int) -> Any:
    db_user = {
        "name": "Alice",
        "password": "secret",
        "email": "alice@example.com",
    }
    return db_user

这种方式的优势是即使函数返回了更多字段,最终响应也只会暴露 UserOut 中声明的字段。

核心区别

对比项 返回类型注解(-> Item response_model=Item
编辑器类型提示 更强,签名即契约 一般,主要依赖装饰器配置
数据过滤能力 有,但更适合“返回值已是目标模型”场景 强,适合对字典/ORM 做输出裁剪
典型适用对象 Pydantic 模型实例 字典、ORM 对象、内部对象到公开模型的转换

Q3:response_model 和返回类型同时声明时,谁优先?

response_model 优先。它会覆盖返回类型用于响应过滤与文档。

Q4:如何避免把密码等敏感字段返回给前端?

把“输入模型”和“输出模型”分离,并在路由上声明 response_model=输出模型。这是最推荐、最安全、最清晰的做法。

from typing import Any

from fastapi import FastAPI
from pydantic import BaseModel, EmailStr

app = FastAPI()


class UserIn(BaseModel):
    username: str
    password: str
    email: EmailStr


class UserOut(BaseModel):
    username: str
    email: EmailStr


@app.post("/users/", response_model=UserOut)
async def create_user(user: UserIn) -> Any:
    return user

Q6:什么时候要设 response_model=None

场景一:动态返回不同结构的数据

例如同一个接口可能返回重定向,也可能返回普通 JSON。若返回注解写成 Response | dict[str, str],建议同时设置 response_model=None

from fastapi import FastAPI, Response
from fastapi.responses import RedirectResponse

app = FastAPI()


@app.get("/portal", response_model=None)
async def get_portal(teleport: bool = False) -> Response | dict[str, str]:
    if teleport:
        return RedirectResponse(url="https://example.com")
    return {"message": "Here is your portal."}

场景二:直接返回原生响应对象

当你明确手动构造响应(例如自定义状态码、Header、媒体类型、文件流)时,通常不需要让 response_model 参与过滤。

from fastapi import FastAPI
from fastapi.responses import FileResponse, JSONResponse

app = FastAPI()


@app.get("/download", response_model=None)
async def download_file(export: bool = False):
    if export:
        return FileResponse(path="report.csv", media_type="text/csv", filename="report.csv")
    return JSONResponse(content={"message": "set export=true to download"})

Q7:如何只返回“被显式设置”的字段?

使用 response_model_exclude_unset=True。只给前端返回用户在代码里明确赋值过的字段,而跳过那些仅仅是因为有默认值才存在的字段。

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):
    name: str
    description: str | None = None
    price: float
    tax: float = 10.5
    tags: list[str] = []


items = {
    "foo": {"name": "Foo", "price": 50.2},
    "bar": {"name": "Bar", "description": "The bartenders", "price": 62, "tax": 20.2},
}


@app.get("/items/{item_id}", response_model=Item, response_model_exclude_unset=True)
async def read_item(item_id: str):
    return items[item_id]

返回结果示例:

# 请求
GET /items/foo

# 响应
{
  "name": "Foo",
  "price": 50.2
}

# 请求
GET /items/bar

# 响应
{
  "name": "Bar",
  "description": "The bartenders",
  "price": 62,
  "tax": 20.2
}

可以看到:foo 没有显式设置 descriptiontaxtags,因此不会返回这些默认字段。

Q8:response_model_include / response_model_exclude

response_model_include (包含):指定只返回哪些字段。不在列表里的字段都会被过滤掉。

response_model_exclude (排除):指定不要返回哪些字段。列表里的字段会被过滤,其余的正常返回。

Q9:响应状态码

统一在路径操作装饰器上使用 status_code(它是装饰器参数,不是函数参数)。

  • 默认状态码是 200
  • 可传数字(如 201),也可传枚举常量(推荐 fastapi.status,可读性更好)。
  • 显式设置后,实际响应与 OpenAPI 文档都会使用该状态码。
  • 204304 等状态码不应包含响应体。
from fastapi import FastAPI, status

app = FastAPI()


# 默认 200
@app.get("/items/{item_id}")
async def read_item(item_id: str):
    return {"item_id": item_id}


# 显式声明 201(推荐使用 fastapi.status 常量)
@app.post("/items/", status_code=status.HTTP_201_CREATED)
async def create_item(name: str):
    return {"name": name}


# 204 不返回响应体
@app.delete("/items/{item_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_item(item_id: str):
    return None
12. 额外模型

Q1:为什么要用“额外模型”?

同一个业务实体在不同阶段通常有不同数据形态:请求入参、响应出参、数据库落库。把它们拆成多个模型可以保证职责清晰,避免把不该暴露的数据返回给客户端。

  • 输入模型:可包含明文密码等仅请求期字段。
  • 输出模型:只保留可对外字段。
  • DB 模型:保存哈希密码等内部字段。

Q3:model_dump()** 解包在这里起什么作用?

model_dump() 会把 Pydantic 模型转成 dict,再用 ** 解包给另一个模型构造函数,实现“在保留公共字段的同时补充新字段”。

user_dict = user_in.model_dump()
user_in_db = UserInDB(**user_dict, hashed_password=hashed_password)

Q4:如何减少模型字段重复?

用“基类 + 继承”提取公共字段,这是官方推荐的低重复实践。

Q5:响应如果可能是“多种模型之一”怎么声明?

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


class BaseItem(BaseModel):
    description: str
    type: str


class CarItem(BaseItem):
    type: str = "car"


class PlaneItem(BaseItem):
    type: str = "plane"
    size: int


items = {
    "item1": {"description": "Low rider", "type": "car"},
    "item2": {"description": "Aeroplane", "type": "plane", "size": 5},
}


@app.get("/items/{item_id}", response_model=PlaneItem | CarItem)
async def read_item(item_id: str):
    return items[item_id]

Q6:如何声明“模型列表”响应?

直接用 response_model=list[Model]

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):
    name: str
    description: str


@app.get("/items/list", response_model=list[Item])
async def read_items_list():
    return [
        {"name": "Foo", "description": "There comes my hero"},
        {"name": "Red", "description": "It's my aeroplane"},
    ]
13. 处理错误

Q1:FastAPI 中业务错误该怎么返回?

推荐直接 raise HTTPException(...),而不是 return 错误对象。抛出后会立即中断当前请求流程并返回对应 HTTP 错误响应。

from fastapi import FastAPI, HTTPException

app = FastAPI()


items = {"foo": "The Foo Wrestlers"}


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

Q2:HTTPException.detail 只能是字符串吗?

不是。可以是任何可 JSON 化的数据(如 dictlist)。FastAPI 会自动序列化。

from fastapi import FastAPI, HTTPException

app = FastAPI()


@app.get("/orders/{order_id}")
async def read_order(order_id: str):
    raise HTTPException(
        status_code=409,
        detail={
            "code": "ORDER_CONFLICT",
            "message": "Order state conflict",
            "order_id": order_id,
        },
    )

Q3:错误响应里可以加自定义 Header 吗?

可以。通过 HTTPException(headers={...}) 增加响应头,常用于安全或客户端提示场景。

from fastapi import FastAPI, HTTPException

app = FastAPI()


@app.get("/items-header/{item_id}")
async def read_item_header(item_id: str):
    if item_id != "foo":
        raise HTTPException(
            status_code=404,
            detail="Item not found",
            headers={"X-Error": "ITEM_NOT_FOUND"},
        )
    return {"item": "The Foo Wrestlers"}

Q4:如何全局处理自定义异常?

使用 @app.exception_handler(YourException) 注册全局处理器。适合统一业务错误结构。

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

app = FastAPI()


class UnicornException(Exception):
    def __init__(self, name: str):
        self.name = name


@app.exception_handler(UnicornException)
async def unicorn_exception_handler(request: Request, exc: UnicornException):
    return JSONResponse(
        status_code=418,
        content={"message": f"Oops! {exc.name} did something."},
    )


@app.get("/unicorns/{name}")
async def read_unicorn(name: str):
    if name == "yolo":
        raise UnicornException(name=name)
    return {"unicorn_name": name}

Q5:如何覆盖默认的请求验证错误处理?

可覆盖 RequestValidationError,把默认 422 JSON 改为你想要的格式(如纯文本或统一 JSON 结构)。

from fastapi import FastAPI, Request
from fastapi.exceptions import RequestValidationError
from fastapi.responses import PlainTextResponse

app = FastAPI()


@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    lines = ["Validation errors:"]
    for error in exc.errors():
        lines.append(f"field={error['loc']} msg={error['msg']}")
    return PlainTextResponse("\n".join(lines), status_code=400)


@app.get("/numbers/{value}")
async def read_number(value: int):
    return {"value": value}

Q6:如何覆盖 HTTPException 的默认处理?

建议对 StarletteHTTPException 注册处理器,这样能覆盖 FastAPI/Starlette 内部抛出的同类异常。

from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import PlainTextResponse
from starlette.exceptions import HTTPException as StarletteHTTPException

app = FastAPI()


@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request: Request, exc: StarletteHTTPException):
    return PlainTextResponse(str(exc.detail), status_code=exc.status_code)


@app.get("/items/{item_id}")
async def read_item(item_id: int):
    if item_id == 3:
        raise HTTPException(status_code=418, detail="Nope! I don't like 3.")
    return {"item_id": item_id}

Q7:RequestValidationError.body 有什么用?

它保存了导致校验失败的原始请求体。开发阶段可用于排错与日志记录。

from fastapi import FastAPI, Request
from fastapi.encoders import jsonable_encoder
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):
    title: str
    size: int


@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    return JSONResponse(
        status_code=422,
        content=jsonable_encoder({"detail": exc.errors(), "body": exc.body}),
    )


@app.post("/items")
async def create_item(item: Item):
    return item

Q8:可以在自定义处理器里复用 FastAPI 默认异常处理吗?

可以。先做日志、埋点,再调用官方默认处理器返回标准结构,是非常实用的生产实践。

from fastapi import FastAPI, HTTPException, Request
from fastapi.exception_handlers import (
    http_exception_handler,
    request_validation_exception_handler,
)
from fastapi.exceptions import RequestValidationError
from starlette.exceptions import HTTPException as StarletteHTTPException

app = FastAPI()


@app.exception_handler(StarletteHTTPException)
async def custom_http_exception_handler(request: Request, exc: StarletteHTTPException):
    print(f"HTTP error: {exc!r}")
    return await http_exception_handler(request, exc)


@app.exception_handler(RequestValidationError)
async def custom_validation_exception_handler(request: Request, exc: RequestValidationError):
    print(f"Validation error: {exc}")
    return await request_validation_exception_handler(request, exc)


@app.get("/items/{item_id}")
async def read_item(item_id: int):
    if item_id == 3:
        raise HTTPException(status_code=418, detail="Nope! I don't like 3.")
    return {"item_id": item_id}
14. 路径操作配置

Q1:路径操作配置参数应该写在哪里?

写在装饰器上(如 @app.get(...)@app.post(...)),不是写在函数参数里。它们会影响 OpenAPI 文档和接口行为。

from fastapi import FastAPI, status
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):
    name: str
    description: str | None = None
    price: float


@app.post(
    "/items/",
    response_model=Item,
    status_code=status.HTTP_201_CREATED,
    tags=["items"],
    summary="Create an item",
    response_description="The created item",
)
async def create_item(item: Item):
    return item

Q2:status_code 在路径操作配置里怎么用?

可用数字,也可用 fastapi.status 常量(推荐后者,可读性更好)。该状态码会同时用于真实响应和 OpenAPI 文档。

分类 范围 常用常量示例 含义
1xx 信息响应 HTTP_101_SWITCHING_PROTOCOLS 切换协议(如 WebSocket)
2xx 成功 HTTP_200_OK
HTTP_201_CREATED
请求成功;资源创建成功
3xx 重定向 HTTP_301_MOVED_PERMANENTLY
HTTP_304_NOT_MODIFIED
永久移动;资源未修改(缓存)
4xx 客户端错误 HTTP_400_BAD_REQUEST
HTTP_401_UNAUTHORIZED
HTTP_403_FORBIDDEN
HTTP_404_NOT_FOUND
请求无效;未授权;访问禁止;找不到资源
5xx 服务器错误 HTTP_500_INTERNAL_SERVER_ERROR 服务器内部错误

Q3:如何配置标签 tags

使用 tags=[...] 给接口分组,文档里会按标签聚合显示。大型项目建议统一标签命名。

from fastapi import FastAPI

app = FastAPI()


@app.get("/items/", tags=["items"])
async def read_items():
    return [{"name": "Foo", "price": 42}]


@app.get("/users/", tags=["users"])
async def read_users():
    return [{"username": "johndoe"}]

Q4:标签可以用枚举统一管理吗?

可以。推荐用 str, Enum 管理标签常量,避免拼写不一致。

from enum import Enum

from fastapi import FastAPI

app = FastAPI()


class Tags(str, Enum):
    items = "items"
    users = "users"


@app.get("/items/", tags=[Tags.items])
async def get_items():
    return ["Portal Gun", "Plumbus"]


@app.get("/users/", tags=[Tags.users])
async def get_users():
    return ["Rick", "Morty"]

Q5:summarydescription 分别做什么?

summary 是短标题,description 是详细说明。两者都会展示在文档中。

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):
    name: str
    price: float


@app.post(
    "/items/meta",
    response_model=Item,
    summary="Create an item",
    description="Create an item with name and price.",
)
async def create_item_meta(item: Item):
    return item

Q6:描述能写在函数 docstring 里吗?

可以。长描述推荐写在函数 docstring,支持 Markdown,文档展示更清晰。常见做法是装饰器保留 summary,详细说明写到 docstring。

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):
    name: str
    price: float
    tax: float | None = None


@app.post("/items/doc", response_model=Item, summary="Create an item")
async def create_item_doc(item: Item):
    """
    Create an item with complete information.

    - **name**: required
    - **price**: required
    - **tax**: optional
    """
    return item

Q7:response_descriptiondescription 有什么区别?

description 描述整个路径操作;response_description 只描述响应内容。OpenAPI 规范要求响应有描述,不填时 FastAPI 会给默认值(如 Successful response)。

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):
    name: str


@app.post(
    "/items/response-desc",
    response_model=Item,
    description="Create an item resource.",
    response_description="The created item",
)
async def create_item_response_desc(item: Item):
    return item

Q8:如何把旧接口标记为弃用但暂不删除?

在装饰器设置 deprecated=True。这样文档会明确标记为已弃用,便于平滑迁移。

from fastapi import FastAPI

app = FastAPI()


@app.get("/v1/items", tags=["items"], deprecated=True)
async def read_items_v1():
    return [{"name": "legacy-item"}]


@app.get("/v2/items", tags=["items"])
async def read_items_v2():
    return [{"name": "new-item"}]
15. JSON 兼容编码器

Q1:什么是 jsonable_encoder

当你尝试把一些复杂的 Python 对象直接变成 JSON 时,会遇到麻烦,例如:

  • Pydantic 模型:JSON 不认识 Item(name="Apple") 这种对象。
  • 日期时间(datetime):JSON 不识别 datetime.now()
  • 数据库对象:例如 SQLAlchemy 模型实例。

如果你直接把这些对象交给 json 模块,会报错:Object of type datetime is not JSON serializable

jsonable_encoder() 会接收这些复杂对象,并递归地将每一部分转换成 JSON 能理解的类型,例如:

  • datetime ➜ ISO 格式字符串(如 "2026-03-25T22:39:10")。
  • Pydantic 模型 ➜ 普通 dict
  • Enum ➜ 其原始值(字符串或数字)。

当你需要把 Pydantic 模型或包含特殊类型的数据转换为“可被 JSON 序列化”的 Python 数据结构时,使用 jsonable_encoder()

  • 典型场景:写入 NoSQL、缓存、消息队列、日志系统。
  • 它返回的是 dict/list 等 Python 对象,而不是 JSON 字符串。

Q2:它和 model_dump() 有什么区别?

model_dump() 负责把模型导出为字典;jsonable_encoder() 进一步确保所有值都 JSON 兼容。

Q3:标准写法示例(写入只接受 JSON 兼容数据的存储)

from datetime import datetime

from fastapi import FastAPI
from fastapi.encoders import jsonable_encoder
from pydantic import BaseModel

app = FastAPI()
fake_db: dict[str, dict] = {}


class Item(BaseModel):
    title: str
    timestamp: datetime
    description: str | None = None


@app.put("/items/{item_id}")
async def update_item(item_id: str, item: Item):
    json_compatible_item = jsonable_encoder(item)
    fake_db[item_id] = json_compatible_item
    return {"saved": fake_db[item_id]}

Q4:jsonable_encoder 会返回 JSON 字符串吗?

不会。它返回的是 JSON 兼容的 Python 数据结构。

  • json.dumps(obj):将 Python 对象序列化为 JSON 格式字符串。
  • json.loads(text):将 JSON 格式字符串反序列化为 Python 对象(如 dictlist)。
import json
from datetime import datetime

from fastapi.encoders import jsonable_encoder
from pydantic import BaseModel


class Event(BaseModel):
    name: str
    at: datetime


event = Event(name="deploy", at=datetime(2026, 3, 25, 20, 0, 0))
payload = jsonable_encoder(event)   # dict
text = json.dumps(payload)          # str

Q5:为什么 FastAPI 返回模型时通常不需要手动调用它?

因为 FastAPI 内部已经会做编码与序列化。只有在“你自己处理持久化/转发”时才需要显式调用 jsonable_encoder

16. 请求体更新

Q1:PUTPATCH 在更新场景有什么区别?

PUT 语义是“整体替换”,而 PATCH 语义是“部分更新”。如果用 PUT 只提交了部分字段,未提交的字段会被模型默认值覆盖。

Q2:用 PUT 的标准替换写法是什么?

from fastapi import FastAPI
from fastapi.encoders import jsonable_encoder
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):
    name: str | None = None
    description: str | None = None
    price: float | None = None
    tax: float = 10.5
    tags: list[str] = []


items: dict[str, dict] = {
    "foo": {"name": "Foo", "price": 50.2},
    "bar": {"name": "Bar", "description": "The bartenders", "price": 62, "tax": 20.2},
}


@app.put("/items/{item_id}", response_model=Item)
async def replace_item(item_id: str, item: Item):
    encoded = jsonable_encoder(item)
    items[item_id] = encoded
    return encoded

Q3:为什么 PUT 可能会“误覆盖”?

如果请求体没有包含某些字段,它们会被模型默认值填充并写回,从而覆盖原有值。

# 假设原始数据
# {"name": "Bar", "description": "The bartenders", "price": 62, "tax": 20.2}

# PUT 仅传:
# {"name": "Barz", "price": 3, "description": null}

# 结果会变成:
# {"name": "Barz", "description": null, "price": 3, "tax": 10.5}
# 注意 tax 被默认值覆盖

Q4:如何用 PATCH 实现真正的“局部更新”?

核心是组合使用两个步骤:

  • model_dump(exclude_unset=True):只提取请求里“明确传入”的字段,避免把未传字段用默认值写回数据库。
  • model_copy(update=...):在已有模型基础上合并更新字段,生成新的已更新模型,保持类型与校验行为一致。
from fastapi import FastAPI
from fastapi.encoders import jsonable_encoder
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):
    name: str | None = None
    description: str | None = None
    price: float | None = None
    tax: float = 10.5
    tags: list[str] = []


items: dict[str, dict] = {
    "foo": {"name": "Foo", "price": 50.2},
    "bar": {"name": "Bar", "description": "The bartenders", "price": 62, "tax": 20.2},
}


@app.patch("/items/{item_id}", response_model=Item)
async def update_item(item_id: str, item: Item):
    stored = items[item_id]
    stored_model = Item(**stored)
    update_data = item.model_dump(exclude_unset=True)
    updated = stored_model.model_copy(update=update_data)
    items[item_id] = jsonable_encoder(updated)
    return updated

Q5:部分更新的完整流程?

  1. 读取已有数据并构建 Pydantic 模型。
  2. 对输入模型执行 model_dump(exclude_unset=True)
  3. model_copy(update=...) 合并。
  4. jsonable_encoder 转成可持久化结构。
  5. 保存并返回更新结果。

Q6:如果希望“空字段也允许省略”,该怎么设计模型?

更新模型应把字段设置为可选(带默认值或 None),与“创建模型”区分开。建议用 ItemCreate / ItemUpdate 这样的模型拆分。

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


class ItemCreate(BaseModel):
    name: str
    price: float
    description: str | None = None


class ItemUpdate(BaseModel):
    name: str | None = None
    price: float | None = None
    description: str | None = None


@app.post("/items/", response_model=ItemCreate)
async def create_item(item: ItemCreate):
    return item


@app.patch("/items/{item_id}")
async def patch_item(item_id: str, item: ItemUpdate):
    return {"item_id": item_id, "updates": item.model_dump(exclude_unset=True)}
17. 依赖项

Q1:什么是依赖注入?

通常情况下,你写一个函数时需要在函数内部自己创建它依赖的资源(例如数据库连接、Token 解析器)。而在 FastAPI 中,你只需要在参数里声明“我需要什么”,FastAPI 会在执行路径函数前自动准备并注入这些依赖。

核心区别:声明 vs 创建

  • 传统方式:每个函数都自己处理依赖创建与清理,重复代码多,改动范围大。
  • 依赖注入方式:路径函数专注业务逻辑,依赖由框架统一提供。

为什么重要

  • 代码复用:同一鉴权、分页、数据库会话逻辑可被多个接口复用。
  • 解耦:路径函数不关心依赖如何构建,只关心如何使用。
  • 易测试:测试时可替换真实依赖为测试桩/假对象,不必改业务函数内部实现。

Q2:最小可用依赖怎么写?

依赖本质就是普通函数(或可调用对象),再通过 Depends(...) 声明到路径函数参数中。

from typing import Annotated, Any

from fastapi import Depends, FastAPI

app = FastAPI()


async def common_parameters(
    q: str | None = None,
    skip: int = 0,
    limit: int = 100,
) -> dict[str, Any]:
    return {"q": q, "skip": skip, "limit": limit}


@app.get("/items/")
async def read_items(
    commons: Annotated[dict[str, Any], Depends(common_parameters)],
):
    return commons


@app.get("/users/")
async def read_users(
    commons: Annotated[dict[str, Any], Depends(common_parameters)],
):
    return commons

Q3:Depends 使用时最容易犯什么错?

不要在 Depends 里手动调用依赖函数。应写 Depends(common_parameters),不要写 Depends(common_parameters())

Q4:如何减少重复声明依赖的代码?

推荐使用 Annotated 类型别名,把依赖声明成可复用类型。

from typing import Annotated, Any

from fastapi import Depends, FastAPI

app = FastAPI()


async def common_parameters(
    q: str | None = None,
    skip: int = 0,
    limit: int = 100,
) -> dict[str, Any]:
    return {"q": q, "skip": skip, "limit": limit}


CommonsDep = Annotated[dict[str, Any], Depends(common_parameters)]


@app.get("/orders/")
async def read_orders(commons: CommonsDep):
    return commons


@app.get("/reports/")
async def read_reports(commons: CommonsDep):
    return commons

Q5:依赖函数必须是 async def 吗?

不必须。defasync def 都可以,且可与路径函数混用。FastAPI 会按上下文正确执行。

from typing import Annotated

from fastapi import Depends, FastAPI

app = FastAPI()


def sync_dep() -> str:
    return "sync-ok"


async def async_dep() -> str:
    return "async-ok"


@app.get("/mix")
async def read_mix(
    a: Annotated[str, Depends(sync_dep)],
    b: Annotated[str, Depends(async_dep)],
):
    return {"a": a, "b": b}

Q6:依赖会不会进入 OpenAPI 文档?

会。依赖及其子依赖中声明的参数与校验规则会被合并进同一 OpenAPI Schema,并在 /docs 中可见。

Q7:依赖注入在工程实践中的价值是什么?

  • 复用:共享分页、排序、过滤、鉴权等逻辑。
  • 解耦:路径函数专注业务,基础设施逻辑放进依赖。
  • 可组合:依赖可以再依赖,形成分层能力树。
  • 可测试:依赖天然适合替换/覆盖,方便测试隔离。
18. 安全

Q1:FastAPI 的“安全”到底覆盖什么?

核心是两件事:认证(Authentication)授权(Authorization)。认证解决“你是谁”,授权解决“你能做什么”。FastAPI 通过标准化安全工具,减少你手写协议细节的成本。

Q2:为什么官方强调 OAuth2?

OAuth2 是目前 Web API 里最主流的授权规范,覆盖第三方登录与令牌授权等常见场景。像“使用 Google/GitHub 登录”底层都和它相关。

  • OAuth2 本身不负责传输加密,生产环境仍必须使用 HTTPS。

Q3:OpenID Connect 和 OAuth2 是什么关系?

OpenID Connect(OIDC)是建立在 OAuth2 之上的身份层标准,用于更统一地表达“登录用户是谁”。它让不同提供商的行为更可互操作。

Q4:FastAPI 与 OpenAPI 安全方案如何对应?

FastAPI 基于 OpenAPI,会把安全声明自动写入文档,所以 /docs 能直接出现授权输入框并参与调试。

Q5:FastAPI 安全工具怎么开始用?

fastapi.security 模块入手,把安全组件作为依赖注入到路径函数中。因为安全天然是“横切关注点”:多个接口复用同一套登录态与权限判断。用依赖可做到:统一校验、分层组合(当前用户→活跃用户→管理员)、自动文档、低重复。

from typing import Annotated

from fastapi import Depends, FastAPI
from fastapi.security import OAuth2PasswordBearer

app = FastAPI()

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")


@app.get("/users/me")
async def read_users_me(token: Annotated[str, Depends(oauth2_scheme)]):
    return {"access_token": token, "token_type": "bearer"}
19. 中间件

Q1:什么是中间件?它和路由函数是什么关系?

中间件是位于“请求进入应用”和“响应离开应用”两端的统一处理层。

  • 请求进入时先经过中间件。
  • 中间件再把请求交给路由处理。
  • 路由返回响应后,中间件还能再加工一次响应。

Q2:如何创建一个 HTTP 中间件?

通过定义一个带有 @app.middleware("http") 装饰器的异步函数来实现。函数必须接收 requestcall_next

  • request:包含请求的所有信息。
  • call_next:一个函数,它会接收 request 作为参数,并返回路径函数产生的 response
from fastapi import FastAPI, Request

app = FastAPI()


@app.middleware("http")
async def simple_middleware(request: Request, call_next):
    response = await call_next(request)
    return response

Q3:如何在中间件里统计耗时并写入响应头?

推荐使用 time.perf_counter() 进行高精度计时,并在响应头中添加自定义字段。

import time

from fastapi import FastAPI, Request

app = FastAPI()


@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    start = time.perf_counter()
    response = await call_next(request)
    process_time = time.perf_counter() - start
    response.headers["X-Process-Time"] = f"{process_time:.6f}"
    return response

Q4:中间件里“前置逻辑”和“后置逻辑”怎么写?

await call_next(request) 之前写前置逻辑,在它之后写后置逻辑。

Q5:多个中间件的执行顺序是什么?

中间件是“栈式包裹”结构:后添加的在最外层。

  • 请求阶段:最外层先执行。
  • 响应阶段:最外层最后执行。
from fastapi import FastAPI, Request

app = FastAPI()


@app.middleware("http")
async def middleware_a(request: Request, call_next):
    response = await call_next(request)
    response.headers["X-A"] = "A"
    return response


@app.middleware("http")
async def middleware_b(request: Request, call_next):
    response = await call_next(request)
    response.headers["X-B"] = "B"
    return response

在这个示例里,middleware_b 后注册,因此请求先到 B,再到 A;响应返回时先经过 A,再经过 B。

Q6:中间件与依赖项(尤其 yield 依赖)和后台任务的执行先后如何?

核心执行流程(从请求进入到结束)如下:

  1. 中间件(Request 阶段)先接收请求。
  2. 依赖项进入阶段:若是 yield 依赖,会先执行到 yield 之前(如创建 DB 会话)。
  3. 路径函数执行业务逻辑并生成响应。
  4. 依赖项退出阶段:执行 yield 之后的清理代码(如 db.close())。
  5. 中间件(Response 阶段)处理并返回响应。
  6. 后台任务在响应发送后执行。

容易混淆的点yield 依赖不是普通依赖,它具备生命周期管理能力,像上下文管理器一样能在“前后”两个阶段执行代码;而普通 return 依赖只负责一次性提供值。

特性 普通依赖(return yield 依赖
执行方式 执行一次并返回结果 前置执行 → yield 暂停 → 后置清理
典型用途 配置读取、参数组装、简单校验 数据库连接、会话、文件句柄等资源管理
清理机制 无内置清理阶段 yield 后代码自动执行清理
from collections.abc import Generator


def get_db() -> Generator[str, None, None]:
    db = "db-session"
    try:
        yield db
    finally:
        # 响应完成后执行清理
        print("close db session")
20. CORS

Q1:什么是 CORS?为什么前后端分离经常遇到它?

CORS 是浏览器的安全机制:当前端页面和后端 API 不同源(协议/域名/端口任一不同)时,浏览器会限制跨域请求,除非后端明确返回允许跨域的响应头。

Q2:什么叫“不同源(Origin)”?

源由 协议 + 域名 + 端口 组成。任意一个不同,就是不同源。

  • http://localhost:3000http://localhost:8000:端口不同。
  • http://localhost:8000https://localhost:8000:协议不同。

Q3:FastAPI 里如何正确开启 CORS?

推荐使用 CORSMiddleware,并显式列出允许的前端源。

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

origins = [
    "http://localhost:3000",
    "http://127.0.0.1:3000",
]

app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "PATCH", "DELETE", "OPTIONS"],
    allow_headers=["Authorization", "Content-Type"],
    expose_headers=["X-Request-ID", "X-Process-Time"],
    max_age=600,
)


@app.get("/health")
async def health_check():
    return {"status": "ok"}

Q4:allow_origins=["*"] 能不能直接用?

allow_origins=["*"] 表示允许任意来源访问,适合本地快速联调,不适合生产默认长期开启。

allow_credentials 决定浏览器是否允许在跨源请求中携带凭证(如 Cookie、Authorization 等):

  • allow_credentials=False(默认):浏览器不会按跨域凭证模式处理请求/响应。
  • allow_credentials=True:后端明确允许跨域携带凭证,并允许处理相关凭证响应。

一旦开启 allow_credentials=True,就不能再使用 allow_origins=["*"],必须显式列出可信前端源(例如 ["https://myfrontend.com"])。这是为了避免任意站点都能借浏览器上下文发送用户凭证,降低安全风险。

Q5:什么是预检请求?

浏览器在某些跨域请求前会先发 OPTIONS 请求(带 OriginAccess-Control-Request-Method),用来确认后端是否允许该跨域操作。CORSMiddleware 会自动处理这类请求并返回对应 CORS 响应头。

Q6:什么是简单请求?

Origin 的普通请求会直接到业务路由,中间件会在响应中补充 CORS 响应头,浏览器据此决定是否放行前端 JS 读取响应。

Q7:各参数最常用含义是什么?

  • allow_origins:允许跨域的源白名单。
  • allow_origin_regex:按正则匹配源。
  • allow_methods:允许的 HTTP 方法。
  • allow_headers:允许的请求头。
  • allow_credentials:是否允许 Cookie/Authorization 等凭证。
  • expose_headers:允许浏览器 JS 读取的响应头。
  • max_age:预检结果缓存秒数。

Q8:前端读不到自定义响应头怎么办?

需要把该响应头加入 expose_headers。否则即使响应里有该头,浏览器端 JavaScript 也可能拿不到。

app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
    expose_headers=["X-Process-Time", "X-Request-ID"],
)
21. SQL 数据库

Q1:FastAPI 强制使用 SQLModel 吗?

官方结论:不强制。FastAPI 可以配合任意 SQL/NoSQL 工具。选择 SQLModel,是因为它在 FastAPI 场景中能显著减少“重复建模”的成本。

SQLModel = Pydantic + SQLAlchemy

工具 角色 核心功能
SQLAlchemy ORM / 数据库引擎层 负责表映射、SQL 生成、连接与事务。
Pydantic 数据验证层 负责请求/响应校验、类型转换与文档基础。
SQLModel 统一封装层 让同一模型同时具备 Pydantic 校验能力与 ORM 表定义能力。

SQLModel 出现的动机是解决传统 FastAPI 项目中的常见问题:同一业务实体往往要写两份近似代码(Pydantic 模型 + SQLAlchemy 模型),字段变更要改两处,维护成本高。

  • 减少冗余:字段尽量只定义一次。
  • 提高一致性:API 契约与数据库结构更容易同步演进。
  • 增强体验:类型提示与自动补全更连贯。

补充:生产环境通常迁移到 PostgreSQL,并使用迁移工具(如 Alembic)管理 schema 变更。

Q3:生产模式(全异步 + Alembic + 分层装配)推荐模板?

生产实践建议:用 Alembic 管迁移(而不是启动时 create_all)、使用 SQLAlchemy AsyncEngine + AsyncSession,并把 DB 初始化与 app 装配拆分。lifespan 是应用级“启动/关闭钩子”,用于启动时初始化、关闭时清理资源。

app/db.py

from collections.abc import AsyncGenerator

from sqlalchemy.ext.asyncio import (
    AsyncEngine,
    AsyncSession,
    async_sessionmaker,
    create_async_engine,
)

DATABASE_URL = "postgresql+asyncpg://app:password@127.0.0.1:5432/dreamlog"

engine: AsyncEngine = create_async_engine(
    DATABASE_URL,
    pool_pre_ping=True,
)
SessionLocal = async_sessionmaker(
    bind=engine,
    class_=AsyncSession,
    expire_on_commit=False,
)


async def get_session() -> AsyncGenerator[AsyncSession, None]:
    async with SessionLocal() as session:
        yield session

app/models.py

from datetime import datetime

from sqlalchemy import DateTime, String, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


class Base(DeclarativeBase):
    pass


class Hero(Base):
    __tablename__ = "heroes"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100), index=True)
    secret_name: Mapped[str] = mapped_column(String(100))
    age: Mapped[int | None]
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        server_default=func.now(),
    )

app/schemas.py

from pydantic import BaseModel


class HeroCreate(BaseModel):
    name: str
    secret_name: str
    age: int | None = None


class HeroPublic(BaseModel):
    id: int
    name: str
    age: int | None = None

app/routers/heroes.py

from typing import Annotated

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.db import get_session
from app.models import Hero
from app.schemas import HeroCreate, HeroPublic

router = APIRouter(prefix="/heroes", tags=["heroes"])
SessionDep = Annotated[AsyncSession, Depends(get_session)]


@router.post("/", response_model=HeroPublic)
async def create_hero(payload: HeroCreate, session: SessionDep) -> HeroPublic:
    hero = Hero(name=payload.name, secret_name=payload.secret_name, age=payload.age)
    session.add(hero)
    await session.commit()
    await session.refresh(hero)
    return HeroPublic.model_validate(hero, from_attributes=True)


@router.get("/{hero_id}", response_model=HeroPublic)
async def read_hero(hero_id: int, session: SessionDep) -> HeroPublic:
    hero = await session.get(Hero, hero_id)
    if hero is None:
        raise HTTPException(status_code=404, detail="Hero not found")
    return HeroPublic.model_validate(hero, from_attributes=True)


@router.get("/", response_model=list[HeroPublic])
async def read_heroes(session: SessionDep) -> list[HeroPublic]:
    result = await session.execute(select(Hero))
    heroes = result.scalars().all()
    return [HeroPublic.model_validate(h, from_attributes=True) for h in heroes]

app/main.py

from contextlib import asynccontextmanager

from fastapi import FastAPI

from app.db import engine
from app.routers.heroes import router as heroes_router


@asynccontextmanager
async def lifespan(app: FastAPI):
    # lifespan = 应用级“启动/关闭钩子”:
    # 启动时做一次初始化,关闭时做一次清理。
    # 生产环境不在这里 create_all,结构迁移统一由 Alembic 执行。
    yield
    await engine.dispose()


app = FastAPI(lifespan=lifespan, title="Dream Log API")
app.include_router(heroes_router)

Alembic(命令示例)

alembic revision --autogenerate -m "create heroes table"
alembic upgrade head
21. 更大的应用

Q1:为什么要把 FastAPI 拆成“更大的应用”结构?

当接口数量增加后,把所有路由都写在一个文件会迅速变得难维护。官方推荐使用 APIRouter + 分模块目录结构,让路由、依赖、模型按领域拆分,提升可读性、可测试性与团队协作效率。

  • 按领域拆分:如 usersitemsauth
  • 统一前缀与标签:每个 router 自带 prefix/tags
  • 依赖可分层复用:应用级、路由级、路径级逐层叠加。

Q3:APIRouter 怎么定义与拆分?

每个业务文件创建自己的 router = APIRouter(...),并在内部声明路径操作。推荐在 router 上集中声明 prefixtagsresponses

from typing import Annotated

from fastapi import APIRouter, Depends, HTTPException

router = APIRouter(
    prefix="/items",
    tags=["items"],
    responses={404: {"description": "Not found"}},
)

Q4:如何在主应用中注册多个 Router?

main.py 里通过 app.include_router(...) 组装。这样每个模块独立开发,主入口只负责“装配”。

from fastapi import FastAPI

from app.routers import items, users

app = FastAPI(title="Dream Log API")

app.include_router(users.router)
app.include_router(items.router)


@app.get("/")
async def root() -> dict[str, str]:
    return {"message": "API is running"}

Q5:依赖项可以放在哪几层?执行关系是什么?

  • 应用级FastAPI(dependencies=[Depends(...)]),作用于全站。
  • 路由级APIRouter(..., dependencies=[Depends(...)]),作用于该组路由。
  • 路径级:写在具体路径函数参数中,粒度最细。

执行时会叠加:应用级 → 路由级 → 路径级。越往后越接近具体业务。

Q6:如何写“可复用且可测试”的共享依赖文件?

可把鉴权、请求头校验等放到 dependencies.py,供多个 router 引用。推荐使用 Annotated + Header() 明确输入来源。

from typing import Annotated

from fastapi import Header, HTTPException


async def verify_token(x_token: Annotated[str, Header()]) -> str:
    if x_token != "fake-super-secret-token":
        raise HTTPException(status_code=400, detail="X-Token invalid")
    return x_token


async def verify_key(x_key: Annotated[str, Header()]) -> str:
    if x_key != "fake-super-secret-key":
        raise HTTPException(status_code=400, detail="X-Key invalid")
    return x_key

Q7:如何在 Router 级统一套用依赖与错误文档?

from fastapi import APIRouter, Depends

from app.dependencies import verify_key, verify_token

router = APIRouter(
    prefix="/admin",
    tags=["admin"],
    dependencies=[Depends(verify_token), Depends(verify_key)],
    responses={403: {"description": "Forbidden"}},
)


@router.get("/stats")
async def read_admin_stats() -> dict[str, int]:
    return {"active_users": 128}

这样该 router 下所有接口都会自动执行依赖,无需每个函数重复声明。