Python 基础体系 · 第 76/112 篇。示例统一以 Python 3.14 为语言基线；第三方库使用与其兼容的现代稳定版本，版本敏感行为会单独说明。

FastAPI 完整基础：路由、依赖注入、校验、异步和生命周期

FastAPI 是一个基于 Python 类型注解构建 HTTP API 的 Web 框架。它负责把 HTTP 请求转换成 Python 函数参数，把函数返回值转换成 HTTP 响应，并利用类型注解生成 OpenAPI 文档。它运行在 Starlette 提供的 ASGI 能力之上，数据校验和序列化主要由 Pydantic 完成。(fastapi.tiangolo.com)

本文使用 Python 3.14 语法。示例中的 str | None、list[Item]、Annotated 和异步上下文管理器都属于现代 Python 写法；实际项目仍应锁定 FastAPI、Starlette、Pydantic 和 ASGI 服务器的兼容版本，而不能只根据 Python 版本推断整个依赖栈已经兼容。

---

1. 先建立运行模型：请求如何到达 Python 函数

一个最小 FastAPI 应用包含两个对象：

from fastapi import FastAPI

app = FastAPI()


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

app 是应用对象，@app.get("/") 是路由注册操作，read_root 是路径操作函数。这里的“路径操作”指的是 HTTP 方法和 URL 路径共同定义的一种操作，例如：

GET /users
POST /users
GET /users/{user_id}

HTTP 方法不是普通的函数参数，而是路由匹配条件的一部分。GET /users 和 POST /users 可以对应不同函数，因为它们代表不同的 HTTP 操作。

启动应用：

python -m pip install "fastapi[standard]"
fastapi dev main.py

也可以直接使用 ASGI 服务器：

uvicorn main:app --reload

其中：

• main 表示 main.py 模块；
• app 表示模块中的 FastAPI 应用对象；
• --reload 适合开发环境，会监视代码变化并重新加载进程；
• 生产环境不应把 --reload 当作进程管理方案。

请求 GET / 时，数据流大致如下：

sequenceDiagram
    participant C as 客户端
    participant S as ASGI 服务器
    participant A as FastAPI 应用
    participant R as 路由系统
    participant F as 路径操作函数

    C->>S: HTTP 请求
    S->>A: scope + receive + send
    A->>R: 查找 HTTP 方法和路径
    R->>F: 提供已解析、已校验的参数
    F-->>R: 返回 Python 对象
    R-->>A: 序列化响应
    A-->>S: ASGI response 事件
    S-->>C: HTTP 响应

FastAPI 并不是直接从 socket 中读取请求。它接收 ASGI 服务器传入的 scope、receive 和 send。ASGI 3 将应用定义为异步调用对象：

async def application(scope, receive, send):
    ...

scope 描述连接或请求，receive 用于异步接收事件，send 用于异步发送事件。HTTP 请求体不是一次性放入 scope，而是通过事件传入；WebSocket 则会在连接生命周期内接收多个事件。(asgi.readthedocs.io)

因此，FastAPI 的路径操作函数是应用层函数，而 ASGI callable 是服务器和框架之间的协议边界。你通常不需要手写 scope、receive 和 send，但理解这一层有助于解释中间件、异步并发、WebSocket 和生命周期。

---

2. 路由：HTTP 方法、路径参数和匹配顺序

2.1 路径参数

路径中的 {user_id} 是路径参数：

from fastapi import FastAPI

app = FastAPI()


@app.get("/users/{user_id}")
async def get_user(user_id: int) -> dict[str, int]:
    return {"user_id": user_id}

访问：

curl http://127.0.0.1:8000/users/42

预期响应：

{"user_id":42}

函数参数 user_id: int 同时表达了两件事：

1. 参数名 user_id 必须和路径模板中的 {user_id} 对应；
2. 请求中的文本 "42" 必须能被解析为整数。

访问 /users/abc 时，路由路径本身能够匹配，但参数校验失败，FastAPI 不会调用 get_user，而是返回验证错误。这里要区分“路由匹配失败”和“参数解析失败”：

