FastAPI 依赖注入：从入门到实践

# 一、为什么你需要理解依赖注入

代码复用：避免在多个路由中重复编写相同的逻辑（如认证、数据库连接）。
关注点分离：将通用逻辑（如权限检查）与业务逻辑分离。
可测试性：更容易为测试提供模拟依赖。
自动文档：依赖项的参数会自动包含在 API 文档中。

如果你曾在写 API 时重复粘贴数据库连接代码，或是为处理用户认证头痛，那 FastAPI 的依赖注入系统就是你的救星。它像一个“智能组装机”，让你只需声明需要什么，框架会自动帮你准备好一切——从数据库会话到用户权限验证，甚至自动清理资源。

# 二、依赖注入：告别重复代码的魔法

想象你要给 100 个房间装灯：

传统方式：每个房间单独拉线接灯泡（重复写数据库连接、认证逻辑）。
依赖注入：统一设计电路系统，每个房间只需“声明需要灯”，电会自动接通（一次定义依赖，到处复用）。

# 核心概念：什么是依赖注入

依赖注入是一种用于解耦组件并实现代码复用的机制。

依赖注入（Dependency Injection, DI）是一种设计模式，其中对象或函数接收其依赖项（如服务<数据库连接>、工具或资源<用户对象>）作为参数，而不是自己创建它们。在 FastAPI 中，依赖项是通过 依赖函数（Dependency Functions） 定义的，这些函数可以被路径操作函数（如 @app.get）或其他依赖项引用。

在 FastAPI 中，你通过Depends()声明依赖，框架会帮你：

解析依赖关系（比如 A 依赖 B，B 依赖 C，框架自动按顺序准备）。
管理资源生命周期（如用yield自动关闭数据库连接）。
绑定请求数据（如从 Header 获取 Token，从 Query 获取参数）。

# 三、从“手动调用”到“声明式依赖”

# 场景：记录请求日志并计算耗时

# 传统方式（重复代码噩梦）

def log_request():
    start_time = time.time()
    print(f"请求开始: {start_time}")
    return start_time

@app.get("/items/")
async def read_items():
    start_time = log_request()  # 手动调用开始日志
    try:
        return {"items": ["item1", "item2"]}
    finally:
        end_time = time.time()
        print(f"请求结束: {end_time}, 耗时: {end_time - start_time}s")

1
2
3
4
5
6
7
8
9
10
11
12
13

# 依赖注入方式（一次定义，处处复用）

def log_request():
    start_time = time.time()
    print(f"请求开始: {start_time}")
    try:
        yield  # 关键点：用 yield 分割请求前后
    finally:
        end_time = time.time()
        print(f"请求结束: {end_time}, 耗时: {end_time - start_time}s")

@app.get("/items/")
async def read_items(_=Depends(log_request)):  # 声明依赖
    return {"items": ["item1", "item2"]}

1
2
3
4
5
6
7
8
9
10
11
12

关键区别：

yield的魔法：把代码分成“请求前”和“请求后”，框架自动在中间插入你的业务逻辑，无需手动写try-finally。
维护成本：依赖注入只需修改 log_request 函数，所有路由自动生效

# 场景演化 2：增加用户信息

需求变更：需要在日志中包含用户 ID（从 Token 中解析）。

# 直接调用

def log_request(user_id: Optional[str] = None):
    start_time = time.time()
    print(f"请求开始: {start_time}, 用户: {user_id}")
    return start_time

@app.get("/items/")
async def read_items(x_token: str = Header(...)):
    user_id = decode_token(x_token)  # 手动解析 Token
    start_time = log_request(user_id)  # 传递用户信息
    try:
        return {"items": ["item1", "item2"]}
    finally:
        end_time = time.time()
        print(f"请求结束: {end_time}, 耗时: {end_time - start_time}s")

1
2
3
4
5
6
7
8
9
10
11
12
13
14

# 依赖注入

