咖啡工坊 · FastAPI 22 章教程合集

☕ 咖啡工坊 · FastAPI 22 章教程合集

本文件由 22 篇原始文档按章节顺序合并而成，原始文档已保留在 docs/ 目录下，可随时查阅。

目录

• 第 01 章 什么是 FastAPI
• 第 02 章 FastAPI 核心特性
• 第 03 章 FastAPI 首个项目运行
• 第 04 章 基础路由与请求方式
• 第 05 章 三大请求参数详解
• 第 06 章 参数校验
• 第 07 章 响应数据处理
• 第 08 章 静态文件与模板渲染
• 第 09 章 文件上传与下载
• 第 10 章 异常处理与全局异常捕获
• 第 11 章 中间件与跨域
• 第 12 章 依赖注入
• 第 13 章 基础认证方式
• 第 14 章 JWT 令牌认证
• 第 15 章 规范项目目录结构
• 第 16 章 数据库的联动开发
• 第 17 章 接口版本管理
• 第 18 章 同步和异步接口
• 第 19 章 异步数据库和异步请求
• 第 20 章 高并发注意要点
• 第 21 章 日志 / 测试 / 接口文档
• 第 22 章 项目部署与上线

第 01 章 什么是 FastAPI

本章为概念章，不涉及可运行代码。读完后请直接进入第 02 章动手实践。

一、FastAPI 是什么

FastAPI 是一个用于构建 API（应用程序编程接口）的现代 Python Web 框架。它脱胎于 Starlette（处理 ASGI 网络层）与 Pydantic（数据校验层），站在两个优秀项目的肩膀上，因此天生具备：

二、FastAPI 与其它框架的关系

• Starlette：负责 HTTP 解析、路由、中间件、依赖注入，是 FastAPI 的"骨架"。
• Pydantic：负责数据校验、序列化、模型定义，是 FastAPI 的"血肉"。
• Uvicorn：基于 uvloop + httptools 的 ASGI 服务器，真正把代码跑起来。

也就是说，FastAPI 自身代码量不大，但它把社区最强的两个库粘合在一起，所以你写出来的应用又快又稳。

三、为什么选择 FastAPI

四、本教程的定位

读完本章你应该带着一个疑问：“它到底快在哪里？开发体验到底如何？”，带着这个疑问进入第 02 章。

第 02 章 FastAPI 核心特性

本章为特性预览章，所有示例均可在第 03 章之后动手验证。

一、五大核心特性一览

下面我们逐条拆解。

二、特性 ①：类型即文档

from datetime import date

def create_order(customer: str, items: list[str], pickup_date: date) –> dict:
...

FastAPI 会把这段类型提示直接翻译成 OpenAPI 文档片段：

{
"customer": {"type": "string"},
"items": {"type": "array", "items": {"type": "string"}},
"pickup_date": {"type": "string", "format": "date"}
}

IDE 在你写代码时就已经知道 customer 必须是 str，items 是 list[str]，连错误都会提前提醒。

三、特性 ②：Pydantic 自动校验

当你用 Pydantic 模型声明请求体：

from pydantic import BaseModel, Field

class OrderIn(BaseModel):
customer: str = Field(min_length=1, max_length=50)
cups: int = Field(gt=0, le=20)

客户端若发送 cups: 0，FastAPI 会自动返回 422 状态码 + 错误详情，你不需要写一行校验代码。

四、特性 ③：自动文档

启动应用后浏览器访问：

• http://127.0.0.1:8803/docs → Swagger UI（可调试）
• http://127.0.0.1:8803/redoc → ReDoc（只读，更适合交付）

文档与代码始终同步，因为文档就是从代码"长出来的"。

五、特性 ④：async 优先

FastAPI 同时支持同步 (def) 与异步 (async def) 处理函数：

@app.get("/sync")
def sync_route(): # 阻塞型
return {"kind": "sync"}

@app.get("/async")
async def async_route(): # 协程型
return {"kind": "async"}

• def 适合调用阻塞型库（如部分老版数据库驱动）。
• async def 适合 I/O 密集场景（HTTP、数据库、WebSocket），能释放事件循环。

第 18 章会专门对比二者性能差异。

六、特性 ⑤：依赖注入

from typing import Annotated
from fastapi import Depends

def get_menu_service():
return MenuService()

@app.get("/menu")
def list_menu(svc: Annotated[MenuService, Depends(get_menu_service)]):
return svc.all()

• 依赖可以是普通函数、类、生成器。
• 可嵌套：依赖里再依赖另一个依赖。
• 同一个请求内依赖只执行一次（缓存）。

七、把五大特性串成一条主线

类型提示 → Pydantic 校验 → OpenAPI 文档 → 异步执行 → 依赖复用
① ② ③ ④ ⑤

后续章节会按 ① → ⑤ 的顺序逐步展开，每一个特性都会成为下一章的"自然延伸"。第 03 章我们就从"启动第一个项目"开始。

第 03 章 FastAPI 首个项目运行

本章是动手的第一站，我们会启动一个最小化的 FastAPI 应用，并理解每个文件、每行代码的作用。 端口约定：本章使用 8803。

一、运行前准备

确保你已经在项目根目录执行过：

uv sync # 同步依赖（生成 .venv）
uv add fastapi uvicorn # 已包含在 pyproject.toml，无需重复

本项目使用 uv 管理依赖与虚拟环境，无需手动 python -m venv。

二、最小化可运行代码

完整源码见 chapters/ch03_first_run/main.py。下面逐行拆解。

2.1 导入与创建实例

from fastapi import FastAPI # 导入 FastAPI 类
app = FastAPI() # 创建一个 ASGI 应用实例

• FastAPI() 实例就是整个 Web 应用的"根"。
• 同一个进程内可以创建多个实例（多应用隔离时很有用），但本教程统一一个实例。

2.2 注册第一个路由

@app.get("/") # 把下面的函数注册到 GET /
def root() –> dict[str, str]: # 函数返回 dict，FastAPI 会自动转 JSON
return {"message": "欢迎来到咖啡工坊"}