• /unknown：没有匹配的路径操作，通常是 404；
• /users/abc：匹配了 /users/{user_id}，但 abc 不能满足 int，属于请求参数校验错误；
• /users/42：路由和参数都成立，才会调用函数。

2.2 固定路径必须放在参数路径之前

路由通常按照注册顺序匹配：

@app.get("/users/me")
async def current_user() -> dict[str, str]:
    return {"user": "current"}


@app.get("/users/{user_id}")
async def get_user(user_id: str) -> dict[str, str]:
    return {"user": user_id}

如果反过来注册，/users/{user_id} 可能先把 me 当作 user_id。因此，固定路径 /users/me 应先于通配路径 /users/{user_id} 注册。FastAPI 文档明确指出路径操作按顺序求值，重复注册同一方法和路径时，先注册的操作会先匹配。(fastapi.tiangolo.com)

这不是类型校验可以补救的问题。即使 user_id 声明为 int，/users/me 也会先进入参数转换阶段；它可能最终产生验证错误，而不是落到固定路径处理函数。

2.3 查询参数、路径参数和请求体

FastAPI 根据函数签名推断参数来源：

from pydantic import BaseModel


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


@app.put("/items/{item_id}")
async def update_item(
    item_id: int,
    item: Item,
    q: str | None = None,
) -> dict:
    return {
        "item_id": item_id,
        "query": q,
        "item": item.model_dump(),
    }

调用：

curl -X PUT \
  'http://127.0.0.1:8000/items/7?q=book' \
  -H 'Content-Type: application/json' \
  -d '{"name":"Python","price":59.9}'

参数来源是：

参数	来源	原因
item_id	路径	名称出现在 /items/{item_id} 中
q	查询字符串	是普通标量类型，且不在路径中
item	JSON 请求体	类型是 Pydantic 模型

FastAPI 官方文档采用同样的判定规则：路径中出现的参数从路径读取，普通标量参数默认作为查询参数，Pydantic 模型参数作为请求体。(fastapi.tiangolo.com)

如果想显式声明来源，可以使用 Path、Query、Body：

from typing import Annotated

from fastapi import Body, Path, Query


@app.put("/items/{item_id}")
async def update_item(
    item_id: Annotated[int, Path(gt=0)],
    item: Annotated[Item, Body()],
    q: Annotated[str | None, Query(max_length=20)] = None,
) -> dict:
    return {
        "item_id": item_id,
        "query": q,
        "item": item.model_dump(),
    }

显式声明的价值不是让框架“更能解析”，而是让约束、来源和 OpenAPI 文档更加清楚。

2.4 路由配置和路由函数参数不是一回事

响应状态码、标签和摘要等配置属于装饰器：

from fastapi import status


@app.post(
    "/items",
    status_code=status.HTTP_201_CREATED,
    tags=["items"],
    summary="创建商品",
)
async def create_item(item: Item) -> Item:
    return item

status_code 是路径操作装饰器的配置，而不是函数参数。它会影响实际响应，也会进入生成的 OpenAPI 描述。(fastapi.tiangolo.com)

---

3. 校验：从输入结构到业务条件

3.1 Pydantic 模型负责数据边界

Pydantic 模型描述的是数据结构和字段约束：

from pydantic import BaseModel, Field


class ItemCreate(BaseModel):
    name: str = Field(min_length=1, max_length=100)
    price: float = Field(gt=0)
    tags: list[str] = Field(default_factory=list)

这段模型表达：

• name 必须是字符串；
• 字符串长度至少为 1，最多为 100；
• price 必须大于 0；
• tags 是字符串列表；
• 每次创建模型时，tags 使用新的列表实例。

使用 default_factory=list 比直接写 tags: list[str] = [] 更能明确表达“每个实例拥有独立默认列表”的意图。

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

输入：

{
  "name": "键盘",
  "price": 199,
  "tags": ["hardware"]
}

成功后，item 已经是 ItemCreate 实例，而不是未经处理的原始字典：

item.name
item.price
item.model_dump()

Pydantic v2 中常用的模型操作包括：

item.model_dump()          # Python 字典
item.model_dump_json()     # JSON 字符串
ItemCreate.model_validate(data)
ItemCreate.model_json_schema()

FastAPI 会把模型字段约束用于请求校验、OpenAPI Schema 和响应序列化。