def get_user_id(x_token: str = Header(...)):
    return decode_token(x_token)  # 解析用户 ID

def log_request(user_id: str = Depends(get_user_id)):
    start_time = time.time()
    print(f"请求开始: {start_time}, 用户: {user_id}")
    try:
        yield
    finally:
        end_time = time.time()
        print(f"请求结束: {end_time}, 耗时: {end_time - start_time}s")

@app.get("/items/")
async def read_items(_=Depends(log_request)):  # 声明依赖
    return {"items": ["item1", "item2"]}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

此时的区别：

代码复用：依赖注入方案中，get_user_id 可以被其他依赖复用
参数传递：直接调用需要手动将 user_id 从路由传递到日志函数
关注点分离：依赖注入让路由函数完全不需要关心日志和用户解析逻辑

# 场景演化 3：应用到 100 个路由

假设项目扩展到 100 个路由，其中 50 个需要日志记录。

# 直接调用方案

需要在 50 个路由函数中复制粘贴日志代码
如果日志格式变更（如增加请求路径），需要修改 50 个地方
如果忘记在某个路由中添加日志，排查困难

# 依赖注入

# 定义一次，全局生效
app = FastAPI(dependencies=[Depends(log_request)])  # 所有路由自动记录日志

# 或选择性应用
@app.get("/items/", dependencies=[Depends(log_request)])
async def read_items():
    return {"items": ["item1", "item2"]}

1
2
3
4
5
6
7

此时的区别：

代码行数：依赖注入方案减少约 1000 行重复代码
维护成本：修改只需 1 处，生效于所有路由
扩展性：未来可以轻松添加新的依赖（如性能监控），无需修改原有路由

维度	直接调用	依赖注入
代码复用	低（需复制粘贴）	高（一次定义，到处使用）
生命周期管理	手动管理（复杂）	自动管理（通过 yield）
依赖传递	手动传递（层级深时复杂）	自动解析（框架处理依赖树）
测试隔离	难（需修改调用逻辑）	易（直接替换依赖实现）
关注点分离	差（业务逻辑与日志混合）	好（路由只需关注业务）
错误风险	高（容易漏写清理逻辑）	低（框架保证执行）

什么时候应该直接调用？当满足以下所有条件时，可以考虑直接调用：
逻辑只会在一个地方使用（无复用需求）
无需管理资源生命周期（如打开 / 关闭连接）
不依赖其他组件（如用户认证、配置）
未来修改可能性极低

但在实际项目中，这些条件很少同时满足。特别是在 FastAPI 这种强调高性能、可维护性的框架中，依赖注入几乎总是更好的选择。

思维转换：从「执行动作」到「声明需求」

依赖注入的核心是 声明式编程：

直接调用：我需要记录日志，所以我调用 log_request()
依赖注入：我需要这个路由被日志记录，框架帮我处理具体实现

可以用一个形象的比喻来解释：依赖注入中的 yield 就像是电路中的「断点」，框架负责在这个断点处插入逻辑，形成完整的电路。

让我用代码对比和可视化进一步说明：

# 一、断点插入

# 传统方式（手动拆分）

# 传统方式：每次调用都要手动拆分逻辑
@app.get("/items/")
async def read_items():
    # 上半部分：开始日志
    start_time = time.time()
    print(f"请求开始: {start_time}")
    
    # 中间：业务逻辑
    result = {"items": ["item1", "item2"]}
    
    # 下半部分：结束日志
    end_time = time.time()
    print(f"请求结束: {end_time}, 耗时: {end_time - start_time}s")
    
    return result

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

# 依赖注入（框架组装）

# 依赖函数：定义完整生命周期
def log_request():
    start_time = time.time()
    print(f"请求开始: {start_time}")
    try:
        yield  # 断点：业务逻辑将在这里插入
    finally:
        end_time = time.time()
        print(f"请求结束: {end_time}, 耗时: {end_time - start_time}s")