• @app.get 是装饰器：把函数 root 与 URL / 的 HTTP GET 方法绑定。
• 返回 dict，FastAPI 默认使用 application/json 返回。

2.3 启动命令

我们约定用 uvicorn 启动：

uv run uvicorn chapters.ch03_first_run.main:app –reload –port 8803

参数说明：

启动成功后浏览器访问 http://127.0.0.1:8803/ 即可看到返回的 JSON。

三、章节代码逐行注释

请打开 chapters/ch03_first_run/main.py，每一行都附有中文注释。重点关注：

四、自动生成的文档

启动后访问以下地址：

• http://127.0.0.1:8803/docs —— Swagger UI，可直接调试接口。
• http://127.0.0.1:8803/redoc —— ReDoc，只读文档风格。

你会看到刚刚定义的 GET / 接口，且元信息正是我们传入的 title / version。

五、本章小结

下一章我们会把 / 扩展成"咖啡菜单"相关接口，自然延伸到 基础路由与请求方式。

第 04 章 基础路由与请求方式

在第 03 章我们跑起来一个 GET /。本章把它扩展为一个完整的 咖啡订单 REST 接口， 演示所有常用 HTTP 方式。端口约定：8804。

一、HTTP 方法与业务动作的对应

REST 设计中，每个方法都有明确语义：

二、章节代码位置

chapters/ch04_routes/main.py 中我们实现：

三、关键代码拆解

3.1 GET 列表

@app.get("/menu", tags=["菜单"])
def list_menu() –> list[dict]:
return MENU

3.2 GET 详情 + 路径参数

@app.get("/menu/{item_id}")
def get_menu_item(item_id: int) –> dict:
for item in MENU:
if item["id"] == item_id:
return item
raise HTTPException(status_code=404, detail="找不到该咖啡")

{item_id} 会被自动捕获，并通过类型提示 int 自动转换为整数。 转换失败（如 /menu/abc）会得到 422 错误。

3.3 POST 创建订单

@app.post("/orders", status_code=201)
def create_order(payload: dict) –> dict:
...

• status_code=201 让成功创建的资源返回 201 Created，符合 REST 习惯。

3.4 PUT / PATCH / DELETE

• PUT：客户端必须传完整字段，服务端全量覆盖。
• PATCH：只更新客户端发送的字段。
• DELETE：删除资源并返回 204。

四、运行验证

uv run uvicorn chapters.ch04_routes.main:app –reload –port 8804

依次执行：

curl -X POST http://127.0.0.1:8804/orders \\
-H "Content-Type: application/json" \\
-d "{\\"customer\\":\\"小明\\",\\"item_id\\":1,\\"cups\\":2}"

curl http://127.0.0.1:8804/orders
curl -X DELETE http://127.0.0.1:8804/orders/1

观察不同 HTTP 方法返回的状态码与响应体。

五、本章要点

下一章我们会深入 路径参数、查询参数、请求体 三大参数的细节，并补齐类型提示。

第 05 章 三大请求参数详解

第 04 章我们用 dict 接住了请求体。本章升级为 类型化参数， 并彻底讲清楚 Path / Query / Body 三大参数的来源、用途与细节。 端口约定：8805。

一、参数地图

HTTP 请求
│
┌────────┼─────────┐
│ │
│ │
路径 (Path) 查询 (Query) 请求体 (Body)
/orders/{oid} ?status=paid JSON / 表单
出现在 URL 路径里 出现在 ? 之后 出现在 Body 中

二、本章业务：咖啡订单查询

GET /orders?status=pending&page=1&size=10

• status、page、size 都是 Query 参数。

新增 POST /orders 时传入 JSON Body：

{"customer":"小明","item_id":1,"cups":2,"note":"少糖"}

这是 Body 参数，用 Pydantic 模型声明。

三、Path 参数高级写法

from typing import Annotated
from fastapi import Path

@app.get("/orders/{oid}")
def get_order(oid: Annotated[int, Path(ge=1, description="订单ID，必须 ≥ 1")]):
...

• Annotated[T, Path(…)] 是 3.9+ 推荐写法，把元信息与类型分开。
• ge=1 表示大于等于 1，越界自动返回 422。

四、Query 参数细节

from fastapi import Query

@app.get("/orders")
def list_orders(
status: Annotated[str | None, Query(description="过滤状态")] = None,
page: Annotated[int, Query(ge=1, le=999)] = 1,
size: Annotated[int, Query(ge=1, le=100)] = 20,
):
...

要点：

• 默认值让参数变可选。
• Annotated[…, Query(…)] 同时提供类型与校验。

五、Body 参数与 Pydantic

from pydantic import BaseModel, Field

class OrderIn(BaseModel):
customer: str = Field(min_length=1, max_length=50)
item_id: int = Field(ge=1)
cups: int = Field(ge=1, le=20)
note: str | None = None

OrderIn 会在请求到达时自动校验，并直接给出强类型对象。

六、本章代码运行

uv run uvicorn chapters.ch05_params.main:app –reload –port 8805

测试用例：

# 路径参数 + 查询参数
curl "http://127.0.0.1:8805/orders/1?note=hello"

# 查询参数 + 分页
curl "http://127.0.0.1:8805/orders?status=pending&page=1&size=5"

# Body 参数
curl -X POST http://127.0.0.1:8805/orders \\
-H "Content-Type: application/json" \\
-d "{\\"customer\\":\\"小明\\",\\"item_id\\":1,\\"cups\\":2}"

七、章节要点

下一章我们会在 Body 上进一步做字段级精细校验（正则、枚举、嵌套模型）。

第 06 章 参数校验

第 05 章我们已经让 Pydantic 做了基础校验。本章把校验做到 字段级： 正则、枚举、嵌套、列表元素、Email、手机号、URL 格式……全部交给 Pydantic。 端口约定：8806。

一、为什么需要细粒度校验

二、Field 常用约束

from pydantic import Field
from decimal import Decimal

price: Decimal = Field(ge=0, le=999, decimal_places=2)

