FastAPI依赖注入系统完全指南

📂 所属阶段：第二阶段 — 进阶黑科技（核心篇）
🔗 相关章节：FastAPI异步编程深度解析 · FastAPI中间件应用

目录

什么是依赖注入？

不用依赖注入的痛点

假设要写多个需要用户认证的路由：

# ❌ 没有依赖注入：重复逻辑散落在每个路由
from fastapi import FastAPI, HTTPException, Request
from db import fake_db

app = FastAPI()

@app.get("/users/me")
def get_current_user(request: Request):
    token = request.headers.get("Authorization")
    if not token or not token.startswith("Bearer "):
        raise HTTPException(401, "Unauthorized")
    user = fake_db.get_user(token.replace("Bearer ", ""))
    if not user:
        raise HTTPException(401, "Invalid token")
    return user

@app.get("/orders")
def get_orders(request: Request):
    token = request.headers.get("Authorization")
    # 重复的认证逻辑！
    if not token or not token.startswith("Bearer "):
        raise HTTPException(401, "Unauthorized")
    user = fake_db.get_user(token.replace("Bearer ", ""))
    if not user:
        raise HTTPException(401, "Invalid token")
    return fake_db.get_orders(user["id"])

核心问题：代码复用率低、修改成本高、可测试性差。

用依赖注入解决

FastAPI的依赖注入通过Depends让路由声明式地告诉系统“我需要什么”，而不是自己去获取：

# ✅ 依赖注入：抽成公共依赖，按需注入
from fastapi import Depends, Header, HTTPException

# 1. 抽成公共认证依赖
async def get_current_user(
    authorization: str = Header(...)
) -> dict:
    if not authorization.startswith("Bearer "):
        raise HTTPException(401, "Invalid token format")
    token = authorization.replace("Bearer ", "")
    user = fake_db.get_user(token)
    if not user:
        raise HTTPException(401, "Invalid token")
    return user

# 2. 路由只需声明需要的依赖
@app.get("/users/me")
async def get_me(user: dict = Depends(get_current_user)):
    return user

@app.get("/orders")
async def get_orders(user: dict = Depends(get_current_user)):
    return fake_db.get_orders(user["id"])

核心优势

  1. 代码复用：公共逻辑只需定义一次
  2. 解耦清晰：路由只关注业务逻辑
  3. 可测试性强：依赖可轻松替换为模拟对象
  4. 维护成本低：修改一处即可生效
  5. 内置优化：请求级缓存避免重复执行

基础用法

1. Depends函数详解

Depends(可调用对象)是FastAPI依赖注入的核心，参数可以是普通函数、异步函数、类、生成器。

# 普通函数依赖：处理查询参数
def get_search_query(q: str = "default"):
    return f"搜索内容: {q}"

@app.get("/search")
async def search(q: str = Depends(get_search_query)):
    return {"result": q}

2. 带Query/Path的依赖工厂

依赖本身也可以接受FastAPI的验证参数（如Query、Path）：

from fastapi import Query

# 分页依赖工厂
def pagination(
    page: int = Query(1, ge=1, description="页码"),
    page_size: int = Query(10, ge=1, le=100, description="每页条数")
):
    return {"page": page, "page_size": page_size}

@app.get("/items")
async def list_items(p: dict = Depends(pagination)):
    # 自动解析并验证参数
    return {"data": ["item1", "item2"], "pagination": p}

3. 类作为依赖

类依赖更适合封装复杂逻辑（如服务类），FastAPI会自动调用__init__实例化：

class AuthService:
    def __init__(self, authorization: str = Header(...)):
        self.token = authorization.replace("Bearer ", "")
        self.user = self._verify()

    def _verify(self):
        if self.token == "admin123":
            return {"id": 1, "role": "admin"}
        elif self.token == "user456":
            return {"id": 2, "role": "user"}
        raise HTTPException(401, "Invalid token")

@app.get("/profile")
async def get_profile(auth: AuthService = Depends()):
    # 类实例直接使用，简洁直观
    return auth.user

数据库连接与依赖

依赖注入是管理数据库连接的最佳实践，天然支持连接池复用和自动清理。

异步SQLAlchemy依赖示例

使用yield返回资源，请求结束后自动执行后续清理代码：

from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker

# 配置连接池
DATABASE_URL = "sqlite+aiosqlite:///./app.db"
async_engine = create_async_engine(
    DATABASE_URL,
    pool_size=20,
    max_overflow=30,
    pool_pre_ping=True
)
AsyncSessionLocal = sessionmaker(
    async_engine, class_=AsyncSession, expire_on_commit=False
)

# 数据库依赖
async def get_db() -> AsyncSession:
    async with AsyncSessionLocal() as session:
        try:
            yield session
        except Exception:
            await session.rollback()
            raise
        else:
            await session.commit()

# 使用方式
@app.get("/users/{user_id}")
async def get_user(
    user_id: int,
    db: AsyncSession = Depends(get_db)
):
    from sqlalchemy import select
    from models import User
    result = await db.execute(select(User).where(User.id == user_id))
    user = result.scalar_one_or_none()
    if not user:
        raise HTTPException(404, "User not found")
    return user

依赖链与级联

依赖可以嵌套其他依赖，形成清晰的依赖链（如：提取Token → 验证用户 → 检查权限）。

三层认证授权依赖链示例

