FastAPI依赖注入与认证

Web开发专题 · 掌握依赖注入与认证机制

专题：Python Web开发系统学习

关键词：Python, Web开发, 依赖注入, Depends, OAuth2, JWT, Token认证, FastAPI安全, password_hash

一、依赖注入概述

依赖注入（Dependency Injection, DI）是一种软件设计模式，其核心思想是将对象所需的依赖（如数据库连接、配置信息、认证服务等）从外部传入，而非由对象自身创建。这种"控制反转"（Inversion of Control）的方式极大提升了代码的灵活性和可维护性。

FastAPI框架内置了强大的依赖注入系统，这是其区别于Flask和Django等传统Python Web框架的重要特性之一。在Flask中，开发者通常依赖全局代理对象（如current_app、request、g）来访问共享资源；Django则依赖全局设置和中间件体系。这些方式虽然便捷，但在测试和模块化方面存在天然缺陷——全局状态使得单元测试变得困难，组件间的隐式耦合增加了重构成本。

FastAPI的依赖注入系统采用显式声明的方式，路径函数通过类型注解明确声明其需要的依赖项，框架在运行时自动解析并注入这些依赖。这种方式带来的核心优势包括：

可重用性：依赖项可以跨多个路径操作共享，避免重复代码
可测试性：在测试中可以轻松替换依赖项的实现，无需修改业务逻辑
松耦合：组件之间通过接口（依赖声明）交互，而非直接依赖具体实现
类型安全：利用Python类型注解，在编码阶段即可捕获依赖类型错误
自动文档：依赖项自动集成到OpenAPI文档中，API消费者可清晰了解认证等信息

依赖注入的核心价值不在于技术本身，而在于它所倡导的"面向接口编程"思想。当你的函数声明"我需要什么"而不是"我去拿什么"时，代码的可测试性和可维护性便得到了质的提升。

二、基本依赖使用

Depends() 函数

Depends()是FastAPI依赖注入系统的核心入口。它是一个可调用对象，接收一个可调用对象（函数、类或任意实现了__call__方法的对象）作为参数，并返回一个特殊标记。FastAPI在路由处理时识别此标记，自动执行依赖函数并将结果注入到路径操作中。

函数作为依赖

最简单的依赖形式是普通的Python函数。依赖函数可以像路径函数一样接收参数，包括查询参数、路径参数、请求体等，FastAPI会自动处理这些参数的解析和验证。

from fastapi import FastAPI, Depends, Query

app = FastAPI()

# 定义一个依赖函数，模拟数据库查询公共参数
def common_params(
    page: int = Query(1, ge=1),
    size: int = Query(20, ge=1, le=100),
    sort: str = "created_at"
):
    return {"page": page, "size": size, "sort": sort}

# 路径函数通过 Depends() 声明依赖
@app.get("/items")
async def list_items(params: dict = Depends(common_params)):
    # params 自动包含 page, size, sort 三个字段
    return {"params": params}

上述代码中，common_params函数本身就是一个独立的可测试单元，不依赖任何FastAPI全局状态。路径函数list_items通过Depends(common_params)声明对其的依赖，框架自动完成参数提取、验证和注入。

类作为依赖

除了函数，类也可以作为依赖使用。FastAPI会调用类的构造函数（__init__）并返回实例。当依赖项包含较多配置项或有状态时，类形式的依赖更为合适。

from fastapi import FastAPI, Depends
from typing import Optional

class PaginationParams:
    def __init__(self, page: int = Query(1, ge=1),
                 size: int = Query(20, ge=1, le=100)):
        self.page = page
        self.size = size
        self.offset = (page - 1) * size

@app.get("/users")
async def list_users(pag: PaginationParams = Depends()):
    # pag 是 PaginationParams 的实例
    return {"offset": pag.offset, "limit": pag.size}

注意：当使用类作为依赖时，Depends()可以不传参数（FastAPI会自动识别类本身），也可以显式写为Depends(PaginationParams)。两种方式效果相同。

共享依赖

依赖项可以在多个路径操作之间共享，这是依赖注入系统提升代码复用性的核心体现。只需在多个路径函数的参数中声明相同的依赖，该依赖的逻辑便被自动复用，无需任何额外的注册或配置。