3.2 类型校验不等于业务校验

“价格是正数”是字段约束；“商品名称不能和已有商品重复”是业务规则。后者通常需要访问数据库，不能仅靠 Field 表达：

from fastapi import HTTPException


@app.post("/items")
async def create_item(item: ItemCreate) -> ItemCreate:
    if item.name == "已存在商品":
        raise HTTPException(
            status_code=409,
            detail="商品名称已存在",
        )

    return item

输入数据不符合模型时，路径操作函数不会执行。业务代码主动发现冲突时，则由应用抛出 HTTPException。两者的因果路径不同：

请求体解析
  ├─ JSON 无法解析       -> 请求错误
  ├─ 字段类型/约束失败    -> RequestValidationError
  └─ 模型校验成功
       └─ 业务逻辑执行
            ├─ 资源冲突    -> HTTPException 409
            └─ 成功         -> 返回响应

FastAPI 会在请求数据无效时内部产生 RequestValidationError；也可以通过异常处理器统一改变错误响应格式。(fastapi.tiangolo.com)

3.3 响应模型是输出边界

输入模型和输出模型通常不应复用：

from pydantic import EmailStr


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


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


@app.post("/users", response_model=UserPublic)
async def create_user(user: UserCreate) -> UserPublic:
    # 真实项目中应先哈希密码，再持久化
    return user

虽然返回对象中包含 password，但 response_model=UserPublic 会将未声明的字段排除。响应模型同时承担：

1. 校验返回数据；
2. 生成响应 Schema；
3. 序列化；
4. 限制输出字段。

因此，响应模型不只是文档工具，也是防止敏感字段泄露的边界。FastAPI 文档特别强调，响应模型会过滤未声明字段，这对安全很重要。(fastapi.tiangolo.com)

一个常见错误是把数据库实体、ORM 对象或内部领域对象直接作为公开响应。更可靠的流程是：

数据库对象
   -> 映射为公共响应模型
   -> response_model 再次约束
   -> JSON 响应

类型注解和 response_model 可以同时存在，但当二者冲突时，response_model 是 FastAPI 用于响应处理的优先声明。(fastapi.tiangolo.com)

3.4 自定义校验器应该放在模型层

Pydantic v2 中可以用 field_validator 和 model_validator 表达跨字段或字段级规则：

from pydantic import BaseModel, field_validator


class UserCreate(BaseModel):
    username: str
    password: str

    @field_validator("username")
    @classmethod
    def username_must_not_contain_space(cls, value: str) -> str:
        if any(char.isspace() for char in value):
            raise ValueError("用户名不能包含空白字符")
        return value

这个校验器适合“只依赖当前字段”的规则。需要比较多个字段时，可以使用模型级校验；需要查数据库、调用外部服务或依赖当前用户身份时，不应把这些副作用塞进纯数据模型校验器，而应放入依赖或业务服务层。

---

4. 依赖注入：把请求前置条件组织成依赖图

4.1 依赖注入解决什么问题

依赖注入指的是：函数声明自己需要什么，框架负责创建或查找这些对象，然后将结果传入函数。

from typing import Annotated

from fastapi import Depends, Header, HTTPException


async def get_request_id(
    x_request_id: Annotated[str | None, Header()] = None,
) -> str:
    return x_request_id or "generated-request-id"


RequestId = Annotated[str, Depends(get_request_id)]


@app.get("/debug")
async def debug(request_id: RequestId) -> dict[str, str]:
    return {"request_id": request_id}

路径操作函数不直接调用 get_request_id()。Depends(get_request_id) 是声明，FastAPI 在请求处理过程中调用它并注入结果。官方文档也明确指出，不应手动直接调用依赖函数，依赖函数由 FastAPI 执行和注入。(fastapi.tiangolo.com)

4.2 依赖是有向图，而不是简单的“工具函数列表”

依赖可以继续依赖其他依赖：

from fastapi import Query


async def get_pagination(
    limit: Annotated[int, Query(ge=1, le=100)] = 20,
    offset: Annotated[int, Query(ge=0)] = 0,
) -> tuple[int, int]:
    return limit, offset