# 路由：只需声明依赖，无需关心拆分
@app.get("/items/")
async def read_items(_=Depends(log_request)):
    return {"items": ["item1", "item2"]}  # 业务逻辑自动插入到 yield 位置

1
2
3
4
5
6
7
8
9
10
11
12
13
14

# 二、可视化对比

# 传统方式：代码碎片化

路由函数1:
┌───────────────────────────────────────────────────┐
│  日志开始逻辑  │  业务逻辑A  │  日志结束逻辑          │
└───────────────────────────────────────────────────┘

路由函数2:
┌───────────────────────────────────────────────────┐
│  日志开始逻辑  │  业务逻辑B  │  日志结束逻辑          │
└───────────────────────────────────────────────────┘

1
2
3
4
5
6
7
8
9

# 依赖注入：逻辑完整封装

依赖函数:
┌───────────────────────────────────────────────────┐
│  日志开始逻辑  │     yield（业务逻辑插入点）     │  日志结束逻辑  │
└───────────────────────────────────────────────────┘

路由函数1:               路由函数2:
┌─────────────────┐     ┌─────────────────┐
│  声明依赖       │     │  声明依赖       │
│  业务逻辑A      │     │  业务逻辑B      │
└─────────────────┘     └─────────────────┘

1
2
3
4
5
6
7
8
9
10

# 三、进阶理解：`yield` 的本质

在依赖注入中，yield 实现了一种 控制反转（IoC）：

依赖函数 定义了完整的生命周期（开始→业务→结束）
框架负责在 yield 处插入路由函数的业务逻辑
路由函数 只需关注核心业务，无需关心生命周期管理

这就像电影拍摄：

依赖函数 是剧本框架（包括开场和结尾）
路由函数 是主角台词
框架是导演，负责把主角台词插入到剧本的合适位置

# 四、为什么这种设计很重要

# 场景 1：修改日志格式

传统方式：需要修改所有路由中的日志代码
依赖注入：只需修改 log_request 函数

# 场景 2：添加新功能

假设需要在日志中添加请求 ID：

def log_request():
    request_id = uuid.uuid4()  # 生成唯一请求 ID
    print(f"请求开始: {request_id}")
    try:
        yield {"request_id": request_id}  # 将 ID 传递给业务逻辑
    finally:
        print(f"请求结束: {request_id}")

1
2
3
4
5
6
7

路由函数可以通过依赖获取 request_id：

@app.get("/items/")
async def read_items(request_info=Depends(log_request)):
    # request_info 包含 request_id
    return {"items": ["item1", "item2"], "request_id": request_info["request_id"]}

1
2
3
4

# 场景 3：复用日志逻辑到其他系统

依赖函数可以轻松复用：

# 在WebSocket连接中复用相同的日志逻辑
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket, _=Depends(log_request)):
    await websocket.accept()
    await websocket.send_json({"message": "Hello"})
    await websocket.close()

1
2
3
4
5
6

# 五、总结：依赖注入的「组装思维」

依赖注入的核心是 「组装」而非「编写」：

传统编程：像手工打造一把椅子，每个步骤都要重复操作
依赖注入：像乐高积木，定义好标准模块，按需组装

当你习惯这种思维后，会发现代码变得更加：

模块化：每个依赖专注解决单一问题
可测试：可以独立测试依赖函数
可扩展：新增功能只需添加新依赖，无需修改现有代码
声明式：用更少的代码表达更多意图

建议在小项目中刻意练习这种思维，你会逐渐体会到其优雅之处。

# 为什么推荐依赖注入而不能是装饰器

FastAPI 当然可以使用装饰器！不过它的依赖注入系统提供了比传统装饰器更强大、更灵活的解决方案。让我来详细解释两者的区别和适用场景：

# 一、FastAPI 中使用传统装饰器

# 1. 基本示例：记录请求时间

from fastapi import FastAPI
import time