例如，在多个需要分页的接口中共享common_params依赖，或者在多个需要数据库会话的接口中共享数据库连接依赖，都可以大幅减少重复代码。

三、依赖的高级用法

可调用类作为依赖

如果一个类实现了__call__方法，其实例可以作为可调用对象传递给Depends()。这种方式在需要维护依赖状态或进行复杂初始化时特别有用。

class AuthGuard:
    def __init__(self, api_key: str):
        self.api_key = api_key

    async def __call__(self, token: str = Header(...)):
        if token != self.api_key:
            raise HTTPException(status_code=403)
        return {"authorized": True}

auth = AuthGuard(api_key="secret-key")

@app.get("/secure")
async def secure_endpoint(auth_result: dict = Depends(auth)):
    return {"message": "Access granted", "auth": auth_result}

依赖嵌套（依赖的依赖）

FastAPI依赖注入系统支持任意深度的嵌套。一个依赖可以声明对另一个依赖的依赖，框架会自动解析整个依赖链。这种能力使得构建复杂的依赖关系图变得简单而清晰。

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

def get_current_user(db: Database = Depends(get_db)):
    # 此依赖依赖 get_db 提供的数据库连接
    user = db.query(User).first()
    return user

@app.get("/profile")
async def get_profile(
    user: User = Depends(get_current_user)
):
    return user

在上面的例子中，get_profile依赖get_current_user，而get_current_user又依赖get_db。FastAPI会自动按顺序解析：先执行get_db获取数据库连接，再执行get_current_user获取当前用户，最后将用户对象注入到get_profile中。

路径装饰器依赖（dependencies参数）

有时候我们需要的依赖并不返回值（比如只需要执行某个校验逻辑），或者需要应用到整个路由组。此时可以使用路径装饰器的dependencies参数，它接受一个Depends列表，这些依赖会被执行但不关心其返回值。

@app.get("/admin/dashboard",
        dependencies=[Depends(verify_admin_role)])
async def admin_dashboard():
    return {"data": "sensitive data"}

全局依赖覆盖（dependency_overrides）

app.dependency_overrides是一个字典，允许在应用级别替换依赖实现。这个机制是FastAPI测试能力的基石，也是生产环境中条件性切换依赖实现的利器。dependency_overrides的键是原始依赖函数/类，值是替换后的可调用对象。

# 定义原始依赖
def get_settings():
    return Settings(env="production")

# 在测试中覆盖
def get_test_settings():
    return Settings(env="test")

app.dependency_overrides[get_settings] = get_test_settings

# 后续所有使用 Depends(get_settings) 的地方都会自动使用 get_test_settings

四、依赖覆盖测试

依赖覆盖（Dependency Override）是FastAPI最强大的测试特性之一。它允许在不修改任何业务代码的前提下，在测试环境中完全替换依赖行为，这比Django和Flask中通过mock全局对象的方式更加优雅和可靠。

基础测试模式

使用FastAPI的TestClient进行测试时，标准的做法是创建应用的副本并在其上设置依赖覆盖，确保测试环境与生产环境隔离。

from fastapi.testclient import TestClient

# 创建测试客户端
client = TestClient(app)

# 模拟认证依赖
def mock_get_current_user():
    return User(id=1, username="testuser")

# 覆盖依赖
app.dependency_overrides[get_current_user] = mock_get_current_user

# 执行测试
response = client.get("/profile")
assert response.status_code == 200
assert response.json()["username"] == "testuser"

# 清理覆盖（重要！避免影响其他测试）
app.dependency_overrides.clear()

使用 pytest fixture 管理依赖覆盖

在实际项目中，推荐将依赖覆盖的管理封装到pytest fixture中，确保每个测试用例的依赖环境独立且可自动清理。

import pytest
from fastapi.testclient import TestClient

@pytest.fixture()
def client():
    # 在所有测试前设置依赖覆盖
    app.dependency_overrides[get_db] = get_test_db
    app.dependency_overrides[get_current_user] = get_test_user
    yield TestClient(app)
    # 测试结束后清理，防止泄漏
    app.dependency_overrides.clear()

def test_create_item(client):
    response = client.post("/items", json={"name": "test"})
    assert response.status_code == 201