# 第一层：提取并清理Token
def get_token(authorization: str = Header(...)):
    if not authorization.startswith("Bearer "):
        raise HTTPException(401, "Invalid token format")
    return authorization.replace("Bearer ", "")

# 第二层：根据Token验证用户（依赖第一层）
def get_current_user(token: str = Depends(get_token)):
    user = fake_db.get_user(token)
    if not user:
        raise HTTPException(401, "Invalid token")
    return user

# 第三层：检查管理员权限（依赖第二层）
def require_admin(current_user: dict = Depends(get_current_user)):
    if current_user["role"] != "admin":
        raise HTTPException(403, "Admin access required")
    return current_user

# 使用依赖链
@app.get("/admin/dashboard")
async def admin_dashboard(admin: dict = Depends(require_admin)):
    return {"admin": admin}

可选依赖与默认值

使用Optional可以创建可选依赖，适合公开/私有混合的接口。

from typing import Optional

# 可选认证依赖
async def get_optional_user(
    authorization: Optional[str] = Header(None)
) -> Optional[dict]:
    if not authorization:
        return None
    return fake_db.get_user(authorization.replace("Bearer ", ""))

# 公开接口，有用户信息返回个性化内容，否则返回通用内容
@app.get("/home")
async def home(user: Optional[dict] = Depends(get_optional_user)):
    if user:
        return {"message": f"Welcome back, {user['name']}!"}
    return {"message": "Welcome, guest!"}

依赖缓存机制

1. 默认请求级缓存

同一请求中多次使用同一个依赖，只执行一次，避免重复查询/计算：

def get_user():
    print("🔍 查询用户（只打印一次）")
    return {"id": 1, "name": "Alice"}

@app.get("/a")
@app.get("/b")
async def endpoints(user: dict = Depends(get_user)):
    return {"user": user}

2. 禁用缓存：use_cache=False

如果需要每次获取新的结果（如随机数、实时ID），可以禁用缓存：

import uuid

def get_request_id():
    return str(uuid.uuid4())

@app.get("/uuid")
async def get_uuid(
    req_id1: str = Depends(get_request_id, use_cache=False),
    req_id2: str = Depends(get_request_id, use_cache=False)
):
    # req_id1和req_id2会不同
    return {"req_id1": req_id1, "req_id2": req_id2}

实战：认证授权体系

依赖分层设计

请求 → 中间件（可选） → 依赖注入链
                  ↓
         1. get_db          （数据库连接）
         2. get_token       （提取Token）
         3. get_current_user（验证用户）
         4. require_roles   （角色检查）
                  ↓
             路由处理器

核心代码示例

# auth_dependencies.py
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from typing import List

security = HTTPBearer(auto_error=False)

def decode_token(token: str) -> dict:
    # 模拟JWT验证
    if token == "admin":
        return {"id": 1, "role": "admin", "sub": "admin@example.com"}
    elif token == "editor":
        return {"id": 2, "role": "editor", "sub": "editor@example.com"}
    raise HTTPException(status.HTTP_401_UNAUTHORIZED, "Invalid token")

def require_roles(allowed_roles: List[str]):
    """角色权限检查工厂函数"""
    async def checker(
        cred: HTTPAuthorizationCredentials = Depends(security)
    ) -> dict:
        user = decode_token(cred.credentials)
        if user["role"] not in allowed_roles:
            raise HTTPException(status.HTTP_403_FORBIDDEN, "Permission denied")
        return user
    return checker

# 预定义常用角色依赖
require_admin = require_roles(["admin"])
require_editor = require_roles(["admin", "editor"])
# main.py
from fastapi import FastAPI, Depends
from auth_dependencies import require_admin, require_editor

app = FastAPI()

@app.get("/public")
async def public():
    return {"status": "ok"}

@app.get("/me")
async def me(user: dict = Depends(require_editor)):
    return user

@app.delete("/users/{user_id}")
async def delete_user(user_id: int, _: dict = Depends(require_admin)):
    # 下划线表示不需要使用返回值
    return {"deleted": user_id}

进阶用法

1. 全局路由级依赖

在APIRouter或单独的路由参数中添加dependencies，可以让一组路由共用前置依赖：

from fastapi import APIRouter

admin_router = APIRouter(
    prefix="/admin",
    dependencies=[Depends(require_admin)]  # 所有admin路由都需要管理员权限
)

@admin_router.get("/users")
async def list_admin_users():
    return {"users": ["admin1"]}

app.include_router(admin_router)

避坑指南与性能优化

避坑指南

  1. 避免副作用：依赖函数应该是纯函数或幂等函数
  2. 避免循环依赖：不要让A依赖B，B又依赖A
  3. 避免过度嵌套：依赖链不要超过3-4层
  4. 避免异步依赖中的阻塞操作：把阻塞代码移到线程池

性能优化

  1. 善用默认请求级缓存：减少重复查询
  2. 配置合理的数据库连接池：避免连接耗尽
  3. 依赖执行顺序优化：把快速验证的依赖放在前面（如Token格式检查），快速失败

总结

模式说明适用场景
Depends(func)基础依赖注入参数提取、简单验证
Depends(Class)类依赖服务类、复杂逻辑
依赖链嵌套依赖认证→用户→权限
use_cache=False禁用缓存实时数据、随机数
全局dependencies路由组依赖权限控制分组

💡 核心思想：依赖注入让代码变得可测试、可复用、声明式、易维护，是构建企业级FastAPI应用的核心工具。

🔗 官方扩展阅读