FastAPI-03.依赖注入与分层

1. 问题：路由函数越来越胖

典型坏味道：

@app.get("/items/{item_id}")
async def get_item(item_id: int):
    db = connect_db()          # 连接从哪来？
    user = parse_jwt(...)      # 鉴权混在业务里
    item = db.query(item_id)   # SQL 与 HTTP 缠在一起
    if not item:
        raise HTTPException(404)
    return item

依赖注入（Dependency Injection，DI）要回答：谁创建连接、谁校验身份、谁查库——路由只编排。

段末注释：DI 指由框架在调用路由前，按声明自动解析并注入依赖对象，而不是在函数内部 new 或全局单例硬取。

---

2. 费曼白话：前台、后勤、柜员

延续 01 的比喻：

• 柜员 = 路由函数（接待这一单客户）。
• 后勤 = Depends(...) 提供的对象（数据库会话、当前用户、配置）。
• 前台系统 = FastAPI 依赖图解析器：先算后勤，再叫柜员；后勤之间也可互相依赖。

双重编码（结构图）：

请求 → 中间件（04）
         → 依赖树解析（Depends）
              ├─ get_settings()
              ├─ get_db() ──依赖──► get_settings()
              └─ get_current_user() ──依赖──► get_db()
         → 路由函数(item_id, db, user)
         → 响应

---

3. Depends 组块

G1 简单依赖

from fastapi import Depends, FastAPI

app = FastAPI()


def pagination(skip: int = 0, limit: int = 20):
    return {"skip": skip, "limit": limit}


@app.get("/items")
async def list_items(page: dict = Depends(pagination)):
    return page

G2 可复用依赖（yield 清理）

from collections.abc import Generator


def get_db() -> Generator:
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

• 适用：每请求一个 DB 会话，结束必关。
• 误用：在 yield 前做重 IO 却不缓存——应放 lifespan（01）。

G3 依赖链

def get_settings():
    return Settings()


def get_db(settings: Settings = Depends(get_settings)):
    ...


def get_current_user(
    token: str = Depends(oauth2_scheme),
    db=Depends(get_db),
):
    ...

解析顺序：settings → db → user（框架按图拓扑排序）。

G4 Annotated 别名（推荐）

from typing import Annotated

DbDep = Annotated[Session, Depends(get_db)]
# 一种缩写形式
# DbDep = Annotated[get_db, Depends()]
UserDep = Annotated[User, Depends(get_current_user)]


@app.get("/me")
async def read_me(user: UserDep, db: DbDep):
    return user

路由签名更短，测试时 app.dependency_overrides 更好换假实现。

G5 路由级依赖

router = APIRouter(dependencies=[Depends(verify_api_key)])


@router.get("/secret")
async def secret():
    return {"ok": True}

整组路由共享同一依赖，不必每个函数重复写。

---

G6 类依赖

同时需要注意，DEPEND不只是可以使用函数，也可以使用类，例如：

fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}]

class CommonQueryParams:
    def __init__(self, q: str | None = None, skip: int = 0, limit: int = 100):
        self.q = q
        self.skip = skip
        self.limit = limit

@app.get("/items/")
async def read_items(commons: Annotated[CommonQueryParams, Depends()]):
    response = {}
    if commons.q:
        response.update({"q": commons.q})
    items = fake_items_db[commons.skip : commons.skip + commons.limit]
    response.update({"items": items})
    return response

4. 分层组块（工程目录）

小中型 API 推荐四层，依赖方向单向向下：

app/
├── main.py              # FastAPI 实例、lifespan、挂 router
├── core/
│   ├── config.py        # Settings（pydantic-settings）
│   └── deps.py          # get_db、get_current_user
├── routers/
│   └── items.py         # 只处理 HTTP：参数、状态码
├── services/
│   └── items.py         # 业务规则、事务边界
├── repositories/
│   └── items.py         # 持久化/SQL/外部 API
└── schemas/
    └── items.py         # Pydantic 模型（Create/Read）

层	允许	禁止
router	调 service、抛 HTTPException	直接写 SQL
service	调 repo、业务校验	依赖 Request 对象
repository	CRUD、查询	决定 HTTP 状态码
schemas	数据结构	业务逻辑

最小贯通示例：

# services/items.py
def get_item_or_none(service_db, item_id: int):
    return repo_get_by_id(service_db, item_id)


# routers/items.py
@router.get("/{item_id}", response_model=ItemRead)
async def get_item(item_id: int, db: DbDep):
    item = get_item_or_none(db, item_id)
    if item is None:
        raise HTTPException(status_code=404, detail="Item not found")
    return item

---

5. 测试挂钩：dependency_overrides

def fake_db():
    yield FakeSession()


app.dependency_overrides[get_db] = fake_db
# 测试结束后：app.dependency_overrides.clear()

这是 DI 的工程收益：不换路由代码，换实现（07 展开）。

---

6. 合书自测

1. 用一句话向同事解释 Depends 在请求里的执行时机。
2. yield 依赖和 lifespan 各解决什么？各跑几次？
3. 画 router → service → repo 三层，标一条「读 item」调用链。

---

7. 闪卡候选

正面	背面
Depends 解析顺序？	按依赖图，被依赖者先执行
dependency_overrides 用途？	测试/本地替换依赖实现
router 能否 import repository？	能但不推荐；经 service 保持业务集中
APIRouter(dependencies=[...])？	路由组级共享依赖

---

小结

• DI 把横切能力（DB、用户、配置）从路由里拔出，FastAPI 用 Depends 自动接线。
• 分层 把 HTTP 与业务、持久化分开，后续鉴权（06）、测试（07）、MCP 实战（08）都受益。
• 下一篇 04 中间件、异常与日志：全局横切与排错。