测试最佳实践

隔离测试数据库：为每个测试用例创建独立的数据库会话，避免测试数据互相污染
覆盖所有外部依赖：包括数据库、缓存、第三方API调用等，确保测试不依赖外部环境
使用依赖覆盖而非mock.patch：fastapi的dependency_overrides是官方推荐的测试方式，比mock.patch更稳定且与框架集成更好
定时清理覆盖：在fixture的teardown阶段清理覆盖，避免测试间泄漏

五、OAuth2密码流认证

OAuth2是一种授权框架，定义了多种授权流程（称为grant types）。在FastAPI中，最常用的认证方式是OAuth2密码流（Password Flow），即客户端使用用户名和密码直接向认证服务器换取访问令牌（Access Token）。

OAuth2PasswordBearer

OAuth2PasswordBearer是FastAPI提供的依赖类，用于从请求头中提取Bearer Token。它会在请求头中查找Authorization: Bearer <token>，如果找不到则自动返回401错误。

from fastapi.security import OAuth2PasswordBearer

# tokenUrl 指向获取 token 的路径
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login")

@app.post("/auth/login")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    # 验证用户名密码并返回 token
    user = authenticate_user(form_data.username, form_data.password)
    if not user:
        raise HTTPException(status_code=401, detail="Invalid credentials")
    token = create_access_token(data={"sub": user.username})
    return {"access_token": token, "token_type": "bearer"}

用户认证依赖组合

在实际应用中，认证系统通常由多个层次化的依赖组成：首先提取token，然后验证token有效性，最后从数据库加载完整的用户信息。

async def get_current_user(
    token: str = Depends(oauth2_scheme),
    db: Database = Depends(get_db)
) -> User:
    # 解码 token 获取用户标识
    payload = decode_token(token)
    username = payload.get("sub")
    if username is None:
        raise HTTPException(status_code=401)
    # 从数据库加载用户
    user = db.query(User).filter(User.username == username).first()
    if user is None:
        raise HTTPException(status_code=401)
    return user

# 在需要认证的接口中使用
@app.get("/users/me")
async def read_current_user(
    current_user: User = Depends(get_current_user)
):
    return current_user

密码验证

密码绝不能以明文形式存储。正确的做法是存储密码的哈希值，验证时比较用户输入的密码哈希与存储的哈希是否匹配。FastAPI推荐使用passlib库进行密码哈希处理。

from passlib.context import CryptContext

# 配置密码上下文，推荐使用 bcrypt 算法
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def hash_password(password: str) -> str:
    return pwd_context.hash(password)

def verify_password(plain_password: str, hashed_password: str) -> bool:
    return pwd_context.verify(plain_password, hashed_password)

六、JWT（JSON Web Token）

JWT是目前最流行的Token格式，被广泛应用于分布式系统的认证和授权。它的自包含特性使得服务端无需维护会话状态，非常适合微服务架构和无状态API设计。

JWT结构

一个JWT由三部分组成，以点号（.）分隔：

Header（头部）：包含令牌类型（typ）和签名算法（alg），如{"alg": "HS256", "typ": "JWT"}
Payload（负载）：包含声明（claims），即实际传递的数据，如用户ID、过期时间等。标准注册声明包括iss（签发者）、sub（主题）、exp（过期时间）、iat（签发时间）等
Signature（签名）：对Header和Payload进行签名，确保Token未被篡改。签名算法使用Header中指定的算法，密钥由服务端保管

JWT的格式示例：eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJ1c2VyMSJ9.xxxxx。每一部分都是Base64URL编码的JSON字符串，客户端和服务端均可解码查看Payload内容，但只有持有密钥的服务端才能验证签名。

Token生成与验证（python-jose）

python-jose库提供了JWT的编码与解码功能，推荐与FastAPI配合使用。

from jose import jwt, JWTError
from datetime import datetime, timedelta

# 密钥配置（生产环境中应使用环境变量存储）
SECRET_KEY = "your-secret-key-must-be-very-long"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

def create_access_token(data: dict, expires_delta: timedelta = None):
    to_encode = data.copy()
    expire = datetime.utcnow() + (expires_delta or
              timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES))
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

