如何使用`FastAPI`的`依赖`注入`系统`实现`可测试`和`可维护`的`API`。

好的，我们开始。

使用 FastAPI 依赖注入构建可测试和可维护的 API

大家好！今天我们来深入探讨如何利用 FastAPI 的依赖注入系统构建可测试和可维护的 API。依赖注入是一种强大的设计模式，可以显著提高代码的模块化程度、可复用性和可测试性。FastAPI 对此提供了优雅且强大的支持。

什么是依赖注入？

简单来说，依赖注入是一种将组件（类、函数等）的依赖关系从组件自身解耦的设计模式。组件不负责创建或查找其依赖项，而是由外部“注入”这些依赖项。

考虑一个简单的例子：

class DatabaseConnection:
    def connect(self):
        print("Connecting to the database...")
        # ... 实际连接数据库的代码 ...
        return "Connected to database"

class UserService:
    def __init__(self, db_connection: DatabaseConnection):
        self.db_connection = db_connection

    def get_user(self, user_id: int):
        connection_status = self.db_connection.connect()
        print(connection_status)
        # ... 使用数据库连接获取用户信息的代码 ...
        return {"id": user_id, "name": "John Doe"}

在这个例子中，UserService 依赖于 DatabaseConnection。传统的做法是在 UserService 的 __init__ 方法中直接创建 DatabaseConnection 的实例。但是，使用依赖注入，我们将 DatabaseConnection 作为参数传递给 __init__ 方法。这使得我们可以轻松地替换 DatabaseConnection 的实现，例如在测试中使用一个模拟的数据库连接。

FastAPI 中的依赖注入

FastAPI 利用 Python 的类型提示系统和函数签名来实现依赖注入。它允许你声明函数或类所需的依赖项，并自动为你解决这些依赖项。

基本用法：Depends()

FastAPI 提供了 Depends() 函数来声明依赖项。Depends() 接受一个可调用对象（函数、类、方法等）作为参数，该可调用对象被称为依赖项提供者。

from fastapi import FastAPI, Depends

app = FastAPI()

async def get_db():
    db = "Database Connection"  # 模拟数据库连接
    try:
        yield db
    finally:
        print("Closing connection")

@app.get("/items/")
async def read_items(db: str = Depends(get_db)):
    return {"db": db}

在这个例子中：

• get_db 是一个依赖项提供者，它返回一个模拟的数据库连接。
• Depends(get_db) 告诉 FastAPI，read_items 函数需要 get_db 的返回值作为参数。
• FastAPI 会自动调用 get_db，并将返回值作为 db 参数传递给 read_items。
• yield 关键字使得这个dependency可以被正确处理，在请求完成后会自动执行finally语句块，用于关闭数据库连接等。

依赖项的层次结构

依赖项可以形成复杂的层次结构。一个依赖项可以依赖于其他依赖项。

from fastapi import FastAPI, Depends

app = FastAPI()

async def get_query(q: str | None = None):
    return q

async def get_api_key(query: str = Depends(get_query)):
    if query == "special-key":
        return "special-key"
    else:
        return None

@app.get("/items/")
async def read_items(api_key: str | None = Depends(get_api_key)):
    if api_key:
        return {"api_key": api_key}
    else:
        return {"message": "No API key provided"}

在这个例子中，get_api_key 依赖于 get_query。FastAPI 会首先调用 get_query，然后将返回值传递给 get_api_key。

类作为依赖项提供者

类也可以作为依赖项提供者。在这种情况下，FastAPI 会创建类的实例，并将实例作为依赖项注入。

from fastapi import FastAPI, Depends

app = FastAPI()

class DatabaseConnection:
    def __init__(self):
        self.connection = "Database Connection"

    def get_connection(self):
        return self.connection

async def get_db(db_connection: DatabaseConnection = Depends(DatabaseConnection)):
    return db_connection.get_connection()

@app.get("/items/")
async def read_items(db: str = Depends(get_db)):
    return {"db": db}

在这个例子中，DatabaseConnection 类作为依赖项提供者。FastAPI 会创建一个 DatabaseConnection 的实例，并将该实例传递给 get_db 函数。

use_cache 参数

Depends() 函数有一个 use_cache 参数，默认为 True。如果 use_cache 为 True，FastAPI 会缓存依赖项提供者的返回值，并在同一个请求中多次使用相同的依赖项时，只调用一次依赖项提供者。

from fastapi import FastAPI, Depends

app = FastAPI()

counter = 0

async def get_counter():
    global counter
    counter += 1
    return counter

@app.get("/items/")
async def read_items(
    counter1: int = Depends(get_counter, use_cache=True),
    counter2: int = Depends(get_counter, use_cache=True),
):
    return {"counter1": counter1, "counter2": counter2}