async def get_current_user(
    request_id: RequestId,
    pagination: Annotated[tuple[int, int], Depends(get_pagination)],
) -> dict[str, str]:
    return {
        "user": "demo",
        "request_id": request_id,
        "pagination": str(pagination),
    }


CurrentUser = Annotated[dict[str, str], Depends(get_current_user)]


@app.get("/profile")
async def profile(user: CurrentUser) -> dict[str, str]:
    return user

一次请求的解析顺序可以表示为：

profile
  └── get_current_user
       ├── get_request_id
       └── get_pagination

FastAPI 先解析叶子依赖，再将结果传给上层依赖，最后调用路径操作函数。依赖及其子依赖的参数声明也会进入 OpenAPI 文档。(fastapi.tiangolo.com)

4.3 依赖缓存和请求级复用

同一个请求中，同一个依赖默认会缓存结果：

async def get_settings() -> dict[str, str]:
    print("load settings")
    return {"env": "dev"}


Settings = Annotated[dict[str, str], Depends(get_settings)]


async def dependency_a(settings: Settings) -> str:
    return settings["env"]


async def dependency_b(settings: Settings) -> str:
    return settings["env"]

如果一个路径操作同时依赖 dependency_a 和 dependency_b，默认情况下 get_settings 通常只需解析一次，两个上层依赖共享同一请求上下文中的结果。这个缓存是请求级的，不是跨请求的全局缓存。

需要强制每次重新执行时，可以关闭缓存：

FreshSettings = Annotated[
    dict[str, str],
    Depends(get_settings, use_cache=False),
]

关闭缓存的原因应当明确，例如依赖函数刻意读取当前变化的请求状态。否则重复执行会增加开销，甚至造成重复查询或重复创建资源。

4.4 yield 依赖：请求级资源的创建和释放

依赖注入也能管理请求范围资源：

from collections.abc import AsyncIterator


class Session:
    async def close(self) -> None:
        print("session closed")


async def create_session() -> AsyncIterator[Session]:
    session = Session()
    try:
        yield session
    finally:
        await session.close()


SessionDep = Annotated[Session, Depends(create_session)]


@app.get("/orders")
async def list_orders(session: SessionDep) -> dict[str, str]:
    return {"status": "queried"}

执行顺序是：

进入依赖
  -> 创建 Session
  -> yield Session
  -> 执行路径操作
  -> finally 中关闭 Session

finally 必须存在，因为路径操作可能正常返回，也可能抛出异常。数据库会话、事务、临时文件和锁等资源都需要考虑异常路径。

这里的资源范围是“一次请求”，不能把它误当成应用级连接池。连接池通常在生命周期中创建；从连接池中借出的单个会话则可以通过 yield 依赖按请求管理。

---

5. 异步：并发等待，不是自动并行

5.1 async def 和 await 的实际含义

异步函数本身不会自动创造并行执行。await 的作用是：当前协程等待某个异步操作时，将控制权交还事件循环，让其他任务运行。

Python asyncio 采用协作式调度：事件循环一次运行一个任务；任务在等待 Future 或 I/O 时，事件循环才有机会运行其他任务。(docs.python.org)

import asyncio
from time import perf_counter


async def fetch(name: str, delay: float) -> str:
    await asyncio.sleep(delay)
    return name


async def main() -> None:
    start = perf_counter()

    first, second = await asyncio.gather(
        fetch("first", 1),
        fetch("second", 1),
    )

    elapsed = perf_counter() - start
    print(first, second)
    print(f"{elapsed:.2f}s")


asyncio.run(main())

两个任务都等待 1 秒时，总时间接近 1 秒，而不是简单顺序执行时的约 2 秒。原因不是 CPU 同时执行了两个 Python 函数，而是第一个任务等待时，事件循环执行了第二个任务。

在 FastAPI 中，适合异步路径操作的条件是：调用链上的外部 I/O 库支持 await。

@app.get("/remote")
async def remote() -> dict:
    result = await async_http_client.get("https://example.test")
    return result

5.2 同步函数并不会阻塞 FastAPI 事件循环

FastAPI 对它主动调用的普通 def 路径操作和依赖，通常会放到外部线程池中执行；async def 则在事件循环中执行。同步和异步依赖也可以混用。(fastapi.tiangolo.com)