app = FastAPI()

# 定义装饰器
def log_time(func):
    def wrapper(*args, **kwargs):
        start_time = time.time()
        result = func(*args, **kwargs)
        end_time = time.time()
        print(f"函数 {func.__name__} 执行耗时: {end_time - start_time}s")
        return result
    return wrapper

# 在路由中使用装饰器
@app.get("/items/")
@log_time  # 注意：装饰器要放在 @app.get 下面
async def read_items():
    return {"items": ["item1", "item2"]}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

# 2. 装饰器与依赖注入的区别

特性	装饰器	依赖注入
参数访问	需通过 `request` 对象间接获取	直接作为函数参数注入
返回值处理	需要在装饰器中显式返回	自动处理，无需额外代码
异常处理	需要手动捕获和处理	框架自动处理并生成文档
异步支持	需要编写异步装饰器	原生支持同步/异步混合
依赖传递	难以传递复杂依赖	自动解析依赖链
与框架集成	无法利用 FastAPI 的自动文档	参数自动包含在 OpenAPI 中

# 二、装饰器无法解决的问题

# 1. 资源清理（如数据库会话）

装饰器实现方式：

def db_session(func):
    async def wrapper(*args, **kwargs):
        db = SessionLocal()
        try:
            result = await func(db=db, *args, **kwargs)  # 必须显式传递 db
            db.commit()
        except Exception as e:
            db.rollback()
            raise e
        finally:
            db.close()
        return result
    return wrapper

@app.get("/items/")
@db_session
async def read_items(db):  # 必须接受 db 参数
    return db.query(Item).all()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

依赖注入实现方式：

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

@app.get("/items/")
async def read_items(db: Session = Depends(get_db)):  # 自动注入 db
    return db.query(Item).all()

1
2
3
4
5
6
7
8
9
10

# 2. 复杂依赖链

假设需要：Token 验证 → 用户解析 → 权限检查

装饰器实现：

@app.get("/admin/")
@verify_token
@parse_user
@check_admin
async def admin_route(user):
    return {"message": "管理员页面"}

1
2
3
4
5
6

依赖注入实现：

def verify_token(x_token: str = Header(...)): ...
def get_current_user(token = Depends(verify_token)): ...
def check_admin(user = Depends(get_current_user)): ...

@app.get("/admin/", dependencies=[Depends(check_admin)])
async def admin_route():
    return {"message": "管理员页面"}

1
2
3
4
5
6
7

Depends 和 dependencies 的差异对比

Depends 和 dependencies 虽然都用于管理依赖，但它们的定位和使用场景有明确区别：

# 一、核心区别：作用对象不同

特性	`Depends(func)`	`dependencies=[Depends(...)]`
作用目标	注入到路由函数参数	应用于整个路由或APIRouter
返回值处理	依赖函数的返回值会作为参数传递给路由函数	返回值被忽略，仅执行依赖逻辑
使用场景	需要依赖的返回值（如获取数据库会话、用户对象）	不需要返回值（如权限验证、日志记录）
链式调用支持	支持（依赖函数可嵌套依赖其他函数）	支持（列表中多个依赖按顺序执行）

# 二、`Depends`：注入返回值到路由参数

# 场景：获取数据库会话并使用返回值

def get_db():
    db = SessionLocal()
    try:
        yield db  # 返回数据库会话
    finally:
        db.close()

@app.get("/items/")
async def read_items(db: Session = Depends(get_db)):  # 注入返回值到 db 参数
    return db.query(Items).all()  # 直接使用 db

1
2
3
4
5
6
7
8
9
10

# 关键点

依赖函数 get_db 的返回值（db）被注入到路由函数的 db 参数
路由函数可以直接使用这个返回值进行业务操作
支持多层嵌套依赖（如 get_current_user 依赖 get_db）

# 三、`dependencies`：应用依赖到整个路由

# 场景：权限验证（不需要返回值）