• ge / le / gt / lt：数值边界。
• min_length / max_length：字符串 / 容器长度。
• pattern：正则校验字符串。
• description：自动写入 OpenAPI。

三、字符串格式校验

Pydantic v2 内置了 EmailStr / HttpUrl / IPvAnyAddress 等。

from pydantic import EmailStr, HttpUrl

customer_email: EmailStr
shop_website: HttpUrl

这些类型在校验失败时会返回非常清晰的 422 错误。

四、枚举与字面量

from enum import Enum

class Roast(str, Enum):
LIGHT = "浅烘"
MEDIUM = "中烘"
DARK = "深烘"

roast: Roast

或者使用 Python 3.12+ 的 Literal：

from typing import Literal
roast: Literal["浅烘", "中烘", "深烘"]

五、嵌套模型

class OrderItem(BaseModel):
item_id: int
cups: int = Field(ge=1)

class OrderIn(BaseModel):
customer: str
items: list[OrderItem] = Field(min_length=1)

items 数组中每个元素都会按 OrderItem 校验。

六、自定义校验器

from pydantic import field_validator

class CustomerIn(BaseModel):
name: str

@field_validator("name")
@classmethod
def name_must_not_be_blank(cls, v: str) –> str:
if not v.strip():
raise ValueError("姓名不能为空")
return v.strip()

七、本章代码要点

chapters/ch06_validation/main.py 中我们落地了：

八、运行验证

uv run uvicorn chapters.ch06_validation.main:app –reload –port 8806

依次尝试发送非法数据，观察 422 响应：

curl -X POST http://127.0.0.1:8806/customers \\
-H "Content-Type: application/json" \\
-d "{\\"name\\":\\" \\",\\"email\\":\\"not-an-email\\"}"

九、本章要点

下一章我们把校验好的数据包装成响应：status_code、headers、response_model、Cookie。

第 07 章 响应数据处理

之前我们直接 return dict。本章介绍 FastAPI 的 响应模型、自定义状态码、 响应头 / Cookie、多种 Response 类型，并把所有要点串到一份咖啡订单接口里。 端口约定：8807。

一、为什么要控制响应

二、response_model

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

class BeanOut(BaseModel):
id: int
name: str
price: float

@app.post("/beans", response_model=BeanOut, status_code=201)
def add_bean(payload: BeanIn): ...

即便函数里返回了多余字段，response_model 也会只保留模型中的字段， 起到"白名单"作用。

进阶：过滤选项

response_model=BeanOut,
response_model_exclude_none=True, # 字段为 None 时不返回
response_model_exclude_unset=True, # 客户端没传就不返回

三、自定义状态码

from fastapi import status

@app.post("/orders", status_code=status.HTTP_201_CREATED)
def create_order(...): ...

status 模块集中了所有 HTTP 状态码常量，便于语义化。

四、Response 家族

from fastapi.responses import PlainTextResponse, RedirectResponse

@app.get("/health", response_class=PlainTextResponse)
def health():
return "ok"

@app.get("/old-path")
def old():
return RedirectResponse(url="/new-path", status_code=307)

五、设置 Headers / Cookies

from fastapi import Response

@app.get("/set-cookie")
def set_cookie(response: Response):
response.set_cookie(key="user", value="alice", max_age=3600)
response.headers["X-Custom"] = "demo"
return {"ok": True}

六、章节代码要点

chapters/ch07_response/main.py 演示：

七、运行与验证

uv run uvicorn chapters.ch07_response.main:app –reload –port 8807

curl -i http://127.0.0.1:8807/health
curl -i http://127.0.0.1:8807/redirect
curl -i http://127.0.0.1:8807/set

观察响应头里的 content-type、location、set-cookie。

八、本章要点

下一章进入"看得见"的部分：静态文件 + 模板渲染。

第 08 章 静态文件与模板渲染

之前所有响应都是 JSON。本章给咖啡工坊加上"门面"：HTML 首页 和 CSS / 图片。 端口约定：8808。

一、为什么需要模板

FastAPI 内置 StaticFiles 处理静态资源，Jinja2Templates 处理 HTML 模板。

二、目录约定

chapters/ch08_static_templates/
├── main.py
├── static/
│ ├── css/
│ │ └── style.css
│ └── img/
│ └── logo.svg
└── templates/
├── base.html
├── index.html
└── menu.html

三、挂载静态文件

from fastapi.staticfiles import StaticFiles

app.mount("/static", StaticFiles(directory="static"), name="static")

之后 http://127.0.0.1:8808/static/css/style.css 即可访问。

四、Jinja2 模板

from fastapi.templating import Jinja2Templates

templates = Jinja2Templates(directory="templates")

@app.get("/", response_class=HTMLResponse)
def home(request: Request):
return templates.TemplateResponse(
"index.html",
{"request": request, "shop": "咖啡工坊"},
)

注意：

• 上下文中必须包含 request（Starlette 模板约定）。
• templates.TemplateResponse 返回 TemplateResponse 对象，可作为 Response。

五、模板语法

templates/menu.html：

{% extends "base.html" %}
{% block content %}
<h1>{{ shop }} · 菜单</h1>
<ul>
{% for item in menu %}
<li>{{ item.name }} — ¥{{ item.price }}</li>
{% endfor %}
</ul>
{% endblock %}

六、章节代码要点

chapters/ch08_static_templates/main.py：

七、运行验证

uv run uvicorn chapters.ch08_static_templates.main:app –reload –port 8808

浏览器访问：

• http://127.0.0.1:8808/ — 首页
• http://127.0.0.1:8808/menu — 菜单页
• http://127.0.0.1:8808/static/css/style.css — 静态资源

八、本章要点

下一章进入"看得见的数据交换"：文件上传与下载。

第 09 章 文件上传与下载

上一章用 Jinja2 渲染页面，本章让用户把菜谱 PDF / 咖啡豆图片上传到服务器， 也可以把成品资料下载下来。端口约定：8809。

一、两类操作

二、上传表单（multipart/form-data）

from fastapi import UploadFile, File

@app.post("/upload")
async def upload_image(file: UploadFile = File(...)):
content = await file.read()
...

