进阶 2 · FastAPI 核心与异步¶

FastAPI 是现代 Python Web 框架的事实标准。对 Java 工程师：它是类型驱动的 Spring Boot——你的类型注解不仅给 IDE/mypy 看，还直接生成请求校验、响应序列化和 OpenAPI 文档。本章讲核心（路由、请求响应、依赖注入、自动文档）和异步实战。

贯穿项目 bookmarks-api 的 main.py 是本章代码的完整范例。

2.1 FastAPI 是什么¶

	Spring Boot	FastAPI
范式	注解 + 反射（运行时）	类型注解驱动（IDE + 运行时）
异步	Servlet 阻塞 / WebFlux 响应式（二选一）	原生 async，async/def 可混用
请求校验	Bean Validation	pydantic（第 1 章）
文档	Springdoc/OpenAPI（需配置）	内置 `/docs` `/redoc`，零配置
服务器	Tomcat（Servlet）/ Netty	ASGI（uvicorn）

安装：

uv add fastapi "uvicorn[standard]"

2.2 第一个应用¶

from fastapi import FastAPI

app = FastAPI(title="My API")

@app.get("/")
async def root():
    return {"message": "hello"}

运行（ASGI 服务器）：

uvicorn main:app --reload      # main.py 里的 app 对象；--reload 开发热重载

访问 http://127.0.0.1:8000/ 得 JSON；自动文档在 http://127.0.0.1:8000/docs（Swagger UI）和 /redoc。

贯穿项目：uvicorn bookmarks_api.main:app --reload。

2.3 路径参数与查询参数¶

Spring（@PathVariable/@RequestParam）

@GetMapping("/users/{id}")
User get(@PathVariable Long id,
         @RequestParam(defaultValue="1") int page) { }

FastAPI（路径/查询自动识别）

@app.get("/users/{id}")
async def get_user(id: int, page: int = 1):
    return {"id": id, "page": page}

FastAPI 按位置和有无默认值自动区分：

在路径字符串 {...} 里的 → 路径参数。
有默认值的普通参数 → 查询参数（?page=1）。
类型注解（id: int）自动校验+转换，不匹配返回 422。

@app.get("/items")
async def list_items(q: str | None = None, limit: int = 10):
    # GET /items?q=book&limit=5
    return {"q": q, "limit": limit}

2.4 请求体（pydantic 模型）¶

把第 1 章的 pydantic 模型直接用作参数，FastAPI 自动解析 JSON + 校验：

from bookmarks_api.schemas import BookmarkCreate

@app.post("/bookmarks")
async def create_bookmark(payload: BookmarkCreate):
    # payload 已是校验过的 BookmarkCreate 实例
    return payload

Spring（@RequestBody + @Valid）

@PostMapping("/bookmarks")
Bookmark create(@Valid @RequestBody BookmarkDto dto) { }

FastAPI

@app.post("/bookmarks")
async def create(payload: BookmarkCreate): ...

校验失败 FastAPI 自动返回 422 + 结构化错误（直接来自 pydantic 的 ValidationError），无需手写。

2.5 响应模型 `response_model`¶

声明 response_model 过滤输出、生成文档的响应 schema：

from bookmarks_api.schemas import BookmarkOut

@app.get("/bookmarks", response_model=list[BookmarkOut])
async def list_bookmarks(...): ...

⚠️ Java 程序员的陷阱

不要把内部 ORM 对象或含敏感字段的 dict 直接返回。用 response_model 声明对外契约，FastAPI 自动只输出该模型的字段——这是"输入契约（Create）≠ 输出契约（Out）"的关键，避免泄露 api_key、password_hash 等。

贯穿项目里 UserCreate（注册时 username）和 UserOut（返回含 api_key）是分开的两个模型。

2.6 自动文档（零配置）¶

FastAPI 基于路由签名 + pydantic 模型，自动生成 OpenAPI 规范：

/docs — Swagger UI（可交互测试 API）
/redoc — ReDoc（更适合阅读的文档）
/openapi.json — 原始 OpenAPI 规范

Java 里这要靠 Springdoc 等插件 + 大量注解配置；FastAPI 声明即文档。这也是它对 Java 工程师最直观的"爽点"。

2.7 异步：何时 `async def`，何时 `def`¶

这是 FastAPI 的核心，也是 Java 工程师最需要理解的。FastAPI 路由有两种写法，行为不同：

@app.get("/a")
async def a():                 # async 路由：在事件循环里直接 await
    data = await fetch()       # 必须用 async 库（httpx.AsyncClient、async DB）
    return data

@app.get("/b")
def b():                       # 同步路由：FastAPI 自动放到线程池，不阻塞事件循环
    data = requests.get(...)   # 同步阻塞调用安全（在线程池里）
    return data

规则：

路由写法	调用	说明
`async def`	必须 `await` 异步库	在事件循环内，不能直接调用同步阻塞函数（会卡住整个循环）
普通 `def`	用同步库即可	FastAPI 自动丢进线程池，不阻塞

致命陷阱

在 async def 路由里调用同步阻塞函数（如 requests.get、同步 DB 驱动、time.sleep），会阻塞整个事件循环——所有并发请求都卡住。要么改用 async 库，要么把阻塞调用丢进线程池：

