当前位置：首页 > news >正文

FastAPI 中间件与依赖注入：打造灵活的 API 架构

news 2025/8/6 13:25:38

在 FastAPI 中，Depends 是一个非常重要的概念，它用于依赖注入。依赖注入是一种设计模式，允许你将组件（如函数、类或服务）之间的依赖关系明确地表达出来，而不是硬编码在组件内部。这使得代码更加模块化、可测试和易于维护。

什么是 `Depends`？

Depends 通常用于以下场景：

参数注入：将参数注入到路由处理函数中。
服务注入：将服务或工具类的实例注入到路由处理函数中。
前置验证：在路由处理函数执行之前执行验证逻辑。

如何使用 `Depends`？

Depends 通常与路由处理函数的参数一起使用，如下所示：

from fastapi import Dependsdef get_query(token: str = Depends(your_dependency_function)):return token@app.get("/items/")
async def read_items(query: str = Depends(get_query)):return {"query": query}

在这个例子中，your_dependency_function 是一个依赖项函数，它返回一个值，该值将被注入到 get_query 函数中，然后 get_query 函数的返回值又被注入到路由处理函数 read_items 中。

依赖项函数

依赖项函数可以是任何返回值的异步或同步函数。它们可以执行各种任务，如：

验证请求数据。
从数据库中获取数据。
执行复杂的计算。
调用外部服务。

依赖项的嵌套

你可以将 Depends 用于嵌套依赖项，这意味着一个依赖项函数可以依赖于另一个依赖项函数：

def dependency_a():return "A"def dependency_b(a: str = Depends(dependency_a)):return a + "B"@app.get("/items/")
async def read_items(b: str = Depends(dependency_b)):return {"b": b}

依赖项的并行执行

FastAPI 能够并行执行依赖项函数，这意味着如果多个参数都依赖于相同的依赖项函数，FastAPI 会执行一次该函数，并将结果注入到所有参数中，从而提高效率。

依赖项的错误处理

如果依赖项函数抛出异常，FastAPI 会自动处理这些异常，并返回适当的 HTTP 响应。这使得错误处理更加简洁和集中。

其他实现依赖注入

在 FastAPI 中，Depends 是实现依赖注入的主要方式，但它并不是唯一的方法。以下是一些其他可以实现依赖注入的技术和方法：

直接参数传递：
你可以直接将依赖项作为参数传递给路由处理函数，而不是使用 Depends。

def some_dependency():return "dependency value"@app.get("/items/")
async def read_items(dependency_value: str = some_dependency()):return {"dependency": dependency_value}

使用类：
通过创建类并使用其实例作为依赖项，可以利用面向对象的编程方法来管理依赖关系。

class DependencyInjector:def get_dependency(self):return "dependency value"injector = DependencyInjector()@app.get("/items/")
async def read_items(dependency_value: str = injector.get_dependency()):return {"dependency": dependency_value}

全局状态：
虽然不推荐，但可以通过全局变量或单例模式来存储和访问依赖项。

dependency_value = "dependency value"@app.get("/items/")
async def read_items(dependency_value: str = dependency_value):return {"dependency": dependency_value}

中间件：
在 FastAPI 中，你可以使用中间件来处理请求和响应。中间件可以在请求处理流程中的特定点注入依赖项。

async def inject_dependency(request: Request, call_next):request.state.dependency = "dependency value"response = await call_next(request)return responseapp.add_middleware(Middleware, inject_dependency)

依赖提供者：
类似于 Depends，你可以创建自定义的依赖提供者函数，这些函数可以在内部处理复杂的依赖逻辑。

def custom_dependency_provider():return "custom dependency value"@app.get("/items/")
async def read_items(dependency_value: str = custom_dependency_provider()):return {"dependency": dependency_value}

工厂函数：
你可以定义工厂函数来创建依赖项，这些函数可以在需要时生成新的依赖项实例。

def create_dependency():class Dependency:def __init__(self):self.value = "dependency value"return Dependency()@app.get("/items/")
async def read_items(dependency: Dependency = Depends(create_dependency)):return {"dependency": dependency.value}

数据库会话管理：
对于数据库依赖，你可以使用数据库会话或 ORM 工具的上下文管理器来确保依赖项的正确注入和生命周期管理。

from sqlalchemy.orm import Session@app.get("/items/")
async def read_items(db: Session = Depends(get_db)):return {"items": [item async for item in db.query(Item)]}