• UploadFile 包装了 SpooledTemporaryFile，大文件不会全部读入内存。
• await file.read() / await file.write(…) 是异步方法。
• file.filename / file.content_type 给出元信息。

三、保存到磁盘

import pathlib
UPLOAD_DIR = pathlib.Path("uploads")
UPLOAD_DIR.mkdir(exist_ok=True)

async def save(file: UploadFile) –> str:
dest = UPLOAD_DIR / file.filename
data = await file.read()
dest.write_bytes(data)
return str(dest)

四、限制上传大小 / 类型

@app.post("/upload")
async def upload(file: UploadFile = File(..., max_length=2*1024*1024)): # 2MB
if file.content_type not in {"image/png", "image/jpeg"}:
raise HTTPException(400, "仅支持 PNG / JPEG")
...

max_length 在新版 Starlette 中可能名称不同，常见做法是在中间件或自己读取后判断。

五、下载：FileResponse

from fastapi.responses import FileResponse

@app.get("/download/{name}")
def download(name: str):
path = UPLOAD_DIR / name
if not path.exists():
raise HTTPException(404, "文件不存在")
return FileResponse(path, filename=name, media_type="application/octet-stream")

六、下载：StreamingResponse

from fastapi.responses import StreamingResponse

@app.get("/stream")
def stream():
def gen():
for i in range(5):
yield f"chunk {i}\\n".encode()
return StreamingResponse(gen(), media_type="text/plain")

适合 大文件 或 实时数据流。

七、章节代码要点

chapters/ch09_files/main.py：

八、运行验证

uv run uvicorn chapters.ch09_files.main:app –reload –port 8809

# 单文件上传
curl -F "file=@./recipe.pdf" http://127.0.0.1:8809/upload

# 多文件上传
curl -F "files=@./a.jpg" -F "files=@./b.jpg" http://127.0.0.1:8809/upload-multiple

九、本章要点

下一章进入"出错怎么办"：异常处理与全局异常捕获。

第 10 章 异常处理与全局异常捕获

业务逻辑出错时直接 raise HTTPException 是基础做法。本章我们构建一份 业务异常类 + 全局异常处理器，让错误响应统一、可扩展。 端口约定：8810。

一、FastAPI 默认行为

raise HTTPException(status_code=404, detail="订单不存在")

响应：

{"detail": "订单不存在"}

简单场景够用，但当业务复杂时（不同异常 → 不同 JSON 结构、不同状态码）就要：

• 自定义异常类；
• 注册 @app.exception_handler；
• 覆盖默认 HTTPException / RequestValidationError。

二、自定义业务异常

class BusinessError(Exception):
def __init__(self, code: str, message: str, status_code: int = 400):
self.code = code
self.message = message
self.status_code = status_code

业务代码：

order = ORDERS.get(oid)
if order is None:
raise BusinessError("ORDER_NOT_FOUND", f"订单 {oid} 不存在", 404)

三、全局异常处理器

from fastapi.responses import JSONResponse

@app.exception_handler(BusinessError)
async def handle_business(request: Request, exc: BusinessError):
return JSONResponse(
status_code=exc.status_code,
content={"code": exc.code, "message": exc.message},
)

四、统一请求校验错误格式

from fastapi.exceptions import RequestValidationError

@app.exception_handler(RequestValidationError)
async def handle_validation(request, exc):
return JSONResponse(
status_code=422,
content={
"code": "VALIDATION_ERROR",
"errors": exc.errors(),
},
)

五、捕获未处理异常（兜底）

@app.exception_handler(Exception)
async def handle_all(request, exc):
return JSONResponse(
status_code=500,
content={"code": "INTERNAL", "message": "服务器开了个小差，请稍后再试"},
)

不要在生产环境把 str(exc) 直接暴露给客户端，可能泄露内部细节。

六、章节代码要点

chapters/ch10_exception/main.py：

七、运行验证

uv run uvicorn chapters.ch10_exception.main:app –reload –port 8810

# 触发业务错误
curl -i http://127.0.0.1:8810/orders/9999

# 触发参数校验错误
curl -X POST http://127.0.0.1:8810/orders \\
-H "Content-Type: application/json" \\
-d '{"customer":"","cups":0}'

# 触发兜底
curl -i http://127.0.0.1:8810/boom

八、本章要点

下一章进入"中间件层"：中间件与跨域。

第 11 章 中间件与跨域

中间件是 FastAPI 处理请求的"流水线"。本章加入 请求耗时日志 和 CORS 跨域。 端口约定：8811。

一、中间件是什么

请求 ──▶ M1 ──▶ M2 ──▶ 路由处理函数 ──▶ M2 ──▶ M1 ──▶ 响应

每个中间件可以：

• 在调用下游前修改 request。
• 在下游返回后修改 response。
• 短路（直接返回，不调用下游）。

二、两种定义方式

2.1 装饰器（推荐）

@app.middleware("http")
async def timing_middleware(request: Request, call_next):
start = time.perf_counter()
response = await call_next(request)
cost = (time.perf_counter() – start) * 1000
response.headers["X-Process-Time-ms"] = f"{cost:.2f}"
return response

2.2 类形式

from starlette.middleware.base import BaseHTTPMiddleware

class TimingMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request, call_next):
start = time.perf_counter()
response = await call_next(request)
response.headers["X-Process-Time-ms"] = f"{(time.perf_counter()–start)*1000:.2f}"
return response

app.add_middleware(TimingMiddleware)

三、CORS（跨域资源共享）

浏览器同源策略会阻止 a.com 调 b.com 的 API。CORS 通过响应头告诉浏览器"我允许谁"。

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
CORSMiddleware,
allow_origins=["https://shop.example.com"], # 或 ["*"] 表示全部
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)

四、自定义中间件举例

4.1 请求 ID

import uuid

@app.middleware("http")
async def add_request_id(request: Request, call_next):
rid = request.headers.get("X-Request-ID", str(uuid.uuid4()))
request.state.request_id = rid
response = await call_next(request)
response.headers["X-Request-ID"] = rid
return response

