第 2 章 · 路由、依赖注入与 lifespan
本章目标:掌握 FastAPI 路由装饰器与 APIRouter 模块化拆分;理解 Depends 依赖注入机制,实现可复用的数据库会话、配置、鉴权占位;使用 lifespan 上下文管理器处理应用启动与关闭(连接池初始化、资源释放);将 svc-demo 从单文件 main.py 重构为可维护的多模块结构;能对照 Django URLConf 与 Flask Blueprint 说明差异。
学时建议:4~5 小时(含 1.5 小时跟练)
前置:完成 fastapi-web ch01;学过 flask-web ch02(蓝图) 或 django-web ch01(urls.py) 更佳。
2.1 场景说明:svc-demo 接口越来越多
ch01 的 main.py 只有 3 个路由。随着 svc-demo 接入商品、订单、用户等域,需要:
| 需求 | ch01 现状 | 本章目标 |
|---|---|---|
| 路由组织 | 全写在 main.py | APIRouter 按业务分包 |
| 公共逻辑 | 无 | Depends 注入配置、日志、DB 占位 |
| 启动初始化 | 无 | lifespan 打开连接池、预热缓存 |
| 关闭清理 | 直接杀进程 | lifespan 优雅关闭连接 |
| URL 前缀 | 手写完整路径 | prefix="/api/v1/products" |
对照其他框架:
| FastAPI | Django(shop-demo) | Flask(api-demo) |
|---|---|---|
APIRouter | include(urls) | Blueprint |
Depends(get_db) | 中间件 / 装饰器 | g / 装饰器 |
lifespan | AppConfig.ready() | @app.before_first_request(已废弃) |
2.2 推荐目录结构
svc-demo/
├── app/
│ ├── __init__.py
│ ├── main.py # FastAPI 工厂 + lifespan
│ ├── config.py # 配置(占位)
│ ├── dependencies.py # 公共 Depends
│ └── routers/
│ ├── __init__.py
│ ├── health.py
│ └── products.py # 商品路由(内存假数据)
├── requirements.txt
└── README.md
启动方式改为:
uvicorn app.main:app --reload --host 127.0.0.1 --port 8000
2.3 APIRouter 基础
2.3.1 健康检查路由 app/routers/health.py
from fastapi import APIRouter
router = APIRouter(tags=["健康检查"])
@router.get("/health")
async def health():
return {"status": "healthy"}
@router.get("/ready")
async def ready():
"""就绪探针:可扩展为检查数据库连通性(ch05)"""
return {"ready": True}
2.3.2 商品路由 app/routers/products.py
from fastapi import APIRouter, Depends, HTTPException
from app.dependencies import get_pagination, Pagination
router = APIRouter(prefix="/api/v1/products", tags=["商品"])
# 教学用内存数据,ch05 换为数据库
_FAKE_PRODUCTS = [
{"id": 1, "sku": "SKU-001", "name": "无线鼠标", "price": 89.0},
{"id": 2, "sku": "SKU-002", "name": "机械键盘", "price": 399.0},
{"id": 3, "sku": "SKU-003", "name": "USB-C 集线器", "price": 129.0},
]
@router.get("/")
async def list_products(pagination: Pagination = Depends(get_pagination)):
start = pagination.offset
end = start + pagination.limit
items = _FAKE_PRODUCTS[start:end]
return {"total": len(_FAKE_PRODUCTS), "items": items}
@router.get("/{product_id}")
async def get_product(product_id: int):
for p in _FAKE_PRODUCTS:
if p["id"] == product_id:
return p
raise HTTPException(status_code=404, detail="商品不存在")
2.3.3 注册路由 app/main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from app.routers import health, products
@asynccontextmanager
async def lifespan(app: FastAPI):
# ── 启动 ──
print("[lifespan] svc-demo 启动:初始化资源…")
app.state.startup_message = "svc-demo is running"
yield
# ── 关闭 ──
print("[lifespan] svc-demo 关闭:释放资源…")
def create_app() -> FastAPI:
app = FastAPI(
title="svc-demo API",
version="0.2.0",
lifespan=lifespan,
)
app.include_router(health.router)
app.include_router(products.router)
return app
app = create_app()
| 参数 | 作用 | 示例 |
|---|---|---|
prefix | 路由组统一前缀 | /api/v1/products |
tags | Swagger 文档分组 | ["商品"] |
include_router | 挂载子路由 | 类似 Flask register_blueprint |
2.4 Depends 依赖注入
依赖注入(DI)让你把「获取数据库会话」「解析分页参数」「校验 API Key」等逻辑写成可复用函数,由框架在每次请求前自动调用。
2.4.1 分页依赖 app/dependencies.py
from dataclasses import dataclass
from fastapi import Query
@dataclass
class Pagination:
offset: int
limit: int
def get_pagination(
offset: int = Query(0, ge=0, description="跳过条数"),
limit: int = Query(20, ge=1, le=100, description="每页条数"),
) -> Pagination:
return Pagination(offset=offset, limit=limit)