fastapi教程（六）：依赖注入

一，依赖注入概念

依赖注入是将依赖项（例如一个类的实例或一个函数的结果）从类或函数的内部逻辑中解耦出来，并通过外部注入的方式提供给它们。这可以提高代码的模块化和可测试性。

依赖注入常用于以下场景：

• 共享业务逻辑（复用相同的代码逻辑）
• 共享数据库连接
• 实现安全、验证、角色权限
• 等……

在 FastAPI 中，依赖项可以是任何一个被注入到路径操作函数中的函数、类实例或其它对象。
依赖项本身也是一个函数，这个函数可以有它自己的依赖项。FastAPI 会自动处理依赖项的解析和注入。

二，在 FastAPI 中使用依赖注入

下面是一个简单的例子🌰，演示如何在 FastAPI 中使用依赖注入。

import uvicornfrom fastapi import Depends, FastAPIapp = FastAPI()# 定义一个依赖项
def common_parameters(q: str = None, skip: int = 0, limit: int = 10):return {"q": q, "skip": skip, "limit": limit}@app.get("/items/")
def read_items(commons: dict = Depends(common_parameters)):return commonsif '__main__' == __name__:uvicorn.run(app, host='127.0.0.1', port=8088)

在这个例子中：

• common_parameters 函数是一个依赖项，它接受一些查询参数并返回一个包含这些参数的字典。
• read_items 路径操作函数通过 Depends 将 common_parameters 作为依赖项注入。当客户端请求 /items/ 时，FastAPI 会自动调用 common_parameters 函数并将返回值传递给 read_items 函数的 commons 参数。

依赖注入的优点：
- 模块化：将逻辑拆分成小的、独立的部分，每个部分可以单独开发和测试。
- 可测试性：因为依赖项是从外部注入的，所以很容易对其进行 mock 或替换，方便单元测试。
- 可维护性：当依赖项发生变化时，只需要修改依赖项本身，而不需要修改依赖它的所有代码。

依赖注入可以用来处理更复杂的场景，例如数据库连接、认证、缓存等。下面是一个更复杂的例子🌰：

import uvicornfrom fastapi import Depends, FastAPI, HTTPException
from pydantic import BaseModelapp = FastAPI()# 模拟的数据库
fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}]# 定义一个模型
class Item(BaseModel):item_name: str# 定义一个数据库依赖项
def get_db():db = fake_items_dbtry:yield dbfinally:pass# 定义一个依赖项，它依赖于数据库
def get_item(item_name: str, db: list = Depends(get_db)):for item in db:if item["item_name"] == item_name:return itemraise HTTPException(status_code=404, detail="Item not found")# 使用依赖项
@app.get("/items/{item_name}", response_model=Item)
def read_item(item: dict = Depends(get_item)):return itemif '__main__' == __name__:uvicorn.run(app, host='127.0.0.1', port=8088)

在这个例子中，get_db 函数模拟了一个数据库连接，而 get_item 函数依赖于 get_db 函数。路径操作函数 read_item 依赖于 get_item 函数。FastAPI 会自动解析并注入这些依赖项。

三，使用函数实现依赖注入

在 FastAPI 中，依赖注入可以通过函数来实现，这种方式使得我们可以将重复的逻辑封装在一个函数中，并在多个路径操作函数中复用它。

🌰：创建一个简单的博客 API，其中包含获取博客文章的端点，并使用依赖注入来处理数据库连接和用户认证。

1，数据库会话依赖（get_db）:

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

• 这个函数创建一个数据库会话，并在使用后关闭它。
• 使用 yield 语句允许 FastAPI 在请求结束时自动关闭数据库连接。
• 在路由函数中，我们可以通过 db: Session = Depends(get_db) 来注入这个依赖。

2，用户认证依赖（get_current_user）:

def get_current_user(token: str = Depends(oauth2_scheme)):if not token:raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)return {"username": "demo_user"}

• 这个函数模拟了用户认证过程。
• 它依赖于 oauth2_scheme，这是另一个依赖项，用于从请求中提取令牌。
• 在实际应用中，这里应该包含真正的令牌验证逻辑。

3，在路由中使用依赖:

@app.get("/posts/")
def read_posts(db: Session = Depends(get_db), current_user: dict = Depends(get_current_user)):posts = db.query(BlogPost).all()return {"posts": posts, "current_user": current_user["username"]}

• 这个路由函数使用了两个依赖：get_db 和 get_current_user。
• FastAPI 会自动解析这些依赖，并将结果注入到函数参数中。

这个例子展示了 FastAPI 中依赖注入的几个关键特点：

• 可重用性：get_db 和 get_current_user 可以在多个路由中重复使用。
• 关注点分离：数据库连接和用户认证的逻辑与路由处理逻辑分离。
• 依赖链：get_current_user 依赖于 oauth2_scheme，形成了一个依赖链。
• 自动处理：FastAPI 自动处理依赖的解析和注入，简化了代码结构。

四，使用类实现依赖注入