4.2 简单访问日志

@app.middleware("http")
async def access_log(request: Request, call_next):
print(f"[{time.strftime('%H:%M:%S')}] {request.method} {request.url.path}")
return await call_next(request)

五、中间件顺序

add_middleware 按调用顺序包裹：

请求 → M1(最先 add) → M2 → 路由 → M2 → M1 → 响应

六、章节代码要点

chapters/ch11_middleware/main.py：

七、运行验证

uv run uvicorn chapters.ch11_middleware.main:app –reload –port 8811

curl -i http://127.0.0.1:8811/menu # 看响应头 X-Process-Time-ms
curl -i -H "Origin: https://x.com" -H "Access-Control-Request-Method: GET" \\
-X OPTIONS http://127.0.0.1:8811/menu

八、本章要点

下一章进入"依赖注入"：模块化复用。

第 12 章 依赖注入

上一章我们写了一份"内存订单服务"。本章把"获取服务实例"做成 可复用依赖， 并展示嵌套依赖、缓存、按参数 yield 资源等高级用法。端口约定：8812。

一、为什么用 Depends

如果每个接口都写 svc = OrderService(); svc.connect()，会有大量重复。 依赖注入把这些"准备资源"的代码集中起来：

def get_order_service() –> OrderService:
return OrderService()

@app.get("/orders/{oid}")
def get_order(oid: int, svc: Annotated[OrderService, Depends(get_order_service)]):
return svc.get(oid)

二、Annotated 写法（推荐）

from typing import Annotated
from fastapi import Depends

svc: Annotated[OrderService, Depends(get_order_service)]

Python 3.9+ 风格，把类型与"依赖来源"分离。

三、类作为依赖

class Pagination:
def __init__(
self,
page: Annotated[int, Query(ge=1)] = 1,
size: Annotated[int, Query(ge=1, le=100)] = 20,
):
self.page = page
self.size = size
self.offset = (page – 1) * size

@app.get("/items")
def list_items(p: Annotated[Pagination, Depends()]):
return {"page": p.page, "size": p.size, "offset": p.offset}

四、嵌套依赖

def get_db():
return DatabaseConn()

def get_user(db: Annotated[DatabaseConn, Depends(get_db)]):
return db.get_current_user()

@app.get("/profile")
def profile(user: Annotated[User, Depends(get_user)]):
return user.dict()

五、yield 依赖（资源自动关闭）

def get_db_session():
db = SessionLocal()
try:
yield db
finally:
db.close()

@app.post("/items")
def create(db: Annotated[Session, Depends(get_db_session)], payload: ItemIn):
db.add(Item(**payload.model_dump()))
db.commit()

yield 之前 = “前置”，之后 = “后置清理”。FastAPI 会保证 finally 一定执行。

六、章节代码要点

chapters/ch12_dependency/main.py：

七、运行验证

uv run uvicorn chapters.ch12_dependency.main:app –reload –port 8812

curl "http://127.0.0.1:8812/orders?page=2&size=5"
curl "http://127.0.0.1:8812/info"
curl "http://127.0.0.1:8812/now"

八、本章要点

下一章进入"访问控制"：基础认证方式。

第 13 章 基础认证方式

咖啡工坊后台只能让员工访问。本章介绍三种轻量级认证： HTTPBasic / APIKey / 简单 Bearer，不依赖 OAuth/JWT。 端口约定：8813。

一、HTTP Basic

客户端在 Header 发送 Authorization: Basic base64(user:pass)。 服务端解码后校验用户名密码。

from fastapi.security import HTTPBasic, HTTPBasicCredentials
import secrets

security = HTTPBasic()

def check_cred(creds: Annotated[HTTPBasicCredentials, Depends(security)]):
ok_user = secrets.compare_digest(creds.username, "alice")
ok_pass = secrets.compare_digest(creds.password, "secret")
if not (ok_user and ok_pass):
raise HTTPException(401, "认证失败", headers={"WWW-Authenticate": "Basic"})
return creds.username

secrets.compare_digest 防时序攻击。

二、API Key

通过 Header / Query 传一个固定 key。

from fastapi.security import APIKeyHeader

api_key = APIKeyHeader(name="X-API-Key")

def check_key(key: Annotated[str, Depends(api_key)]):
if key != "my-secret-key":
raise HTTPException(403, "Invalid API Key")
return key

客户端：

curl -H "X-API-Key: my-secret-key" http://...

三、简易 Bearer

from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer

bearer = HTTPBearer()

def check_bearer(token: Annotated[HTTPAuthorizationCredentials, Depends(bearer)]):
if token.credentials != "my-token":
raise HTTPException(403, "Invalid token")
return token.credentials

四、401 与 403

• 401 Unauthorized：未提供凭证 / 凭证错误。
• 403 Forbidden：凭证有效但无权访问。

HTTPException(401, …, headers={"WWW-Authenticate": "Basic"}) 是标准做法。

五、章节代码要点

chapters/ch13_auth_basic/main.py：

六、运行验证

uv run uvicorn chapters.ch13_auth_basic.main:app –reload –port 8813

# Basic
curl -u alice:secret http://127.0.0.1:8813/admin/orders

# API Key
curl -H "X-API-Key: my-secret-key" http://127.0.0.1:8813/internal/stats

# Bearer
curl -H "Authorization: Bearer my-token" http://127.0.0.1:8813/api/secret

七、本章要点

下一章进入真正的"令牌认证"：JWT。

第 14 章 JWT 令牌认证

上一章我们用静态 Token 鉴权。本章换成真正的 JWT（JSON Web Token）： 登录后签发令牌，后续接口通过令牌识别用户身份。端口约定：8814。

一、JWT 三段式

xxxxx.yyyyy.zzzzz
↑ ↑ ↑
header payload signature

• header：算法 / 类型。
• payload：业务数据（用户 ID、过期时间等）。
• signature：用密钥对前两段签名，防篡改。

二、关键依赖

from jose import jwt, JWTError

python-jose 是 Python 生态里最常用的 JWT 库。