@app.get("/sync")
def sync_endpoint() -> dict[str, str]:
    result = blocking_library_call()
    return {"result": result}

但这个规则有边界：如果你在 async def 中直接调用阻塞函数，FastAPI 不会自动替你把这一次内部调用移入线程池：

@app.get("/bad")
async def bad_endpoint() -> dict[str, str]:
    result = blocking_library_call()  # 可能阻塞事件循环
    return {"result": result}

在异步路径操作中，阻塞调用会占住事件循环，其他请求无法获得调度机会。修复方式通常有三类：

1. 使用支持异步的数据库、HTTP 或文件库；
2. 将路径操作本身写成普通 def，让 FastAPI 调度它；
3. 明确将少量阻塞工作移入线程，例如使用 asyncio.to_thread。

import asyncio


@app.get("/thread")
async def thread_endpoint() -> dict[str, str]:
    result = await asyncio.to_thread(blocking_library_call)
    return {"result": result}

to_thread 适合 I/O 型阻塞工作，不适合期待绕过 CPython 线程执行限制的重 CPU 计算。CPU 密集型任务通常需要进程、专用任务队列或外部计算服务。

5.3 不要把 asyncio.create_task 当成可靠任务队列

@app.post("/emails")
async def send_email() -> dict[str, str]:
    asyncio.create_task(send_email_later())
    return {"status": "accepted"}

这个任务与当前进程和事件循环绑定。进程重启、容器被终止或部署滚动更新时，任务可能尚未完成就消失。它适合短生命周期、可丢失的辅助工作，不适合作为可靠消息投递机制。

如果任务必须重试、持久化、跨进程执行，应使用消息队列或专门的任务系统，而不是依赖 Web 进程中的后台协程。

---

6. 生命周期：应用资源的启动、运行和关闭

6.1 生命周期和请求范围不是同一个概念

生命周期是应用或事件循环级别的状态转换：

未启动
  -> startup
  -> 运行中：接收请求
  -> 停止接收新请求
  -> shutdown
  -> 已关闭

数据库连接池、机器学习模型、共享 HTTP 客户端等资源通常属于应用生命周期；数据库事务、请求身份和单次文件句柄通常属于请求生命周期。

ASGI Lifespan 协议规定了 lifespan.startup 和 lifespan.shutdown 事件。服务器应等待启动完成后再处理请求，并等待关闭完成后再终止。生命周期通常在每个处理请求的事件循环中执行；多进程部署时，每个进程都有自己的生命周期。(asgi.readthedocs.io)

6.2 使用 lifespan 管理共享资源

from collections.abc import AsyncIterator
from contextlib import asynccontextmanager

from fastapi import FastAPI, Request


class AsyncClient:
    async def get(self, path: str) -> dict[str, str]:
        return {"path": path}

    async def aclose(self) -> None:
        print("client closed")


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    client = AsyncClient()
    app.state.client = client

    try:
        yield
    finally:
        await client.aclose()


app = FastAPI(lifespan=lifespan)


@app.get("/health/dependency")
async def health_dependency(request: Request) -> dict[str, str]:
    client: AsyncClient = request.app.state.client
    return await client.get("/health")

yield 之前的代码是启动阶段，yield 之后的代码是关闭阶段。FastAPI 的 lifespan 参数接收异步上下文管理器；如果使用 lifespan，旧式的 startup 和 shutdown 事件处理器不会同时执行。(fastapi.tiangolo.com)

这个例子中的关键状态变化是：

启动前：
    app.state 中没有 client

启动后：
    app.state.client = AsyncClient()

请求期间：
    请求通过 request.app.state.client 访问共享客户端

关闭时：
    await client.aclose()
    资源释放

资源必须在同一个事件循环中创建和使用。ASGI Lifespan 规范特别指出，连接池等对象不应在一个事件循环中创建后又被其他事件循环共享。(asgi.readthedocs.io)

6.3 多进程意味着多份资源

假设使用四个 worker：

uvicorn main:app --workers 4

生命周期启动代码会在四个进程中分别执行。因此：