def verify_admin(user: User = Depends(get_current_user)):
    if user.role != "admin":
        raise HTTPException(403, detail="需要管理员权限")

@app.get("/admin/", dependencies=[Depends(verify_admin)])  # 应用依赖到路由
async def admin_route():
    return {"message": "管理员页面"}  # 不使用依赖的返回值

1
2
3
4
5
6
7

# 关键点

dependencies 列表中的依赖会在路由函数执行前被调用
依赖的返回值被忽略，仅关注依赖的副作用（如权限验证、日志记录）
适合用于横切关注点（如认证、限流、请求记录）

# 四、链式调用：两者都支持，但方式不同

# 1. `Depends` 的链式调用（嵌套依赖）

依赖函数可以嵌套依赖其他函数，形成依赖链：

def verify_token(x_token: str = Header(...)):
    return decode_token(x_token)  # 验证并返回用户ID

def get_current_user(user_id: int = Depends(verify_token)):
    return db.query(User).get(user_id)  # 根据ID查询用户

@app.get("/profile/")
async def read_profile(user: User = Depends(get_current_user)):  # 最终依赖
    return user.profile  # 使用最内层依赖的返回值

1
2
3
4
5
6
7
8
9

# 2. `dependencies` 的链式调用（列表顺序执行）

def log_request():
    print("请求开始")
    try:
        yield
    finally:
        print("请求结束")

def validate_ip(request: Request):
    if request.client.host not in ALLOWED_IPS:
        raise HTTPException(403)

@app.get("/sensitive/", dependencies=[Depends(log_request), Depends(validate_ip)])
async def sensitive_route():
    return {"data": "敏感信息"}

1
2
3
4
5
6
7
8
9
10
11
12
13
14

# 五、组合使用：同时发挥两者优势

# 依赖1：获取数据库会话（需要返回值）
def get_db(): ...

# 依赖2：验证用户权限（不需要返回值）
def verify_permission(user: User = Depends(get_current_user)):
    if "read:items" not in user.permissions:
        raise HTTPException(403)

# 路由：同时使用 Depends 和 dependencies
@app.get("/items/", dependencies=[Depends(verify_permission)])  # 权限验证
async def read_items(db: Session = Depends(get_db)):  # 获取数据库会话
    return db.query(Items).all()

1
2
3
4
5
6
7
8
9
10
11
12

# 六、何时该用哪个？决策指南

需要依赖的返回值 → 使用 Depends 作为路由参数
- 例如：获取数据库会话、当前用户对象
不需要返回值，只关心副作用 → 使用 dependencies
- 例如：权限验证、请求日志、限流

全局应用依赖 → 使用 FastAPI(dependencies=[]) 或 APIRouter(dependencies=[])

# 所有路由都应用的依赖
app = FastAPI(dependencies=[Depends(log_request)])

# 特定路由组应用的依赖
router = APIRouter(dependencies=[Depends(verify_token)])

1
2
3
4
5

需要在路径操作装饰器中复用依赖 → 使用 dependencies

common_deps = [Depends(verify_token), Depends(validate_ip)]

@app.get("/route1/", dependencies=common_deps)
@app.post("/route2/", dependencies=common_deps)
async def route(): ...

1
2
3
4
5

# 总结：简化而不简单的依赖管理

FastAPI 通过 Depends 和 dependencies 提供了灵活的依赖管理方案：

Depends 是“精细控制”，适合需要返回值并直接参与业务逻辑的场景
dependencies 是“批量应用”，适合无返回值的横切关注点

两者都支持链式调用，但设计目的不同。理解这种差异后，你可以更优雅地组织代码，让依赖注入系统发挥最大威力。

# 三、FastAPI 推荐的「装饰器替代方案」

# 1. 使用依赖注入替代简单装饰器

# 装饰器方式
def validate_token(func):
    def wrapper(*args, **kwargs):
        token = kwargs.get("token")
        if not validate(token):
            raise HTTPException(401)
        return func(*args, **kwargs)
    return wrapper