三、签发令牌

SECRET = "change-me"
ALG = "HS256"

def create_token(sub: str, ttl_seconds: int = 3600) –> str:
now = datetime.now(timezone.utc)
payload = {
"sub": sub,
"iat": now,
"exp": now + timedelta(seconds=ttl_seconds),
}
return jwt.encode(payload, SECRET, algorithm=ALG)

四、校验令牌

def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]) –> str:
try:
payload = jwt.decode(token, SECRET, algorithms=[ALG])
except JWTError:
raise HTTPException(401, "Invalid token")
return payload["sub"]

五、登录流程

客户端 服务端
│ │
│── POST /login {user, pass} ────────▶ │
│ │ 校验 bcrypt
│◀── {access_token: "xxxx.yyyy.zzzz"} ─│
│ │
│── GET /me Authorization: Bearer … ─▶│
│ │ 解析 JWT → sub
│◀── {user: "alice"} ──────────────── │

六、FastAPI 内置 OAuth2 工具

from fastapi.security import OAuth2PasswordBearer

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="login")

tokenUrl="login" 会让 Swagger UI 显示"Authorize"按钮，调试更方便。

七、章节代码要点

chapters/ch14_auth_jwt/main.py：

八、运行验证

uv run uvicorn chapters.ch14_auth_jwt.main:app –reload –port 8814

# 登录拿 token
TOKEN=$(curl -s -X POST http://127.0.0.1:8814/login \\
-d "username=alice&password=secret" | jq -r .access_token)

# 访问受保护接口
curl -H "Authorization: Bearer $TOKEN" http://127.0.0.1:8814/me

九、本章要点

下一章开始"工程化"：规范项目目录结构。

第 15 章 规范项目目录结构

前面 14 章每个示例都在单个 main.py 里。真实项目必然要拆分： 路由 / 模型 / 服务 / 配置 / 依赖 各司其职。本章用 APIRouter + 模块化包 重构订单系统。 端口约定：8815。

一、目标结构

chapters/ch15_structure/
├── main.py # 仅负责启动 + 装载路由
├── app/
│ ├── __init__.py
│ ├── config.py # Settings
│ ├── deps.py # 公共依赖
│ ├── schemas.py # Pydantic 模型
│ ├── services.py # 业务逻辑
│ └── routers/
│ ├── __init__.py
│ ├── orders.py # /orders 子路由
│ ├── menu.py # /menu 子路由
│ └── health.py # /healthz 子路由

二、拆分原则

三、关键 API：APIRouter

# app/routers/orders.py
from fastapi import APIRouter
router = APIRouter(prefix="/orders", tags=["订单"])

@router.get("/")
def list_orders():
...

@router.get("/{oid}")
def get_order(oid: int):
...

主应用：

# main.py
from app.routers import orders, menu, health
app.include_router(orders.router)
app.include_router(menu.router)
app.include_router(health.router)

四、配置：pydantic-settings（推荐）

# app/config.py
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
shop_name: str = "咖啡工坊"
tax_rate: float = 0.06
debug: bool = False

class Config:
env_prefix = "COFFEE_" # 读 COFFEE_SHOP_NAME 等

settings = Settings()

pydantic-settings 不在基础依赖里，本章先用简单 dataclass 实现。

五、章节代码要点

chapters/ch15_structure/：

• app/config.py：Settings。
• app/schemas.py：OrderIn / OrderOut。
• app/services.py：OrderService。
• app/deps.py：分页依赖。
• app/routers/orders.py：订单路由。
• app/routers/menu.py：菜单路由。
• app/routers/health.py：健康检查。
• main.py：装配。

六、运行验证

uv run uvicorn chapters.ch15_structure.main:app –reload –port 8815

curl http://127.0.0.1:8815/orders
curl http://127.0.0.1:8815/menu
curl http://127.0.0.1:8815/healthz
curl http://127.0.0.1:8815/docs # 各 router 的 tags 已分组

七、本章要点

下一章进入"持久化"：数据库的联动开发。

第 16 章 数据库的联动开发

之前所有数据都在内存里。本章引入 SQLAlchemy 2.0 + SQLite，把订单持久化。 端口约定：8816。

一、SQLAlchemy 2.0 新写法

老写法：

db.execute("SELECT * FROM orders")

新写法（推荐）：

stmt = select(Order).where(Order.id == 1)
result = db.execute(stmt)

二、组件分层

config – 数据库 URL
db.py – engine / SessionLocal / Base
models.py – ORM 模型
schemas.py – Pydantic 模型
crud.py – 增删改查函数
routers/ – FastAPI 路由
main.py – 应用入口

三、Engine + Session

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, DeclarativeBase

