FastAPI 教程

FastAPI 依赖注入

FastAPI 的依赖注入(Dependency Injection, DI) 是其最强大、最核心的特性之一。它允许你声明代码运行所需的“依赖”(如数据库连接、当前用户、配置信息等),FastAPI 会自动为你解析、执行并注入这些依赖。

这种设计不仅让代码更简洁、可复用,还极大地简化了单元测试。

以下是 FastAPI 依赖注入的核心概念和常用用法:


1. 基础用法:使用 Depends

依赖通常是一个普通的 Python 函数(或类)。你可以使用 Depends 将其注入到路径操作函数中。

纯文本
plaintext
from fastapi import FastAPI, Depends

app = FastAPI()

# 1. 定义一个依赖函数
def common_parameters(q: str | None = None, skip: int = 0, limit: int = 100):
    return {"q": q, "skip": skip, "limit": limit}

# 2. 在路径操作中使用 Depends 注入
@app.get("/items/")
async def read_items(commons: dict = Depends(common_parameters)):
    return commons

FastAPI 会自动调用 common_parameters,解析其中的查询参数,并将返回值传递给 commons


2. 现代推荐写法:使用 Annotated (FastAPI 0.95.0+)

官方目前强烈推荐使用 typing.Annotated,它可以让类型提示和依赖声明分离,代码更清晰,且兼容更多静态类型检查工具。

纯文本
plaintext
from typing import Annotated
from fastapi import FastAPI, Depends

app = FastAPI()

def get_current_user(token: str):
    # 模拟验证 token 并返回用户
    return {"username": "admin", "token": token}

# 使用 Annotated 声明依赖
@app.get("/users/me")
async def read_users_me(
    current_user: Annotated[dict, Depends(get_current_user)]
):
    return current_user

3. 类作为依赖

依赖不仅可以是函数,也可以是类。FastAPI 会实例化该类,并将其作为依赖项注入。

纯文本
plaintext
from typing import Annotated
from fastapi import FastAPI, Depends

app = FastAPI()

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()]):
    # 注意:这里 Depends() 里面可以为空,FastAPI 会根据类型提示自动推断
    return {"q": commons.q, "skip": commons.skip, "limit": commons.limit}

4. 带有 yield 的依赖(处理清理工作)

这是 FastAPI 依赖注入最实用的场景之一。如果你的依赖需要在请求结束后执行清理操作(如关闭数据库连接、释放文件句柄),可以使用 yield 代替 return

纯文本
plaintext
from typing import Annotated
from fastapi import FastAPI, Depends

app = FastAPI()

# 模拟数据库会话
class FakeDBSession:
    def __init__(self):
        print("🔗 数据库连接已建立")

    def close(self):
        print("🔌 数据库连接已关闭")

def get_db():
    db = FakeDBSession()
    try:
        yield db  # 将 db 注入给路径操作函数
    finally:
        db.close()  # 请求结束后,无论是否发生异常,都会执行这里

@app.get("/items/")
async def read_items(db: Annotated[FakeDBSession, Depends(get_db)]):
    return {"status": "使用数据库查询数据"}

运行结果:先打印“建立连接”,处理请求,最后打印“关闭连接”。


5. 依赖缓存 (Dependency Caching)

在同一个请求中,如果你多次声明了同一个依赖,FastAPI 默认只会执行一次,并缓存其结果。这可以显著节省资源(例如避免多次查询数据库)。

纯文本
plaintext
def get_query(q: str | None = None):
    print("执行了 get_query") # 这个函数在一次请求中只会打印一次
    return {"q": q}

@app.get("/items/")
async def read_items(
    q1: Annotated[dict, Depends(get_query)],
    q2: Annotated[dict, Depends(get_query)] # 不会再次执行 get_query
):
    return {"q1": q1, "q2": q2}

如果你希望每次都重新执行,可以设置 Depends(get_query, use_cache=False)


6. 子依赖 (Sub-dependencies)

依赖可以拥有自己的依赖。FastAPI 会自动构建一个依赖图(Dependency Graph),并按正确的顺序解析它们。

纯文本
plaintext
def verify_token(x_token: str):
    if x_token != "fake-super-secret-token":
        raise HTTPException(status_code=400, detail="X-Token header invalid")
    return x_token

def verify_key(x_key: str):
    if x_key != "fake-super-secret-key":
        raise HTTPException(status_code=400, detail="X-Key header invalid")
    return x_key

# get_header 依赖于 verify_token 和 verify_key
def get_header(
    x_token: Annotated[str, Depends(verify_token)],
    x_key: Annotated[str, Depends(verify_key)]
):
    return {"x_token": x_token, "x_key": x_key}

@app.get("/items/")
async def read_items(header: Annotated[dict, Depends(get_header)]):
    return header

7. 覆盖依赖 (Dependency Overrides) – 测试神器

在编写单元测试时,你通常不想连接真实的数据库或调用真实的外部 API。FastAPI 提供了 app.dependency_overrides,允许你用 Mock 函数替换真实的依赖。

纯文本
plaintext
from fastapi.testclient import TestClient

# 真实的依赖
def get_db():
    return "Real Database Connection"

@app.get("/data")
async def get_data(db: Annotated[str, Depends(get_db)]):
    return {"db": db}

# --- 测试代码 ---
client = TestClient(app)

def override_get_db():
    return "Mock Database Connection"

# 覆盖依赖
app.dependency_overrides[get_db] = override_get_db

def test_read_data():
    response = client.get("/data")
    assert response.json() == {"db": "Mock Database Connection"}

# 测试结束后,建议清除覆盖
app.dependency_overrides.clear()

常见应用场景总结

  1. 数据库会话管理:使用 yield 获取和关闭 SQLAlchemy/SQLModel 会话。
  2. 身份验证与授权:解析 JWT Token,获取当前登录用户 (get_current_user)。
  3. 速率限制 (Rate Limiting):检查 IP 或用户的请求频率。
  4. 共享业务逻辑:将复杂的参数校验或数据预处理提取为独立的依赖函数。

如果你有具体的场景(比如:如何结合 SQLAlchemy 写依赖,或者如何做 JWT 认证),可以告诉我,我可以提供更针对性的代码示例!

发表回复

您的邮箱地址不会被公开。 必填项已用 * 标注