Python高级技术之：`FastAPI`的依赖注入系统：`Depends`的内部实现与应用。 - 智猿学院-前后端，数据库，人工智能，云计算等领域前沿技术讲座

各位观众老爷，早上/下午/晚上好！今天咱们来聊聊 FastAPI 里的依赖注入系统，特别是那个神秘又强大的 Depends。

开场白：别怕，依赖注入没那么玄乎

很多小伙伴一听到“依赖注入”就觉得高深莫测，好像要学完十年编程才能掌握。其实，它本质上就是一种组织代码的方式，让你的代码更清晰、更易于测试和维护。 FastAPI 的依赖注入系统更是简单易用，理解了它的内部机制，你就能像驾驭自行车一样轻松驾驭它。

正文：Depends 的前世今生

首先，我们要明白，Depends 本身不是一个类，而是一个函数。它的作用就像一个“钩子”，告诉 FastAPI 在处理某个路由时，先执行指定的依赖项，然后将依赖项的结果注入到路由函数中。

from fastapi import FastAPI, Depends

app = FastAPI()

# 定义一个依赖项
async def get_db():
    db = "模拟数据库连接" # 实际应用中会建立数据库连接
    try:
        yield db
    finally:
        # 关闭数据库连接 (可选)
        print("关闭数据库连接")

# 使用 Depends 注入依赖项
@app.get("/items/")
async def read_items(db: str = Depends(get_db)):
    return {"db": db, "items": ["item1", "item2"]}

在这个例子中，get_db 就是一个依赖项。当访问 /items/ 路由时，FastAPI 会先执行 get_db 函数，然后将 get_db 函数的返回值 (也就是 "模拟数据库连接" 这个字符串) 注入到 read_items 函数的 db 参数中。

Depends 的内部实现：扒开它的衣服看看

Depends 的内部实现稍微有点复杂，但我们可以简化理解为：

定义依赖项： 你定义一个函数或类，这个函数或类的返回值就是你的依赖项。
使用 Depends： 你在路由函数的参数中使用 Depends(你的依赖项)。
FastAPI 的魔法： FastAPI 会自动检测到 Depends，然后执行你的依赖项，并将结果传递给路由函数。

更深入一点，Depends 的核心在于 FastAPI 内部的依赖解析器。这个解析器会分析你的路由函数，找到所有的 Depends，然后按照一定的顺序执行这些依赖项。

其实 Depends 本身的代码并不多。它主要做的事情是存储了依赖项函数的引用，并提供一些元数据，供 FastAPI 的依赖解析器使用。

# 简化版的 Depends 源码 (仅用于演示)
class Depends:
    def __init__(self, dependency, use_cache=True):
        self.dependency = dependency
        self.use_cache = use_cache

    def __repr__(self) -> str:
        dependency = getattr(self.dependency, "__name__", repr(self.dependency))
        cache_info = "" if self.use_cache else ", use_cache=False"
        return f"Depends({dependency}{cache_info})"

这个简化版的 Depends 类，主要存储了依赖项函数 (dependency) 和是否使用缓存 (use_cache) 的信息。真正的依赖解析逻辑在 FastAPI 内部，而不是 Depends 本身。

Depends 的应用场景：十八般武艺样样精通

Depends 可以应用于各种场景，比如：

数据库连接： 就像上面的例子，你可以用 Depends 来获取数据库连接。
用户认证： 你可以用 Depends 来验证用户的身份。
权限控制： 你可以用 Depends 来检查用户是否有权限访问某个资源。
配置加载： 你可以用 Depends 来加载配置文件。
缓存： 你可以用 Depends 来实现缓存逻辑。

下面我们来看一些具体的例子：

1. 用户认证

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer

app = FastAPI()

# 模拟用户数据
users = {
    "john": {"username": "john", "password": "password123"},
    "jane": {"username": "jane", "password": "password456"},
}

# OAuth2 密码模式
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

# 验证用户
async def get_current_user(token: str = Depends(oauth2_scheme)):
    username = None  # 实际应用中，会从token中提取用户信息
    for user in users.values():
        if user["password"] == token: # 简化验证逻辑，实际应用中用hash校验
            username = user["username"]
            break

    if not username:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid authentication credentials",
            headers={"WWW-Authenticate": "Bearer"},
        )
    return users[username]

# 需要认证的路由
@app.get("/users/me")
async def read_users_me(current_user: dict = Depends(get_current_user)):
    return current_user

在这个例子中，get_current_user 函数使用 OAuth2PasswordBearer 来获取 token，然后验证用户的身份。如果验证成功，就返回用户信息；否则，就抛出一个异常。

2. 权限控制

from fastapi import FastAPI, Depends, HTTPException, status

app = FastAPI()

# 模拟用户权限
user_permissions = {
    "john": ["read", "write"],
    "jane": ["read"],
}

# 检查权限
async def check_permission(permission: str, current_user: dict = Depends(get_current_user)): # 依赖于上面的get_current_user
    if permission not in user_permissions.get(current_user["username"], []):
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Insufficient permissions",
        )
    return True

# 需要权限的路由
@app.get("/items/{item_id}")
async def read_item(item_id: int, has_read_permission: bool = Depends(lambda: check_permission("read"))):
    return {"item_id": item_id}