在这个例子中，即使 get_counter 被声明了两次，它也只会执行一次。因为 use_cache 设置为 True。

如果 use_cache 设置为 False，则每次使用依赖项时，都会调用依赖项提供者。

依赖注入的优势

1. 可测试性: 通过依赖注入，我们可以轻松地替换真实依赖项为模拟对象（mocks）。这使得我们可以独立地测试每个组件，而无需依赖于其他组件的正确性。

   import pytest
   from fastapi import FastAPI, Depends
   from fastapi.testclient import TestClient
   
   app = FastAPI()
   
   class DatabaseConnection:
       def connect(self):
           return "Real Database Connection"
   
   async def get_db(db_connection: DatabaseConnection = Depends(DatabaseConnection)):
       return db_connection.connect()
   
   @app.get("/items/")
   async def read_items(db: str = Depends(get_db)):
       return {"db": db}
   
   # 测试时使用的模拟数据库连接
   class MockDatabaseConnection:
       def connect(self):
           return "Mock Database Connection"
   
   # 覆盖依赖项
   app.dependency_overrides[DatabaseConnection] = MockDatabaseConnection
   
   client = TestClient(app)
   
   def test_read_items():
       response = client.get("/items/")
       assert response.status_code == 200
       assert response.json() == {"db": "Mock Database Connection"}
   
   #恢复依赖项
   app.dependency_overrides = {}

   在这个例子中，我们使用 dependency_overrides 属性将 DatabaseConnection 替换为 MockDatabaseConnection，从而在测试中使用模拟的数据库连接。测试完成后，需要恢复依赖项。

2. 可维护性: 依赖注入可以降低组件之间的耦合度，使得代码更加模块化和易于维护。当我们修改一个组件时，不需要担心会影响到其他组件。

3. 可复用性: 依赖项可以被多个组件复用，从而减少代码冗余。

4. 灵活性: 依赖注入允许我们在运行时动态地配置依赖项，从而提高应用程序的灵活性。

高级用法

1. 使用 Security 进行安全依赖注入: FastAPI 提供了 Security 类，可以方便地实现安全依赖注入。例如，我们可以使用 Security 来验证用户的身份和权限。

   from fastapi import FastAPI, Depends, HTTPException, status
   from fastapi.security import OAuth2PasswordBearer
   
   app = FastAPI()
   
   oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
   
   async def get_current_user(token: str = Depends(oauth2_scheme)):
       # 模拟用户认证
       if token == "fake_token":
           return {"username": "example_user"}
       else:
           raise HTTPException(
               status_code=status.HTTP_401_UNAUTHORIZED,
               detail="Invalid authentication credentials",
               headers={"WWW-Authenticate": "Bearer"},
           )
   
   @app.get("/users/me")
   async def read_users_me(current_user: dict = Depends(get_current_user)):
       return current_user

   在这个例子中，get_current_user 函数使用 OAuth2PasswordBearer 来验证用户的身份。只有当用户提供了有效的 token 时，才能访问 /users/me 接口。

2. 覆盖路径操作中的依赖项：FastAPI允许你覆盖特定路径操作的依赖关系。

   from fastapi import FastAPI, Depends
   
   app = FastAPI()
   
   async def common_parameters(q: str | None = None, skip: int = 0, limit: int = 100):
       return {"q": q, "skip": skip, "limit": limit}
   
   @app.get("/items/")
   async def read_items(commons: dict = Depends(common_parameters)):
       return commons
   
   @app.get("/users/")
   async def read_users(commons: dict = Depends(common_parameters)):
       return commons
   
   # 覆盖 /items/ 路径的依赖项
   async def override_dependency():
       return {"q": "override", "skip": 0, "limit": 100}
   
   app.dependency_overrides[common_parameters] = override_dependency

   在这个例子中，我们使用 dependency_overrides 属性覆盖了 /items/ 路径的 common_parameters 依赖项。这意味着访问 /items/ 接口时，会使用 override_dependency 函数提供的依赖项，而不是 common_parameters 函数提供的依赖项。/users/ 路径不受影响。

3. 使用 contextvars 处理请求上下文：在某些情况下，你可能需要访问请求的上下文信息，例如请求 ID、用户 ID 等。可以使用 contextvars 模块来实现这一点。

   import contextvars
   from fastapi import FastAPI, Depends, Request
   
   app = FastAPI()
   
   request_id_var = contextvars.ContextVar("request_id")
   
   async def set_request_id(request: Request):
       request_id = request.headers.get("X-Request-ID") or "default_id"
       request_id_var.set(request_id)
   
   async def get_request_id():
       return request_id_var.get()
   
   @app.get("/items/", dependencies=[Depends(set_request_id)])
   async def read_items(request_id: str = Depends(get_request_id)):
       return {"request_id": request_id}

   在这个例子中，我们使用 contextvars 模块创建了一个 request_id_var 上下文变量，用于存储请求 ID。set_request_id 函数从请求头中获取请求 ID，并将其存储到 request_id_var 中。get_request_id 函数从 request_id_var 中获取请求 ID。通过这种方式，我们可以在应用程序的任何地方访问请求 ID。

