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 的内部实现稍微有点复杂,但我们可以简化理解为:

  1. 定义依赖项: 你定义一个函数或类,这个函数或类的返回值就是你的依赖项。
  2. 使用 Depends 你在路由函数的参数中使用 Depends(你的依赖项)
  3. 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 在手,天下我有

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

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

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

发表回复

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