from starlette.concurrency import run_in_threadpool
result = await run_in_threadpool(blocking_function, arg)

决策（呼应核心教程第 11 章）：

有 async 库（async DB、httpx async）→ 用 async def + await，享受高并发。
只能拿到同步库 → 用普通 def，让 FastAPI 的线程池兜底（别在 async 路由里裸调同步阻塞）。

贯穿项目用 async DB（SQLAlchemy async + aiosqlite），所以路由都是 async def。

2.8 异常处理¶

HTTPException 抛出即返回对应状态码（对照 Spring 的 ResponseEntity.status(...)）：

from fastapi import HTTPException, status

@app.get("/bookmarks/{id}")
async def get_bookmark(id: int, ...):
    bm = find(id)
    if bm is None:
        raise HTTPException(status.HTTP_404_NOT_FOUND, "bookmark not found")
    return bm

status.HTTP_404_NOT_FOUND 就是数字 404，用常量更可读（对照 HttpServletResponse.SC_NOT_FOUND）。

全局异常处理（对照 Spring @ControllerAdvice）：

from fastapi import Request
from fastapi.responses import JSONResponse

@app.exception_handler(ValueError)
async def value_error_handler(request: Request, exc: ValueError):
    return JSONResponse(status_code=400, content={"detail": str(exc)})

2.9 中间件与 CORS¶

中间件包裹每个请求（对照 Spring 的 Filter / interceptor）：

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=settings.cors_origins,
    allow_methods=["*"],
    allow_headers=["*"],
)

CORS 几乎是前后端分离 API 的标配（前端跨域调用）。贯穿项目 main.py 已配置。

2.10 依赖注入 `Depends`（FastAPI 的灵魂）¶

FastAPI 的 Depends 让你把"获取某个值/对象"的逻辑声明为依赖，自动注入——这是它区别于其他框架的核心。第 3 章会用它做"DB session"和"认证"，这里先建立概念：

from fastapi import Depends

def common_params(q: str | None = None, page: int = 1):
    return {"q": q, "page": page}

@app.get("/items")
async def list_items(params: dict = Depends(common_params)):
    return params          # params 自动是 common_params() 的返回值

为什么强大：

复用：多个路由共享同一段逻辑（分页、认证、DB session）。
可组合：依赖可以再依赖别的（依赖链）。
可测：测试时用 app.dependency_overrides 替换（贯穿项目测试正是这么换 DB 的）。
生命周期：用 yield 的依赖可在请求后做清理（关闭 session）。

async def get_db():
    async with async_session() as session:
        yield session       # 请求结束自动回到这里关闭

@app.get("/items")
async def list_items(db = Depends(get_db)):   # 自动注入 session
    ...

下一章会把 Depends 用在数据库和API Key 认证上。

本章练习¶

练习 2.1

写一个 GET /temperature/{city} 路由：city 是路径参数（str），unit 是查询参数（默认 "celsius"，可选 "fahrenheit"）。返回 JSON。

参考答案

from fastapi import FastAPI
app = FastAPI()
@app.get("/temperature/{city}")
async def temp(city: str, unit: str = "celsius"):
    return {"city": city, "unit": unit, "value": 25}

练习 2.2

解释为什么下面这段 async def 路由是危险的，并给出两种修复。

@app.get("/slow")
async def slow():
    import time; time.sleep(2)
    return {"ok": True}

参考答案

time.sleep 是同步阻塞，在 async def 里会卡住整个事件循环，期间所有其他请求被阻塞。修复一：改成 asyncio.sleep(2)（异步等待，让出控制权）。修复二：保持阻塞但用普通 def（FastAPI 自动放线程池），或 await run_in_threadpool(time.sleep, 2)。

练习 2.3

用 Depends 写一个分页依赖 pagination(skip: int = 0, limit: int = 10)，在两个路由里复用。

参考答案

def pagination(skip: int = 0, limit: int = 10):
    return {"skip": skip, "limit": limit}
@app.get("/a")
async def a(p=Depends(pagination)): return p
@app.get("/b")
async def b(p=Depends(pagination)): return p

练习 2.4

说明 response_model 如何防止敏感字段泄露，并举一个 Create 模型与 Out 模型不同的例子。

参考答案

response_model 声明对外字段，FastAPI 只序列化该模型的字段。例：UserCreate 含 password，UserOut 只含 id/username——路由 response_model=UserOut 时，即使返回的对象含 password 也不会输出。

上一章：进阶 1 · pydantic ｜ ← 回首页｜下一章：进阶 3 · 持久化与认证

进阶 2 · FastAPI 核心与异步¶

2.1 FastAPI 是什么¶

2.2 第一个应用¶

2.3 路径参数与查询参数¶

2.4 请求体（pydantic 模型）¶

2.5 响应模型 response_model¶

2.6 自动文档（零配置）¶

2.7 异步：何时 async def，何时 def¶

2.8 异常处理¶

2.9 中间件与 CORS¶

2.10 依赖注入 Depends（FastAPI 的灵魂）¶

本章练习¶

2.5 响应模型 `response_model`¶

2.7 异步：何时 `async def`，何时 `def`¶

2.10 依赖注入 `Depends`（FastAPI 的灵魂）¶