# 依赖注入方式
def validate_token(token: str = Query(...)):
    if not validate(token):
        raise HTTPException(401)
    return token

@app.get("/secure/")
async def secure_route(token = Depends(validate_token)):
    return {"message": "安全访问"}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

# 2. 使用 `dependencies` 参数替代路由装饰器

# 装饰器方式
@app.get("/items/")
@log_request
@validate_user
async def read_items():
    pass

# 依赖注入方式
@app.get("/items/", dependencies=[Depends(log_request), Depends(validate_user)])
async def read_items():
    pass

1
2
3
4
5
6
7
8
9
10
11

# 四、何时应该使用装饰器

当满足以下条件时，可以考虑使用装饰器：

不涉及资源管理（如无需 try-finally）
不需要访问请求/响应对象
逻辑简单且不需要与 FastAPI 的其他组件（如依赖、中间件）集成
需要兼容传统 Flask/Django 代码

示例：函数结果缓存

def cache_result(func):
    cache = {}
    async def wrapper(*args, **kwargs):
        key = str(args) + str(kwargs)
        if key in cache:
            return cache[key]
        result = await func(*args, **kwargs)
        cache[key] = result
        return result
    return wrapper

@app.get("/data/")
@cache_result
async def get_data():
    return expensive_computation()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

# 五、依赖注入与装饰器的结合使用

实际上，FastAPI 的 Depends 本身就是一种特殊的装饰器模式！你可以结合两者：

# 定义一个可以作为依赖的装饰器
class ValidatePermission:
    def __init__(self, permission: str):
        self.permission = permission
    
    def __call__(self, user: User = Depends(get_current_user)):
        if self.permission not in user.permissions:
            raise HTTPException(403)
        return user

# 在路由中使用
@app.get("/admin/")
async def admin_route(user: User = Depends(ValidatePermission("admin"))):
    return {"message": f"欢迎 {user.name}"}

1
2
3
4
5
6
7
8
9
10
11
12
13
14

# 总结：为什么 FastAPI 更推荐依赖注入

依赖注入相比传统装饰器有以下优势：

生命周期管理：通过 yield 自动处理资源释放
参数注入：直接作为函数参数，无需从 request 中手动提取
依赖链解析：自动处理嵌套依赖，避免多层装饰器嵌套
测试友好：可以轻松替换依赖实现，无需模拟函数调用
文档集成：参数自动包含在 OpenAPI 文档中

如果你来自 Flask/Django 背景，刚开始可能更习惯用装饰器，但尝试依赖注入后，你可能会发现它更符合 FastAPI 的设计哲学，代码也会更加简洁和可维护。

以下是一些应用示例。

# 四、依赖注入的三大“超能力”

# 1. 复杂依赖链自动解析

比如认证流程：验证 Token → 获取用户 → 检查权限。

def verify_token(x_token: str = Header(...)):  # 从Header获取Token
    if x_token != "valid": raise HTTPException(401)
    return x_token

def get_user(token: str = Depends(verify_token)):  # 依赖Token验证
    return db.query(User).filter_by(token=token).first()

def check_admin(user: User = Depends(get_user)):  # 依赖用户对象
    if user.role != "admin": raise HTTPException(403)

@app.get("/admin/")
async def admin_route(_=Depends(check_admin)):  # 只需声明最终依赖
    return {"message": "管理员专属"}

1
2
3
4
5
6
7
8
9
10
11
12
13

框架会自动按顺序执行verify_token → get_user → check_admin，无需手动传递参数。

# 2. 异步与同步无缝混合

FastAPI 支持同步/异步依赖自由混合，比如同时使用同步数据库和异步 Redis：

async def get_redis():  # 异步依赖（Redis连接）
    redis = await aioredis.connect()
    try: yield redis
    finally: await redis.close()