• 内存中的缓存不是四个进程共享的；
• 模型可能加载四份；
• 每个进程可能创建自己的连接池；
• 应用级计数器不能依赖普通全局变量实现跨进程一致性。

这不是 FastAPI 的特殊缺陷，而是进程隔离的结果。需要共享状态时，应使用数据库、Redis 或其他外部存储；需要控制总资源量时，应结合 worker 数量和每进程连接池大小计算。

---

7. 中间件：包裹整条请求处理链

中间件位于 ASGI 应用外层，可以在请求进入路由前和响应返回后执行逻辑：

from time import perf_counter

from fastapi import Request


@app.middleware("http")
async def add_process_time(request: Request, call_next):
    start = perf_counter()

    response = await call_next(request)

    elapsed = perf_counter() - start
    response.headers["X-Process-Time"] = f"{elapsed:.6f}"
    return response

执行关系近似于：

中间件前置逻辑
  -> 路由匹配
  -> 依赖解析
  -> 路径操作
  -> 响应序列化
中间件后置逻辑

call_next(request) 会将请求交给后续应用，并等待响应。中间件适合实现请求日志、响应头、追踪上下文、CORS 等横切逻辑。FastAPI 基于 Starlette 并实现 ASGI，因此也可以使用 ASGI 中间件。(fastapi.tiangolo.com)

中间件和依赖的范围不同：

• 中间件包裹所有请求，适合观察和修改整体请求/响应；
• 依赖只在声明它的路径操作或路由范围内执行，适合认证、权限、数据库会话和业务前置条件；
• 生命周期只在应用或事件循环启动、关闭时执行，适合共享资源。

---

8. 错误处理：区分客户端错误和服务端错误

8.1 主动返回 HTTP 错误

from fastapi import HTTPException


@app.get("/items/{item_id}")
async def read_item(item_id: int) -> Item:
    item = await find_item(item_id)

    if item is None:
        raise HTTPException(
            status_code=404,
            detail="商品不存在",
        )

    return item

raise HTTPException(...) 会中断当前路径操作和上层依赖，并交给 FastAPI 生成错误响应。相比返回一个普通字典，抛出异常可以保证后续业务代码不会继续执行。

8.2 统一校验错误格式

from fastapi import Request
from fastapi.encoders import jsonable_encoder
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse


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

exc.errors() 包含错误位置、类型和消息，例如：

{
  "loc": ["body", "price"],
  "type": "greater_than",
  "msg": "Input should be greater than 0"
}

生产环境通常不应原样返回内部异常堆栈。请求校验错误可以返回字段级信息；数据库异常、第三方服务异常和编程错误则应记录服务端日志，并向客户端返回稳定的错误码。

8.3 响应校验失败通常意味着服务端缺陷

请求校验失败通常意味着客户端输入不符合接口契约；响应模型校验失败通常意味着服务端代码返回了错误形状。这两者的责任方向不同：

请求模型失败       -> 客户端请求不满足契约
响应模型失败       -> 服务端实现没有满足契约

因此，不能为了“避免报错”而随意把响应类型改成 dict[str, object]。这样虽然减少了约束，也会失去文档、过滤和输出契约。

---

9. 一个可运行的端到端示例

下面的应用把路由、Pydantic v2 模型、依赖、请求级资源、生命周期、异步调用和响应模型组合起来。

# main.py
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from typing import Annotated

from fastapi import Depends, FastAPI, Header, HTTPException, Query, Request, status
from pydantic import BaseModel, Field


class ItemCreate(BaseModel):
    name: str = Field(min_length=1, max_length=100)
    price: float = Field(gt=0)
    tags: list[str] = Field(default_factory=list)


class ItemPublic(ItemCreate):
    id: int


class FakeStore:
    def __init__(self) -> None:
        self.items: dict[int, ItemPublic] = {}
        self.next_id = 1

    async def create(self, item: ItemCreate) -> ItemPublic:
        result = ItemPublic(id=self.next_id, **item.model_dump())
        self.items[self.next_id] = result
        self.next_id += 1
        return result

    async def get(self, item_id: int) -> ItemPublic | None:
        return self.items.get(item_id)

    async def list(self, limit: int, offset: int) -> list[ItemPublic]:
        values = list(self.items.values())
        return values[offset : offset + limit]


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    app.state.store = FakeStore()
    try:
        yield
    finally:
        app.state.store.items.clear()