def decode_token(token: str) -> dict:
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        return payload
    except JWTError:
        raise HTTPException(
            status_code=401, detail="Invalid or expired token"
        )

Token过期与刷新机制

Token设置过期时间是一项重要的安全措施。通常Access Token的过期时间较短（15-30分钟），配合Refresh Token（刷新令牌，有效期更长，如7天）使用，在保证安全性的同时提升用户体验。

Refresh Token的工作流程是：当Access Token过期时，客户端使用Refresh Token向专门的刷新接口请求新的Access Token。服务端验证Refresh Token的有效性后，签发新的Access Token。Refresh Token通常存储在服务端数据库中以支持撤销操作。

将JWT整合到依赖中

将JWT解码和用户认证逻辑封装到依赖函数中，是FastAPI认证的标准做法。通过依赖的组合，可以轻松实现不同级别的访问控制。

async def get_current_active_user(
    current_user: User = Depends(get_current_user)
) -> User:
    if current_user.disabled:
        raise HTTPException(status_code=400,
                            detail="Inactive user")
    return current_user

async def get_current_admin_user(
    current_user: User = Depends(get_current_active_user)
) -> User:
    if current_user.role != "admin":
        raise HTTPException(status_code=403,
                            detail="Admin privileges required")
    return current_user

# 使用不同级别的依赖实现权限控制
@app.get("/admin/users")
async def list_all_users(
    admin: User = Depends(get_current_admin_user)
):
    return {"users": get_all_users()}

七、安全最佳实践

密码哈希与存储

密码安全是整个认证体系的基石。使用passlib库配合bcrypt算法是目前Python生态中最推荐的密码哈希方案。bcrypt算法内置了salt机制，且可通过调整rounds参数控制计算成本，有效抵抗暴力破解和彩虹表攻击。切勿使用MD5、SHA-1等快速哈希算法存储密码。

请求速率限制

登录接口、注册接口等公开接口应当实施速率限制（Rate Limiting），防止暴力破解和DDoS攻击。可以使用slowapi等第三方库实现速率限制，也可通过Nginx或云服务商的API网关在更上层实施。常见的策略包括：

按IP限制：同一IP在1分钟内最多发起10次登录请求
按用户限制：同一用户名在5分钟内最多尝试5次登录
全局限流：API整体请求速率限制

CORS配置

跨域资源共享（CORS）配置是Web API安全的重要组成部分。在FastAPI中通过CORSMiddleware进行配置。

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://your-frontend.com"],
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "DELETE"],
    allow_headers=["*"],
)

注意：生产环境中allow_origins应设置为具体的域名列表，而非["*"]。使用通配符将允许任意域名跨域访问你的API，这在包含认证Cookie的场景下尤其危险。

HTTPS要求

所有涉及认证的API必须通过HTTPS提供服务。HTTP传输的Token和密码可能被中间人攻击截获。在生产部署中，建议在反向代理层（如Nginx）终止SSL/TLS，同时配置HSTS（HTTP严格传输安全）头和安全的SSL密码套件。也可以使用uvicorn直接绑定SSL证书：

uvicorn main:app --ssl-keyfile key.pem --ssl-certfile cert.pem

安全的密码存储与验证总结

以下是构建安全认证系统的关键要点：

绝不存储明文密码：始终使用bcrypt或argon2等慢哈希算法
Secret Key管理：使用强随机密钥，通过环境变量或密钥管理服务注入，绝不硬编码在代码中
Token有效期：Access Token设置为短期有效（15-30分钟），配合Refresh Token使用
防止Token泄漏：仅通过HTTPS传输Token，不在URL参数中传递Token
日志安全：不在日志中记录密码明文和Token内容
账户锁定：连续多次登录失败后临时锁定账户，防止暴力破解
定期轮换密钥：定期更换JWT签名密钥，降低密钥泄漏风险

核心总结：FastAPI的依赖注入系统是其最强大的特性之一，它将认证、数据库、配置等横切关注点优雅地整合到统一的依赖管理框架中。结合OAuth2密码流和JWT，你可以构建一套类型安全、可测试性强且符合业界标准的认证体系。记住：安全性不是一次性的配置，而是贯穿整个开发生命周期的持续关注。