在这个例子中，check_permission 函数检查用户是否有指定的权限。如果没有权限，就抛出一个异常。注意这里使用 lambda 表达式创建一个匿名函数，并将权限字符串传递给 check_permission 函数。

3. 缓存

from fastapi import FastAPI, Depends
import time

app = FastAPI()

# 模拟缓存
cache = {}

# 获取数据 (带缓存)
async def get_data_with_cache(key: str):
    if key in cache:
        print("从缓存中获取数据")
        return cache[key]
    else:
        print("从数据库中获取数据")
        # 模拟从数据库中获取数据
        time.sleep(1)  # 模拟耗时操作
        data = f"Data for {key}"
        cache[key] = data
        return data

# 缓存依赖项
async def cached_data(key: str, data: str = Depends(get_data_with_cache)):
    return data

# 使用缓存的路由
@app.get("/data/{key}")
async def read_data(key: str, data: str = Depends(lambda k=key: get_data_with_cache(k))):
    return {"data": data}

在这个例子中，get_data_with_cache 函数首先检查缓存中是否有数据。如果有，就直接返回缓存中的数据；否则，就从数据库中获取数据，然后将数据放入缓存中。

Depends 的高级用法：更上一层楼

类作为依赖项： 你可以使用类作为依赖项，这样可以更好地组织你的代码。

from fastapi import FastAPI, Depends

app = FastAPI()

class DatabaseConnection:
    def __init__(self):
        self.connection = "模拟数据库连接"

    def __enter__(self):
        return self.connection

    def __exit__(self, exc_type, exc_val, exc_tb):
        print("关闭数据库连接")

    def get_connection(self):
        return self.connection

    def __call__(self): # 使得类可以像函数一样调用
        return self.get_connection()

# 使用类作为依赖项
@app.get("/items/")
async def read_items(db: str = Depends(DatabaseConnection())):
    return {"db": db, "items": ["item1", "item2"]}

@app.get("/items2/")
async def read_items2(db: str = Depends(DatabaseConnection)): # 直接使用类，每次请求都会创建一个新的实例
    return {"db": db.get_connection(), "items": ["item1", "item2"]}

依赖项之间的依赖： 一个依赖项可以依赖于另一个依赖项，形成一个依赖链。

from fastapi import FastAPI, Depends

app = FastAPI()

# 第一个依赖项
async def get_settings():
    return {"setting1": "value1", "setting2": "value2"}

# 第二个依赖项，依赖于第一个依赖项
async def get_database(settings: dict = Depends(get_settings)):
    return {"database_url": "...", "settings": settings}

# 使用依赖链的路由
@app.get("/items/")
async def read_items(db: dict = Depends(get_database)):
    return {"db": db, "items": ["item1", "item2"]}

use_cache=False： 默认情况下，Depends 会缓存依赖项的结果。如果你不希望缓存，可以将 use_cache 设置为 False。

from fastapi import FastAPI, Depends

app = FastAPI()

# 不使用缓存的依赖项
async def get_random_number():
    import random
    return random.randint(1, 100)

# 使用 Depends(..., use_cache=False)
@app.get("/random")
async def read_random(number: int = Depends(get_random_number, use_cache=False)):
    return {"number": number}

yield 关键字： 如果你的依赖项需要在使用完毕后进行清理工作 (比如关闭数据库连接)，可以使用 yield 关键字。

from fastapi import FastAPI, Depends

app = FastAPI()

# 使用 yield 的依赖项
async def get_resource():
    resource = "分配资源"
    try:
        yield resource
    finally:
        print("释放资源")

# 使用 Depends 注入依赖项
@app.get("/items/")
async def read_items(resource: str = Depends(get_resource)):
    return {"resource": resource, "items": ["item1", "item2"]}

Depends 的注意事项：避免踩坑

循环依赖： 避免循环依赖，否则会导致程序崩溃。
类型提示： 使用类型提示可以帮助 FastAPI 更好地理解你的代码，并提供更好的错误提示。
性能： 过度使用依赖注入可能会影响性能。在性能敏感的场景中，需要仔细评估。
测试： 依赖注入可以使你的代码更易于测试。你可以使用 mock 对象来模拟依赖项，从而隔离被测试的代码。

总结：Depends 在手，天下我有

FastAPI 的 Depends 系统是一个非常强大的工具，可以帮助你编写更清晰、更易于测试和维护的代码。掌握了 Depends 的内部机制和应用场景，你就可以在 FastAPI 的世界里自由驰骋，编写出高质量的应用程序。

特性	描述
依赖注入	一种设计模式，允许你将依赖项传递给函数或类，而不是在函数或类内部创建它们。
`Depends` 函数	FastAPI 提供的一个函数，用于声明依赖项。
缓存	默认情况下，`Depends` 会缓存依赖项的结果。可以使用 `use_cache=False` 来禁用缓存。
`yield` 关键字	可以使用 `yield` 关键字来在依赖项中使用完毕后进行清理工作。
应用场景	数据库连接、用户认证、权限控制、配置加载、缓存等等。
注意事项	避免循环依赖、使用类型提示、注意性能问题、方便测试。

好了，今天的讲座就到这里。希望大家能够喜欢。下次再见！别忘了点赞、评论、转发哦！

发表回复 取消回复

发表回复取消回复