app = FastAPI(title="Items API", lifespan=lifespan)


async def get_store(request: Request) -> FakeStore:
    return request.app.state.store


StoreDep = Annotated[FakeStore, Depends(get_store)]


async def get_request_id(
    x_request_id: Annotated[str | None, Header()] = None,
) -> str:
    return x_request_id or "generated"


RequestIdDep = Annotated[str, Depends(get_request_id)]


@app.post(
    "/items",
    response_model=ItemPublic,
    status_code=status.HTTP_201_CREATED,
)
async def create_item(
    item: ItemCreate,
    store: StoreDep,
    request_id: RequestIdDep,
) -> ItemPublic:
    if any(existing.name == item.name for existing in store.items.values()):
        raise HTTPException(status_code=409, detail="商品名称已存在")

    result = await store.create(item)
    return result


@app.get("/items", response_model=list[ItemPublic])
async def list_items(
    store: StoreDep,
    limit: Annotated[int, Query(ge=1, le=100)] = 20,
    offset: Annotated[int, Query(ge=0)] = 0,
) -> list[ItemPublic]:
    return await store.list(limit, offset)


@app.get("/items/{item_id}", response_model=ItemPublic)
async def get_item(
    item_id: int,
    store: StoreDep,
) -> ItemPublic:
    item = await store.get(item_id)

    if item is None:
        raise HTTPException(status_code=404, detail="商品不存在")

    return item

启动：

uvicorn main:app --reload

创建商品：

curl -i -X POST http://127.0.0.1:8000/items \
  -H 'Content-Type: application/json' \
  -H 'X-Request-ID: req-001' \
  -d '{"name":"键盘","price":199,"tags":["hardware"]}'

预期响应状态为 201 Created：

{
  "id": 1,
  "name": "键盘",
  "price": 199.0,
  "tags": ["hardware"]
}

查询商品：

curl 'http://127.0.0.1:8000/items?limit=20&offset=0'

发送非法输入：

curl -i -X POST http://127.0.0.1:8000/items \
  -H 'Content-Type: application/json' \
  -d '{"name":"","price":-1}'

此时不会执行 create_item 的业务逻辑，因为 name 和 price 在请求模型阶段已经失败。

这个示例的执行顺序是：

1. 服务器完成 ASGI startup
2. lifespan 创建 FakeStore
3. 请求进入 FastAPI
4. 解析路径、查询参数、请求头和请求体
5. Pydantic 校验 ItemCreate
6. 解析 StoreDep 和 RequestIdDep
7. 调用路径操作函数
8. 通过 ItemPublic 校验并序列化响应
9. 服务器关闭时执行 lifespan 的 finally

测试生命周期时，要让测试客户端进入上下文：

from fastapi.testclient import TestClient

from main import app


def test_create_item() -> None:
    with TestClient(app) as client:
        response = client.post(
            "/items",
            json={"name": "鼠标", "price": 99},
        )

        assert response.status_code == 201
        assert response.json()["name"] == "鼠标"

如果测试客户端没有触发生命周期上下文，依赖启动阶段创建的资源可能没有被初始化。FastAPI 官方文档也提供了围绕生命周期事件进行测试的用法。(fastapi.tiangolo.com)

---

10. FastAPI、ASGI 与 WSGI 的边界

WSGI 把应用抽象为同步调用，并围绕传统 HTTP 请求—响应迭代器设计；ASGI 则使用异步 callable 和事件消息，可以覆盖 HTTP、WebSocket 以及生命周期协议。ASGI 的设计还允许在异步事件循环中处理多个事件，并将同步 WSGI 应用放入线程池中运行。(asgi.readthedocs.io)

因此，选择 ASGI 并不意味着所有代码都必须写成异步：

异步路径操作 + 异步 I/O
    -> 事件循环中协作等待

同步路径操作或依赖
    -> 框架可在线程池执行

异步路径操作 + 直接阻塞调用
    -> 可能阻塞事件循环