最佳实践

1. 保持依赖项简单: 依赖项提供者应该尽可能简单，只负责创建和配置依赖项，避免包含业务逻辑。

2. 使用类型提示: 使用类型提示可以提高代码的可读性和可维护性，并帮助 FastAPI 自动解决依赖项。

3. 考虑使用单例模式: 如果某个依赖项的创建代价很高，并且不需要在每个请求中都创建一个新的实例，可以考虑使用单例模式。

4. 避免循环依赖: 循环依赖会导致应用程序无法启动。可以使用延迟依赖或重构代码来避免循环依赖。

5. 编写测试: 编写单元测试和集成测试可以确保依赖注入系统的正确性。

实际案例分析：用户认证和授权

让我们看一个更复杂的例子：用户认证和授权。

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, SecurityScopes
from typing import Annotated

app = FastAPI()

# 定义安全作用域
security_scopes = SecurityScopes(
    scopes={"read": "Read access", "write": "Write access"}
)

# OAuth2 密码流程
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token", scopes=security_scopes.scopes)

# 模拟用户数据库
users = {
    "john": {"username": "john", "scopes": ["read"]},
    "jane": {"username": "jane", "scopes": ["read", "write"]},
}

# 验证用户 token
async def get_current_user(token: str = Depends(oauth2_scheme)):
    username = token  # 简化，token 直接作为 username
    if username not in users:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid authentication credentials",
            headers={"WWW-Authenticate": "Bearer"},
        )
    return users[username]

# 验证用户权限
async def get_current_user_with_scopes(
    security_scopes: SecurityScopes, current_user: dict = Depends(get_current_user)
):
    if not current_user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Not authenticated",
            headers={"WWW-Authenticate": f"Bearer scope='{security_scopes.scope_str}'"},
        )
    for scope in security_scopes.scopes:
        if scope not in current_user["scopes"]:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail="Not enough permissions",
                headers={"WWW-Authenticate": f"Bearer scope='{security_scopes.scope_str}'"},
            )
    return current_user

# 需要 "read" 权限的接口
@app.get("/items/", dependencies=[Depends(security_scopes.use("read"))])
async def read_items(
    current_user: Annotated[dict, Depends(get_current_user_with_scopes)]
):
    return {"items": [], "user": current_user["username"]}

# 需要 "write" 权限的接口
@app.post("/items/", dependencies=[Depends(security_scopes.use("write"))])
async def write_items(
    current_user: Annotated[dict, Depends(get_current_user_with_scopes)]
):
    return {"message": "Items written", "user": current_user["username"]}

在这个例子中：

• 我们定义了两个安全作用域：read 和 write。
• get_current_user 函数验证用户的 token，并返回用户信息。
• get_current_user_with_scopes 函数验证用户是否具有所需的权限。
• /items/ 接口需要 read 权限。
• /items/ (POST) 接口需要 write 权限。

这个例子展示了如何使用 FastAPI 的依赖注入系统来实现用户认证和授权。

使用表格总结依赖注入的核心概念

概念	描述	示例
依赖项	组件（类、函数等）所需的资源或服务。	数据库连接、配置对象、认证服务
依赖项提供者	一个可调用对象（函数、类、方法等），负责创建和配置依赖项。	get_db 函数、DatabaseConnection 类
Depends()	FastAPI 提供的函数，用于声明依赖项。	db: str = Depends(get_db)
use_cache	Depends() 函数的参数，用于控制是否缓存依赖项提供者的返回值。	Depends(get_db, use_cache=True)
dependency_overrides	FastAPI 应用的属性，允许你在测试或其他场景中覆盖依赖项。	app.dependency_overrides[DatabaseConnection] = MockDatabaseConnection
安全依赖注入	使用 Security 类和相关工具，将安全相关的逻辑（例如身份验证、权限验证）注入到 API 接口中。	oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")，Depends(oauth2_scheme)

代码之外的考虑

依赖注入不仅仅是代码技巧，它更是一种设计思想。 理解其背后的SOLID原则，特别是依赖倒置原则，能帮助我们更好地运用它。 好的依赖注入设计应该清晰地表达组件之间的关系，并易于理解和修改。

掌握了依赖注入就掌握了构建高质量 API 的钥匙

FastAPI 的依赖注入系统是一个强大的工具，可以帮助你构建可测试和可维护的 API。通过理解依赖注入的核心概念，并掌握 FastAPI 提供的相关工具，你可以显著提高代码的质量和可维护性。