FastAPI 依赖注入指南

1. 基础 Depends

from fastapi import FastAPI, Depends, Query
from typing import Annotated

app = FastAPI()

def common_params(
    skip: int = Query(0, ge=0),
    limit: int = Query(10, ge=1, le=100),
) -> dict:
    return {"skip": skip, "limit": limit}

CommonParams = Annotated[dict, Depends(common_params)]

@app.get("/items")
async def list_items(params: CommonParams):
    return {"params": params}

2. 基于类的依赖

class Pagination:
    def __init__(self, page: int = 1, size: int = 20):
        self.page = page
        self.size = size
        self.offset = (page - 1) * size

@app.get("/products")
async def list_products(pagination: Annotated[Pagination, Depends()]):
    items = await Product.find_all(skip=pagination.offset, limit=pagination.size)
    return {"items": items}

3. Yield 依赖 — 数据库会话

from sqlalchemy.orm import Session
from database import SessionLocal

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

DB = Annotated[Session, Depends(get_db)]

@app.get("/users/{user_id}")
async def get_user(user_id: int, db: DB):
    user = db.query(User).filter(User.id == user_id).first()
    if not user:
        raise HTTPException(status_code=404, detail="用户不存在")
    return user

4. 子依赖链

from fastapi.security import OAuth2PasswordBearer
from jose import jwt, JWTError

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/token")

async def get_token_data(token: Annotated[str, Depends(oauth2_scheme)]):
    try:
        return jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
    except JWTError:
        raise HTTPException(status_code=401, detail="token 验证失败")

async def get_current_user(
    payload: Annotated[dict, Depends(get_token_data)],
    db: DB,
) -> User:
    user = db.query(User).filter(User.id == payload.get("sub")).first()
    if not user:
        raise HTTPException(status_code=401, detail="用户不存在")
    return user

@app.get("/me")
async def read_me(current_user: Annotated[User, Depends(get_current_user)]):
    return current_user

5. 路由级依赖

from fastapi import APIRouter

router = APIRouter(
    prefix="/admin",
    tags=["admin"],
    dependencies=[Depends(require_admin)],
)

@router.get("/stats")
async def get_stats(db: DB):
    return await compute_stats(db)

6. 依赖作用域