DATABASE_URL = "sqlite:///./coffeecraft.db"
engine = create_engine(DATABASE_URL, echo=False, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(bind=engine, autoflush=False)

class Base(DeclarativeBase):
pass

四、ORM 模型

from sqlalchemy import String, Integer, Float
from sqlalchemy.orm import Mapped, mapped_column

class CoffeeBean(Base):
__tablename__ = "beans"
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(String(60))
price: Mapped[float] = mapped_column(Float)

五、依赖：每请求一个 Session

from typing import Generator

def get_db() –> Generator[Session, None, None]:
db = SessionLocal()
try:
yield db
finally:
db.close()

路由里使用：

@app.post("/beans")
def add(payload: BeanIn, db: Annotated[Session, Depends(get_db)]):
bean = CoffeeBean(**payload.model_dump())
db.add(bean)
db.commit()
db.refresh(bean)
return bean

六、章节代码要点

chapters/ch16_database/：

七、运行验证

uv run uvicorn chapters.ch16_database.main:app –reload –port 8816

curl -X POST http://127.0.0.1:8816/beans \\
-H "Content-Type: application/json" \\
-d '{"name":"瑰夏","price":380}'
curl http://127.0.0.1:8816/beans

数据库文件 coffeecraft.db 会自动生成。

八、本章要点

更新提示:本章已改用 lifespan 异步上下文替代已废弃的 @app.on_event("startup"), 建表语句放在 lifespan 中执行,不在 import 时副作用。 生产请把 Base.metadata.create_all 替换为 Alembic 迁移。

下一章进入"演进"：接口版本管理。

第 17 章 接口版本管理

业务会演进：旧字段要弃用、新字段要添加。最稳妥的方案是 URL 版本。 本章用 /api/v1/ 与 /api/v2/ 两套路由共存演示平滑升级。 端口约定：8817。

一、版本控制三大流派

本章使用 URL 路径。

二、目录约定

chapters/ch17_versioning/
├── main.py
├── v1/
│ ├── __init__.py
│ ├── router.py # v1 路由
│ └── schemas.py # v1 模型
└── v2/
├── __init__.py
├── router.py # v2 路由
└── schemas.py # v2 模型（字段可能不同）

三、版本差异示例

v1 的 OrderOut：

class OrderOut(BaseModel):
id: int
item: str
cups: int

v2 的 OrderOut：

class OrderOut(BaseModel):
id: int
items: list[OrderItem] # 结构化、扩展
customer: str
created: datetime
status: str

四、挂载不同版本

app.include_router(v1_router, prefix="/api/v1")
app.include_router(v2_router, prefix="/api/v2")

两个版本可以同时运行，老客户端不需立刻升级。

五、章节代码要点

chapters/ch17_versioning/：

• v1：返回简单结构。
• v2：返回丰富结构，新增 customer / created 字段。
• 公共：依赖注入共享同一份内存数据。

六、运行验证

uv run uvicorn chapters.ch17_versioning.main:app –reload –port 8817

curl http://127.0.0.1:8817/api/v1/orders
curl http://127.0.0.1:8817/api/v2/orders

观察两个版本返回结构不同。

七、本章要点

下一章进入"性能层"：同步与异步接口。

第 18 章 同步和异步接口

def 与 async def 的差异决定了一个接口能否真正发挥出 FastAPI 的高并发能力。 本章对比两种写法的行为，并给出"何时用哪个"的原则。端口约定：8818。

一、定义

二、性能实验

def slow_io() → 阻塞线程池 worker
async def slow_io_async() → 让出事件循环，挂起等待

并发请求 100 个时：

• 全用 def ：线程池满后排队。
• 全用 async def + await asyncio.sleep：100 个一起处理。
• 混合：FastAPI 自动协调。

三、章节代码要点

chapters/ch18_sync_async/main.py：

四、运行验证

uv run uvicorn chapters.ch18_sync_async.main:app –reload –port 8818

并发测试：

# 启动 5 个并发请求
for i in {1..5}; do
(time curl -s http://127.0.0.1:8818/async) &
done
wait

观察 5 个 /async 请求几乎同时返回，而 5 个 /sync 会按顺序各 200ms。

五、何时用哪个

六、踩坑点

七、本章要点

下一章进入"异步数据库"：异步数据库和异步请求。

第 19 章 异步数据库和异步请求

上一章用 httpx 演示异步 HTTP。本章把 SQLAlchemy 也升级到 异步： async_session + aiosqlite。端口约定：8819。

一、SQLAlchemy 异步引擎

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker

DATABASE_URL = "sqlite+aiosqlite:///./ch19.db"

engine = create_async_engine(DATABASE_URL, echo=False)
SessionLocal = async_sessionmaker(bind=engine, expire_on_commit=False)

sqlite+aiosqlite 与第 16 章的 sqlite:// 不同：多了驱动名前缀。

二、异步依赖

from typing import AsyncGenerator

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

async with 保证关闭。yield 让 FastAPI 在请求结束后回到这里清理。

三、异步 CRUD

from sqlalchemy import select

async def list_beans(db: AsyncSession) –> list[CoffeeBean]:
result = await db.execute(select(CoffeeBean))
return list(result.scalars())

async def add_bean(db: AsyncSession, payload: BeanIn) –> CoffeeBean:
bean = CoffeeBean(**payload.model_dump())
db.add(bean)
await db.commit()
await db.refresh(bean)
return bean

注意：所有 IO 操作都要 await。

四、异步 HTTP 请求

import httpx

async def fetch_weather():
async with httpx.AsyncClient() as client:
r = await client.get("https://…")
return r.json()

五、章节代码要点

chapters/ch19_async_db/：

六、运行验证

uv run uvicorn chapters.ch19_async_db.main:app –reload –port 8819

# 创建豆子（异步写库）
curl -X POST http://127.0.0.1:8819/beans \\
-H "Content-Type: application/json" \\
-d '{"name":"耶加雪菲","price":260,"stock":15}'

# 异步 HTTP
curl http://127.0.0.1:8819/external/ip

七、本章要点

更新提示:本章已改用 lifespan 替代已废弃的 @app.on_event("startup"), 异步建表 await conn.run_sync(Base.metadata.create_all) 放在 lifespan 内执行。

下一章进入"压测视角"：高并发注意要点。

第 20 章 高并发注意要点

当 QPS 上升到几千上万，“能跑"≠"扛得住”。本章把 压测视角 中常见的坑汇总。 端口约定：8820。本章主要是思路与代码示例。

一、三座大山

二、Uvicorn 调优

uvicorn app:app \\
–host 0.0.0.0 –port 8820 \\
–workers 4 \\
–loop uvloop \\
–http httptools \\
–backlog 2048

三、数据库连接池

engine = create_async_engine(
DATABASE_URL,
pool_size=20, # 默认 5
max_overflow=10, # 额外连接
pool_pre_ping=True, # 检测死连接
)

并发 ≈ pool_size + max_overflow；超过则排队。

四、避免阻塞事件循环

# ❌ 错误
async def handler():
time.sleep(1) # 阻塞事件循环

# ✅ 正确
async def handler():
await asyncio.sleep(1)

五、限流

from slowapi import Limiter
from slowapi.util import get_remote_address

limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter

@app.get("/api")
@limiter.limit("5/minute")
def handler(): ...

六、缓存热点

from functools import lru_cache

@lru_cache(maxsize=128)
def get_menu(): ... # 内存缓存

或者外置 Redis。

七、章节代码要点

chapters/ch20_concurrency/main.py：

八、运行验证

uv run uvicorn chapters.ch20_concurrency.main:app –reload –port 8820

并发测试（PowerShell）：

1..10 | ForEach-Object –Parallel { Invoke-WebRequest http://127.0.0.1:8820/slow-async } –ThrottleLimit 10

九、本章要点

下一章进入"质量保障"：日志 / 测试 / 接口文档。

第 21 章 日志 / 测试 / 接口文档

完结前的最后一站"质量三角"：可观测（logging）、可验证（pytest）、可发现（OpenAPI）。 端口约定：8821。

一、结构化日志

import logging
logger = logging.getLogger("coffeecraft")
logger.setLevel(logging.INFO)

生产推荐用 loguru / structlog，本章用标准库即可。

二、请求日志中间件

@app.middleware("http")
async def log_requests(request: Request, call_next):
logger.info("→ %s %s", request.method, request.url.path)
response = await call_next(request)
logger.info("← %s %s %d", request.method, request.url.path, response.status_code)
return response

三、Pytest 测试

from fastapi.testclient import TestClient
from chapters.ch21_logging_testing.main import app

def test_health():
client = TestClient(app)
r = client.get("/healthz")
assert r.status_code == 200

TestClient 基于 httpx，同步调用就能测异步接口。

四、pytest-asyncio（异步测试）

import pytest
from httpx import AsyncClient, ASGITransport

@pytest.mark.asyncio
async def test_async():
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as c:
r = await c.get("/healthz")
assert r.status_code == 200

五、自定义 OpenAPI

app = FastAPI(
title="咖啡工坊",
version="1.0.0",
description="…",
openapi_tags=[{"name":"订单","description":"顾客订单相关"}],
contact={"name":"CoffeeCraft","email":"hi@coffee.dev"},
)

还可在 app.openapi 函数里进一步定制（添加 logo、服务器 URL 等）。

六、章节代码要点

chapters/ch21_logging_testing/：

七、运行验证

uv run uvicorn chapters.ch21_logging_testing.main:app –reload –port 8821

# 跑测试
uv run pytest chapters/ch21_logging_testing/ -v

八、本章要点

下一章进入"上线"：项目部署与上线。

第 22 章 项目部署与上线

本教程的最后一站：把咖啡工坊 API 从本机搬到生产服务器。 我们覆盖最常见的三条路径：Uvicorn + 进程管理、Docker 容器、Nginx 反向代理。

一、必备清单

二、生产级 Uvicorn 命令

uvicorn chapters.ch19_async_db.main:app \\
–host 0.0.0.0 \\
–port 8000 \\
–workers 4 \\
–loop uvloop \\
–http httptools \\
–proxy-headers \\
–forwarded-allow-ips='*' \\
–log-level info

三、Docker 镜像

Dockerfile：

FROM python:3.14-slim

WORKDIR /app

# 安装 uv
COPY –from=ghcr.io/astral-sh/uv:latest /uv /uvx /usr/local/bin/

# 复制依赖清单并安装
COPY pyproject.toml uv.lock ./
RUN uv sync –frozen –no-cache

# 复制源码
COPY . .

EXPOSE 8000
CMD ["uv", "run", "uvicorn", "chapters.ch19_async_db.main:app", \\
"–host", "0.0.0.0", "–port", "8000", "–workers", "4"]

.dockerignore：

.venv
__pycache__
.pytest_cache
*.db
.git

构建与运行：

docker build -t coffeecraft:1.0 .
docker run –rm -p 8000:8000 coffeecraft:1.0

四、Nginx 反向代理

/etc/nginx/conf.d/coffee.conf：

upstream coffee_app {
server 127.0.0.1:8000;
}

server {
listen 80;
server_name coffee.example.com;

client_max_body_size 10m;

location / {
proxy_pass http://coffee_app;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
proxy_read_timeout 60s;
}
}

记得在 Uvicorn 启动时加 –proxy-headers 让 request.client.host 取到真实 IP。

五、HTTPS（Caddy 更简单）

Caddyfile：

coffee.example.com {
reverse_proxy 127.0.0.1:8000
}

Caddy 会自动申请 Let’s Encrypt 证书。

六、Systemd 进程守护

/etc/systemd/system/coffee.service：

[Unit]
Description=CoffeeCraft API
After=network.target

[Service]
WorkingDirectory=/srv/coffeecraft
ExecStart=/srv/coffeecraft/.venv/bin/uvicorn chapters.ch19_async_db.main:app \\
–host 0.0.0.0 –port 8000 –workers 4
Restart=always
User=coffee
Environment=PYTHONUNBUFFERED=1

[Install]
WantedBy=multi-user.target

sudo systemctl enable –now coffee
sudo journalctl -u coffee -f

七、Gunicorn + UvicornWorker（备选）

uv add gunicorn
uv run gunicorn chapters.ch19_async_db.main:app \\
-k uvicorn.workers.UvicornWorker \\
-w 4 -b 0.0.0.0:8000

八、生产前清单

• SECRET、JWT_SECRET 等敏感值来自环境变量。
• 日志写入文件 + logrotate。
• 数据库有备份 / 主从。
• 接口限流。
• 监控告警（Prometheus / Sentry）。
• 域名 + HTTPS。
• 防火墙只暴露 80/443。

九、本章要点

十、教程完结 🎉

到这里你已经走完了 22 个章节，覆盖了：

第 01-02 章 概念入门
第 03-07 章 基础（路由 / 参数 / 校验 / 响应）
第 08-14 章 进阶（模板 / 文件 / 异常 / 中间件 / DI / 认证）
第 15-22 章 实战（结构 / 数据库 / 版本 / 异步 / 高并发 / 测试 / 部署）

接下来的进阶路径：

• 真实数据库（PostgreSQL）+ Alembic 迁移
• Docker Compose 一键启动依赖
• Kubernetes 部署
• 监控 / 链路追踪 / 灰度发布
• CI/CD 流水线

祝你做出优秀的 API！