这也是“把函数改成 async def 就会更快”这一误解的来源。性能取决于等待点、阻塞调用、序列化、数据库连接池、线程池和外部服务延迟，而不是函数定义是否包含 async。

如果已有成熟 WSGI 应用，可以通过 WSGI 兼容包装运行在 ASGI 服务器中，但这并不会把同步业务自动变成异步业务。同步代码仍然需要线程池隔离，WebSocket 等 ASGI 特性也不会自动赋予原有 WSGI 应用。

---

11. 真实边界与诊断路径

路由返回 404

先检查：

1. URL 是否包含尾部斜杠；
2. HTTP 方法是否正确；
3. 路由模块是否被导入；
4. 路由是否挂载到主应用；
5. 是否有代理修改了路径前缀。

如果路径已经匹配但返回参数错误，则问题不在路由注册，而在参数类型或约束。

返回 422

检查响应中的 detail：

loc = ["path", "item_id"]

表示路径参数错误；

loc = ["query", "limit"]

表示查询参数错误；

loc = ["body", "price"]

表示请求体字段错误。

这比在路径操作函数中添加打印更有效，因为校验失败时函数根本不会被调用。

异步接口吞吐下降

重点查找：

async def endpoint():
    time.sleep(2)                 # 阻塞事件循环
    requests.get("...")           # 同步 HTTP
    sync_database.query(...)      # 同步数据库
    large_cpu_computation()       # CPU 密集

可采用异步库、普通 def 路径操作、asyncio.to_thread、进程池或外部任务系统。不要只增加 worker 数量掩盖阻塞问题；增加 worker 会复制内存资源，并可能放大数据库连接数。

关闭时资源泄漏

检查：

• 连接池是否在 lifespan 的 finally 中关闭；
• 后台任务是否有明确的停止信号；
• 是否在多个进程中重复创建了不必要的大对象；
• 测试是否使用 with TestClient(app)；
• 是否把请求级资源错误地存成全局共享对象。

生命周期代码最重要的属性不是“启动时执行过”，而是“启动失败能阻止错误实例接收流量，关闭时能释放所有已创建资源”。ASGI Lifespan 协议规定，应用可以发送 startup 或 shutdown failed 事件，让服务器记录失败并退出，而不是继续运行一个未完成初始化的实例。(asgi.readthedocs.io)

---

12. 核心心智模型

FastAPI 的核心不是装饰器本身，而是几条连续的数据转换边界：

HTTP 请求
  -> ASGI scope/events
  -> 路由匹配
  -> 参数来源识别
  -> Pydantic 输入校验
  -> 依赖图解析
  -> 路径操作执行
  -> 响应模型校验与过滤
  -> JSON/HTTP 响应

可以用四个问题检查一个接口设计：

1. 路由：什么 HTTP 方法和路径会进入这个函数？
2. 校验：哪些错误在进入业务逻辑前就应被拒绝？
3. 依赖：这个函数需要哪些可复用的前置条件或资源？
4. 生命周期：资源应该属于一次请求、一个事件循环，还是整个外部系统？

当这四个范围被区分清楚时，FastAPI 的行为就不再是“框架自动完成的魔法”：

• 类型注解决定参数和输出契约；
• Pydantic 负责结构与字段边界；
• 依赖注入负责前置条件和资源组合；
• async/await 负责可等待 I/O 的协作式并发；
• ASGI 负责服务器与应用之间的异步协议；
• 生命周期负责应用级资源的启动、共享和释放。

这些机制共同构成了一个可验证的请求处理系统，而不是几个独立的 API 用法。

---

系列导航与关联阅读

• 系列入口：Python 完整学习路线：从语言模型、并发到 Web、数据、AI 与生产交付
• 上一篇：Python 应用架构：模块边界、依赖方向、领域层和可替换适配器
• 下一篇：Pydantic v2：模型、校验器、序列化、Settings 和性能边界
• 延伸：Python WSGI 与 ASGI：调用协议、并发模型、中间件和部署选择

官方资料

• FastAPI Documentation
• ASGI Specification

本文依据 Python 官方文档、相关 PEP 与生态项目官方文档重新梳理；正文、示例与工程清单由 WR BLOG 编写。