FastAPI Depends() 实战：DI 让 testing 和 modularity 都受益

起因

FastAPI 的 Depends() 是其设计精华之一。新人常忽视它，把所有 view 写
成"自己 new DB connection + 拉 user 信息"，导致：

测试时无法 mock DB
业务逻辑跟 framework 耦合
改 auth 流程要改 100 处 view

学会用 Depends 后代码组织上一个台阶。

1. 基础

from fastapi import FastAPI, Depends

app = FastAPI()

def get_db():
    db = Session()
    try:
        yield db
    finally:
        db.close()

@app.get('/users/{uid}')
def read_user(uid: int, db = Depends(get_db)):
    return db.query(User).filter_by(id=uid).first()

Depends(get_db)：

每个请求自动调 get_db()
yield 之前的代码：创建 session
yield 之后的代码：关闭 session
类似 Django 的 middleware 但更灵活

2. 嵌套依赖

def get_db(): ...

def get_current_user(
    token: str = Depends(oauth2_scheme),
    db: Session = Depends(get_db),
) -> User:
    user = decode_token(token, db)
    if not user:
        raise HTTPException(401)
    return user

@app.get('/me')
def me(user: User = Depends(get_current_user)):
    return user

FastAPI 自动解析依赖图：me 需要 get_current_user 需要 get_db
和 oauth2_scheme。
同一个请求里 get_db 只调一次（即使多个依赖都需要它）。

3. 权限检查依赖

def require_admin(user: User = Depends(get_current_user)) -> User:
    if user.role != 'admin':
        raise HTTPException(403, 'admin required')
    return user

@app.delete('/users/{uid}')
def delete_user(uid: int,
                admin: User = Depends(require_admin),
                db: Session = Depends(get_db)):
    db.query(User).filter_by(id=uid).delete()
    db.commit()

权限检查从 view body 抽出来 → 复用 + 类型安全。

4. 类做依赖（state-ful）

class CommonQueryParams:
    def __init__(self, q: str = '', skip: int = 0, limit: int = 20):
        self.q = q
        self.skip = skip
        self.limit = limit

@app.get('/search')
def search(params: CommonQueryParams = Depends()):
    # params.q / params.skip / params.limit
    ...

Depends() 无参数时自动用类型 CommonQueryParams 当 dependency。
重复的 query param 模式抽成 class。

5. router-level dependency

from fastapi import APIRouter

admin_router = APIRouter(
    prefix='/admin',
    dependencies=[Depends(require_admin)],   # router 级依赖
)

@admin_router.get('/dashboard')
def dashboard():
    pass

@admin_router.get('/users')
def users():
    pass

admin_router 下所有 endpoint 自动要求 admin。不需要每个 view 写 Depends。

或 app 级：

app = FastAPI(dependencies=[Depends(log_request)])   # 全局

6. 测试：覆盖依赖

def get_db_test():
    return MockDB()

app.dependency_overrides[get_db] = get_db_test

def test_read_user():
    client = TestClient(app)
    response = client.get('/users/1')
    assert response.json() == {'id': 1, 'name': 'mock'}

dependency_overrides 让测试用 mock 实现替换生产依赖。
整个测试不真连 DB。

对比："view 内直接 import DB" → 没法 mock，要么 mock 整个 module，
要么真起 DB。

7. Generator dependency（cleanup 必须 finally）

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

不 try/finally 的话，view 内 exception → cleanup 不跑 → 连接泄露。
永远 try/finally。

8. 同步 vs async dependency

混用 OK：

def get_settings():           # 同步
    return Settings()

async def get_user():         # async
    return await fetch_user()

@app.get('/x')
async def x(
    s: Settings = Depends(get_settings),
    u: User = Depends(get_user),
):
    ...

FastAPI 自动决定 thread pool 还是 await。

9. Sub-app / mount

admin_app = FastAPI()

@admin_app.get('/stats')
def stats():
    pass

app.mount('/admin', admin_app)

完全独立的 FastAPI 应用挂载到主 app。各自有独立 dependency / docs /
middleware。复杂场景拆分。

10. lifecycle hooks vs Depends

@app.on_event('startup')
async def startup():
    app.state.db = await create_pool()

@app.on_event('shutdown')
async def shutdown():
    await app.state.db.close()

或现代 lifespan：

from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app):
    app.state.db = await create_pool()
    yield
    await app.state.db.close()

app = FastAPI(lifespan=lifespan)

应用启动 / 关闭一次性资源（DB pool / Redis client）放这里，不是
per-request Depends。

之后 Depends 引用：

def get_db(request: Request):
    return request.app.state.db.acquire()

完整模板：blog API 骨架

# deps.py
from fastapi import Depends, HTTPException, Request
from sqlalchemy.orm import Session

def get_db(request: Request) -> Session:
    db = request.app.state.SessionLocal()
    try:
        yield db
    finally:
        db.close()

def get_current_user(
    token: str = Depends(oauth2_scheme),
    db: Session = Depends(get_db),
) -> User:
    user = decode_jwt(token, db)
    if not user:
        raise HTTPException(401)
    return user

def get_post(post_id: int, db: Session = Depends(get_db)) -> Post:
    post = db.query(Post).get(post_id)
    if not post:
        raise HTTPException(404)
    return post

def require_post_owner(
    post: Post = Depends(get_post),
    user: User = Depends(get_current_user),
) -> Post:
    if post.author_id != user.id:
        raise HTTPException(403)
    return post

# views.py
@app.get('/posts/{post_id}')
def read_post(post: Post = Depends(get_post)):
    return post

@app.put('/posts/{post_id}')
def update_post(
    data: UpdatePostIn,
    post: Post = Depends(require_post_owner),
    db: Session = Depends(get_db),
):
    post.title = data.title
    db.commit()
    return post

@app.delete('/posts/{post_id}')
def delete_post(
    post: Post = Depends(require_post_owner),
    db: Session = Depends(get_db),
):
    db.delete(post)
    db.commit()
    return {'ok': True}

每个 view 体内不超过 5 行业务逻辑。权限 / DB session / 对象加载都
declarative。

性能注意

每请求依赖图都重新解析。深嵌套 / 重操作 dependency 会累加。
但通常 dependency 都很轻（DB query / dict lookup），impact 极小。

与 Django / Flask 对比

Django：middleware + @login_required 装饰器 + view-local code。
Flask：@app.before_request + 装饰器 + flask-login。

FastAPI Depends 优点：

type-safe（IDE / mypy 知道 user 是 User）
显式（看签名就知道这个 endpoint 需要什么）
测试友好（dependency_overrides）

但学习曲线略陡。习惯后回不去。

踩过的坑

Depends 写位置错：
python def view(user=Depends(get_user), uid: int = 1): # ❌
Depends 不能有 default。改顺序：
python def view(uid: int, user=Depends(get_user)):
循环依赖：A depends B，B depends A → import error。架构上重新
设计；通常说明边界划错。
dependency 内 raise → FastAPI 自动 422 / 配你的 HTTPException。
写 raise HTTPException(401, "Bad token") 别 return None。
测试 dependency_overrides 没清 → 测试间互相污染。每个测试
app.dependency_overrides.clear() 或用 fixture。
Depends() 无参 + 类：必须显式 type annotation：
python def view(params: CommonQueryParams = Depends()):
不写 type FastAPI 不知道用什么。