def get_db():  # 同步依赖（数据库会话）
    db = SessionLocal()
    try: yield db
    finally: db.close()

@app.get("/data/")
async def get_data(
    redis=Depends(get_redis),  # 异步依赖
    db=Depends(get_db)        # 同步依赖
):
    return {"redis_data": await redis.get("key"), "db_data": db.query(Item).first()}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

# 3. 测试隔离：一键替换真实依赖

测试时用模拟对象代替真实数据库 API：

def mock_db():  # 模拟数据库
    return Mock()

app.dependency_overrides[get_db] = mock_db  # 替换依赖

1
2
3
4

无需修改路由代码，测试更简单高效。

# 五、对比 Flask/Django：为什么 FastAPI 的依赖注入更简单

如果你熟悉 Flask/Django，可能会觉得依赖注入像它们的“中间件”或“请求钩子”，但 FastAPI 做了三大升级：

局部化控制：可以给单个路由或路由组单独添加依赖，无需全局生效。
参数直接注入：依赖的返回值直接作为函数参数，无需通过全局变量（如 Flask 的g对象）传递。
生命周期管理：yield自动处理资源释放，告别手动关闭连接的烦恼。

# Flask/Django vs FastAPI：对比

场景	Flask（请求钩子）	FastAPI（依赖注入）
记录请求耗时	需在`before_request`和`after_request`分开写	一个依赖函数用`yield`统一处理
传递用户信息	通过全局`g`对象	依赖函数直接返回用户对象并注入
按需应用（仅部分路由）	需在蓝图或路由中单独添加钩子	用`dependencies`参数单独声明

# 将中间件转换为依赖

# Django 中间件
class AuthMiddleware:
    def __call__(self, request):
        user = authenticate(request)
        request.user = user
        return self.get_response(request)

# FastAPI 依赖
def get_current_user(request: Request):
    user = authenticate(request)
    return user

1
2
3
4
5
6
7
8
9
10
11

# 将全局钩子转换为依赖

# Flask 钩子
@app.before_request
def check_auth():
    if not is_authenticated(request):
        abort(401)

# FastAPI 依赖
def verify_auth(request: Request):
    if not is_authenticated(request):
        raise HTTPException(401)

1
2
3
4
5
6
7
8
9
10

# 利用依赖注入简化代码

#Flask 代码
@app.route("/items/")
def get_items():
    db = get_db()  # 手动获取数据库连接
    user = get_current_user()  # 手动获取用户
    items = db.query(Items).filter(owner=user.id).all()
    return jsonify([item.to_dict() for item in items])


#FastAPI 代码
@app.get("/items/")
async def get_items(db: Session = Depends(get_db), 
                  user: User = Depends(get_current_user)):
    items = db.query(Items).filter(owner=user.id).all()
    return items

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

# 六、初学者如何上手？从三个小练习开始

练习 1：最简单的依赖——记录请求路径

from fastapi import Depends, FastAPI

app = FastAPI()

def log_path():
    print(f"访问路径：{request.url.path}")  # 注意：需从函数参数获取request

@app.get("/hello/")
async def hello(request: Request, _=Depends(log_path)):  # 显式传入request
    return {"message": "Hello"}

1
2
3
4
5
6
7
8
9
10

练习 2：管理数据库会话——用yield自动关闭连接

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

engine = create_engine("sqlite:///test.db")
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

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

@app.get("/items/")
async def read_items(db=Depends(get_db)):
    return db.query(Item).all()

1
2
3
4
5
6
7
8
9
10
11
12
13
14

练习 3：异步依赖——获取 Redis 数据

import aioredis

async def get_redis():
    redis = await aioredis.from_url("redis://localhost")
    try: yield redis
    finally: await redis.close()

@app.get("/cache/")
async def get_cache(redis=Depends(get_redis)):
    return await redis.get("key")

1
2
3
4
5
6
7
8
9
10