在 FastAPI 中，依赖注入可以通过类来实现。

🌰：创建一个简单的博客 API

1，数据库上下文管理器类（DBContextManager）:

class DBContextManager:def __init__(self):self.db = SessionLocal()def __enter__(self):return self.dbdef __exit__(self, exc_type, exc_val, exc_tb):self.db.close()

• 这个类管理数据库会话的生命周期。
• 它实现了上下文管理器协议（__enter__ 和 __exit__ 方法），允许使用 with 语句自动管理数据库连接。

2，用户认证处理类（AuthHandler）:

class AuthHandler:def __init__(self, token: str = Depends(oauth2_scheme)):self.token = tokendef get_current_user(self):# 认证逻辑

• 这个类封装了用户认证的逻辑。
• 它在初始化时接受一个令牌（通过 Depends(oauth2_scheme) 注入）。
• get_current_user 方法执行实际的用户验证。

3，博客文章仓库类（BlogPostRepository）:

class BlogPostRepository:def __init__(self, db: Session):self.db = dbdef get_all_posts(self) -> List[BlogPost]:return self.db.query(BlogPost).all()def get_post_by_id(self, post_id: int) -> BlogPost:# 获取单个文章的逻辑

• 这个类封装了与博客文章相关的数据库操作。
• 它接受一个数据库会话作为依赖。

4，依赖注入函数:

def get_blog_repository(db_context: DBContextManager = Depends(DBContextManager)) -> BlogPostRepository:with db_context as db:return BlogPostRepository(db)

• 这个函数创建并返回一个 BlogPostRepository 实例。
• 它使用 DBContextManager 来管理数据库会话的生命周期。

5，在路由中使用类依赖:

@app.get("/posts/")
def read_posts(repo: BlogPostRepository = Depends(get_blog_repository),auth: AuthHandler = Depends(AuthHandler)
):current_user = auth.get_current_user()posts = repo.get_all_posts()return {"posts": posts, "current_user": current_user["username"]}

• 这个路由函数使用了两个类依赖：BlogPostRepository 和 AuthHandler。
• FastAPI 自动解析这些依赖，创建实例，并将它们注入到函数参数中。

这个例子展示了 FastAPI 中类依赖注入的几个关键特点：

• 封装性：每个类封装了特定的功能（数据库操作、认证等）。
• 状态管理：类可以维护状态，比如 AuthHandler 维护令牌状态。
• 依赖组合：get_blog_repository 函数展示了如何组合多个依赖。
• 灵活性：类依赖可以轻松扩展，添加新的方法或属性。
• 可测试性：每个组件都可以独立测试，易于模拟。

使用类实现依赖注入特别适合于：

• 需要维护状态的依赖
• 包含多个相关方法的复杂依赖
• 需要在多个地方重用的依赖逻辑

这种方法提供了更好的代码组织和更清晰的职责分离，特别是在大型应用程序中。它使得代码更容易维护、测试和扩展。

五，没有返回值的依赖

在上面的两个例子中，我们都在路径处理函数中使用了依赖注入并获取了其返回值。但有的时候我们只需要执行依赖而不需要使用其返回值。

在 FastAPI 中，当只需要执行依赖而不需要其返回值时，可以使用 dependencies 参数在路径操作装饰器中列出这些依赖项。这种方法通常用于：

• 验证：检查某些条件是否满足，如果不满足则抛出异常。、
• 日志记录：记录每个请求的信息。
• 设置全局状态：如在请求处理前设置一些上下文信息。
• 执行副作用：如更新某些计数器或执行一些清理操作。

from fastapi import FastAPI, Depends, HTTPException, Header
from typing import Optionalapp = FastAPI()# 1. 验证依赖
def verify_api_key(x_api_key: Optional[str] = Header(None)):if x_api_key is None or x_api_key != "valid_api_key":raise HTTPException(status_code=400, detail="Invalid API Key")# 这个依赖不返回任何值# 2. 日志记录依赖
def log_request(user_agent: Optional[str] = Header(None)):print(f"Request received. User-Agent: {user_agent}")# 这个依赖不返回任何值# 3. 设置全局状态依赖（模拟）
def set_global_context():# 假设这里设置了一些全局上下文print("Setting global context for request")# 这个依赖不返回任何值# 使用 dependencies 参数列出不需要返回值的依赖
@app.get("/protected-route/", dependencies=[Depends(verify_api_key), Depends(log_request), Depends(set_global_context)])
async def protected_route():return {"message": "This is a protected route"}# 混合使用返回值的依赖和不返回值的依赖
def get_user_id(x_user_id: Optional[str] = Header(None)):if x_user_id is None:raise HTTPException(status_code=400, detail="User ID is required")return x_user_id@app.get("/user-info/")
async def user_info(user_id: str = Depends(get_user_id),dependencies=[Depends(log_request), Depends(set_global_context)]
):return {"user_id": user_id, "message": "User info retrieved"}# 在应用级别使用依赖
app.add_middleware(Depends(log_request)
)