# 七、总结：依赖注入让开发更“懒”却更高效

FastAPI 的依赖注入不是让你写更多代码，而是让你“少写重复代码”：

声明式编程：只需说“我需要数据库会话”，不用关心怎么创建和关闭。
智能组装：框架自动处理依赖链，像拼乐高一样组合功能。
开箱即用：兼容异步/同步，无缝集成参数校验、文档生成和测试。

如果你曾为重复代码头疼，或是想让 API 更健壮易维护，依赖注入就是 FastAPI 送给你的“代码简化神器”。从现在开始，试试用Depends()代替手动调用，让框架帮你处理那些“麻烦事”吧！

编辑

#FastAPI #依赖注入

上次更新: 2025-07-04, 09:33:47

← FastAPI 快速上手指南 Python 全栈之路系列之基础篇→

FastAPI 依赖注入：从入门到实践

# 一、为什么你需要理解依赖注入

# 二、依赖注入：告别重复代码的魔法

# 核心概念：什么是依赖注入

# 三、从“手动调用”到“声明式依赖”

# 场景：记录请求日志并计算耗时

# 传统方式（重复代码噩梦）

# 依赖注入方式（一次定义，处处复用）

# 场景演化 2：增加用户信息

# 直接调用

# 依赖注入

# 场景演化 3：应用到 100 个路由

# 直接调用方案

# 依赖注入

# 一、断点插入

# 传统方式（手动拆分）

# 依赖注入（框架组装）

# 二、可视化对比

# 传统方式：代码碎片化

# 依赖注入：逻辑完整封装

# 三、进阶理解：yield 的本质

# 四、为什么这种设计很重要

# 场景 1：修改日志格式

# 场景 2：添加新功能

# 场景 3：复用日志逻辑到其他系统

# 五、总结：依赖注入的「组装思维」

# 为什么推荐依赖注入而不能是装饰器

# 一、FastAPI 中使用传统装饰器

# 1. 基本示例：记录请求时间

# 2. 装饰器与依赖注入的区别

# 二、装饰器无法解决的问题

# 1. 资源清理（如数据库会话）

# 2. 复杂依赖链

# 一、核心区别：作用对象不同

# 二、Depends：注入返回值到路由参数

# 场景：获取数据库会话并使用返回值

# 关键点

# 三、dependencies：应用依赖到整个路由

# 场景：权限验证（不需要返回值）

# 关键点

# 四、链式调用：两者都支持，但方式不同

# 1. Depends 的链式调用（嵌套依赖）

# 2. dependencies 的链式调用（列表顺序执行）

# 五、组合使用：同时发挥两者优势

# 六、何时该用哪个？决策指南

# 总结：简化而不简单的依赖管理

# 三、FastAPI 推荐的「装饰器替代方案」

# 1. 使用依赖注入替代简单装饰器

# 2. 使用 dependencies 参数替代路由装饰器

# 四、何时应该使用装饰器

# 五、依赖注入与装饰器的结合使用

# 总结：为什么 FastAPI 更推荐依赖注入

# 四、依赖注入的三大“超能力”

# 1. 复杂依赖链自动解析

# 2. 异步与同步无缝混合

# 3. 测试隔离：一键替换真实依赖

# 五、对比 Flask/Django：为什么 FastAPI 的依赖注入更简单

# Flask/Django vs FastAPI：对比

# 将中间件转换为依赖

# 将全局钩子转换为依赖

# 利用依赖注入简化代码

# 六、初学者如何上手？从三个小练习开始

# 七、总结：依赖注入让开发更“懒”却更高效

# 三、进阶理解：`yield` 的本质

# 二、`Depends`：注入返回值到路由参数

# 三、`dependencies`：应用依赖到整个路由

# 1. `Depends` 的链式调用（嵌套依赖）

# 2. `dependencies` 的链式调用（列表顺序执行）

# 2. 使用 `dependencies` 参数替代路由装饰器