FastAPI 后台任务和依赖注入:使用依赖项创建可重用组件

FastAPI 是一个现代、快速（高性能）的 Web 框架，用于基于标准 Python 类型提示使用 Python 3.7+ 构建 API。FastAPI 的强大功能之一是其依赖注入系统，它允许您创建可重用的组件，这些组件可以轻松管理并注入到您的路径操作函数中。在这篇博文中，我们将探讨如何使用 FastAPI 的依赖注入系统创建具有依赖项的可重用组件。

什么是依赖注入？

依赖注入 (DI) 是一种设计模式，允许对象从外部源接收其依赖项，而不是自己创建它们。这促进了关注点的彻底分离，使您的代码更加模块化且更易于测试。

在 FastAPI 中，依赖项可以注入到您的路径操作函数、后台任务和其他组件中。FastAPI 使用 Python 的类型提示来声明和管理这些依赖项。

创建可重用的组件

可重用的组件是可以定义一次并在应用程序的多个部分中使用的功能。通过利用 FastAPI 的依赖注入系统，您可以以干净、模块化且易于测试的方式创建这些组件。

示例：数据库依赖项

让我们从可重用组件的简单示例开始：数据库会话依赖项。我们将定义一个提供数据库会话的依赖项，并在多个路径操作函数中使用它。

首先，定义一个提供数据库会话的函数：

from sqlalchemy.orm import Session
from fastapi import Depends
from database import SessionLocal  # Assume we have a SessionLocal class


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

此函数将创建一个新的数据库会话，产生它，并在请求完成后关闭它。

接下来，在路径操作函数中使用此依赖项：

from fastapi import FastAPI
from models import Item  # Assume we have an Item model


app = FastAPI()


@app.get("/items/")
def read_items(db: Session = Depends(get_db)):
    items = db.query(Item).all()
    return items

在此示例中，get_db 函数是提供数据库会话的依赖项。Depends 类用于在 read_items 函数中声明依赖项。FastAPI 负责调用 get_db 函数并将结果传递给 read_items 函数。

示例：当前用户依赖项

让我们通过添加提供当前用户的依赖项来扩展我们的示例。这对于需要用户身份验证的路径操作很有用。

首先，定义一个检索当前用户的函数：

from fastapi import HTTPException, Security
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.orm import Session
from models import User  # Assume we have a User model


oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")


def get_current_user(token: str = Security(oauth2_scheme), db: Session = Depends(get_db)):
    user = db.query(User).filter(User.token == token).first()
    if not user:
        raise HTTPException(status_code=401, detail="Invalid authentication credentials")
    return user

此函数使用提供的 OAuth2 令牌从数据库中提取用户。它使用 get_db 依赖项来访问数据库。

接下来，在路径操作函数中使用此依赖项：

@app.get("/users/me")
def read_users_me(current_user: User = Depends(get_current_user)):
    return current_user

在此示例中，get_current_user 依赖项用于 read_users_me 函数中检索当前用户。此设置可确保用户得到正确的身份验证并从数据库中检索。

组合依赖项

还可以组合依赖项以创建更复杂的可重用组件。让我们创建一个依赖于数据库会话和当前用户的报告生成服务。

首先，定义一个用于生成报告的服务类：

class ReportService:
    def __init__(self, db: Session, user: User):
        self.db = db
        self.user = user


    def generate_report(self):
        # Simulate report generation
        report = f"Report for user {self.user.username}"
        return report

接下来，为报告服务创建一个依赖函数：

def get_report_service(db: Session = Depends(get_db), user: User = Depends(get_current_user)):
    return ReportService(db, user)

该函数结合 get_db 和 get_current_user 依赖关系，创建一个 ReportService 实例。

最后，在路径操作函数中使用报告服务：

@app.get("/users/me/report")
def get_report(report_service: ReportService = Depends(get_report_service)):
    report = report_service.generate_report()
    return {"report": report}

在此示例中，get_report_service 依赖项用于 get_report 函数中，为当前用户生成报告。此设置演示了如何通过组合更简单的依赖项来创建复杂的可重用组件。

测试可重用组件

使用依赖项注入的优点之一是它使您的代码更易于测试。您可以模拟依赖项并单独测试路径操作函数。

示例：使用 Mock 依赖项进行测试

让我们使用 Mock 依赖项为 get_report 路径操作函数编写一个测试：

from fastapi.testclient import TestClient
from unittest.mock import Mock
import pytest


app = FastAPI()


# Mock dependencies
@pytest.fixture
def mock_db():
    return Mock()


@pytest.fixture
def mock_user():
    return Mock(username="testuser")


@pytest.fixture
def mock_report_service(mock_db, mock_user):
    service = ReportService(mock_db, mock_user)
    service.generate_report = Mock(return_value="Mock Report")
    return service


# Override dependencies in the test client
app.dependency_overrides[get_db] = lambda: mock_db
app.dependency_overrides[get_current_user] = lambda: mock_user
app.dependency_overrides[get_report_service] = lambda: mock_report_service


client = TestClient(app)


def test_get_report():
    response = client.get("/users/me/report")
    assert response.status_code == 200
    assert response.json() == {"report": "Mock Report"}

在此测试中，我们使用 unittest.mock 库创建模拟依赖项并覆盖测试客户端中的真实依赖项。这使我们能够单独测试 get_report 函数。

接下来，让我们探索更多关于在 FastAPI 中创建具有依赖项的可重用组件的演示。这些示例将重点介绍不同的用例，包括日志记录、配置管理和外部 API 客户端。

示例：日志记录依赖项

日志记录是任何应用程序的重要组成部分。您可以创建可重用的日志记录依赖项来管理应用程序的日志记录配置。

定义日志依赖项

首先，定义一个提供记录器的函数：

import logging
from fastapi import FastAPI, Depends


def get_logger():
    logger = logging.getLogger("myapp")
    logger.setLevel(logging.DEBUG)
    if not logger.handlers:
        handler = logging.StreamHandler()
        formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
        handler.setFormatter(formatter)
        logger.addHandler(handler)
    return logger

使用日志依赖项

接下来，在路径操作函数中使用此依赖项：

app = FastAPI()


@app.get("/log-example/")
def log_example(logger: logging.Logger = Depends(get_logger)):
    logger.info("This is a log message")
    return {"message": "Check your logs for the message"}

在此示例中，get_logger 函数用于提供记录器实例。log_example 函数记录一条消息并返回响应。

示例：配置管理依赖项

配置管理对于处理不同的环境（开发、测试、生产）至关重要。您可以创建可重复使用的配置依赖项。

定义配置依赖项

首先，创建一个配置类和一个提供该类的函数：

from pydantic import BaseSettings


class Settings(BaseSettings):
    app_name: str = "My FastAPI App"
    admin_email: str
    items_per_user: int = 50


    class Config:
        env_file = ".env"


def get_settings():
    return Settings()

使用配置依赖项

接下来，在路径操作函数中使用此依赖项：

@app.get("/config-example/")
def config_example(settings: Settings = Depends(get_settings)):
    return {
        "app_name": settings.app_name,
        "admin_email": settings.admin_email,
        "items_per_user": settings.items_per_user,
    }

在此示例中，get_settings 函数提供应用程序设置。config_example 函数返回配置值。

示例：外部 API 客户端依赖项

如果您的应用程序与外部 API 交互，您可以创建可重复使用的客户端依赖项来管理 API 请求。

定义外部 API 客户端依赖项

首先，创建一个客户端类和一个提供该类的函数：

import requests


class ExternalAPIClient:
    def __init__(self, base_url: str):
        self.base_url = base_url


    def get_data(self, endpoint: str):
        response = requests.get(f"{self.base_url}/{endpoint}")
        response.raise_for_status()
        return response.json()


def get_external_api_client():
    return ExternalAPIClient(base_url="https://api.example.com")

使用外部 API 客户端依赖项

接下来，在路径操作函数中使用此依赖项：

@app.get("/external-api-example/")
def external_api_example(client: ExternalAPIClient = Depends(get_external_api_client)):
    data = client.get_data("some-endpoint")
    return data

在此示例中，get_external_api_client 函数提供了一个实例ExternalAPIClient。external_api_example 函数使用此客户端从外部 API 获取数据。

示例：缓存依赖项

缓存可以显著提高应用程序的性能。您可以创建可重用的缓存依赖项。

定义缓存依赖项

首先，创建一个简单的内存缓存和一个提供它的函数：

from cachetools import LRUCache, cached


cache = LRUCache(maxsize=100)


def get_cache():
    return cache

使用缓存依赖项

接下来，在路径操作函数中使用此依赖项：

@app.get("/cache-example/")
@cached(cache=get_cache)
def cache_example(cache: LRUCache = Depends(get_cache)):
    # Simulate a slow computation
    result = sum(range(1000000))
    return {"result": result}

在此示例中，get_cache 函数提供 LRU 缓存。 cache_example 函数使用 cached 装饰器来缓存缓慢计算的结果。

示例：用于测试的依赖项覆盖

使用依赖项注入进行测试更加简单，因为您可以覆盖依赖项。

定义依赖项

让我们定义一个返回问候消息的简单依赖项：

def get_greeting():
    return "Hello, world!"

在路径操作中使用依赖项

接下来，在路径操作函数中使用此依赖项：

@app.get("/greeting/")
def greeting(greeting: str = Depends(get_greeting)):
    return {"message": greeting}

覆盖用于测试的依赖项

在测试中覆盖依赖项：

from fastapi.testclient import TestClient


client = TestClient(app)


def override_get_greeting():
    return "Hello, test!"


app.dependency_overrides[get_greeting] = override_get_greeting


def test_greeting():
    response = client.get("/greeting/")
    assert response.status_code == 200
    assert response.json() == {"message": "Hello, test!"}

在此示例中，override_get_greeting 函数提供了特定于测试的问候消息。test_greeting 函数验证覆盖的依赖项是否在问候路径操作中使用。

FastAPI 的依赖项注入系统允许您创建干净、模块化且易于测试的可重用组件。通过定义和使用用于日志记录、配置管理、外部 API 客户端、缓存等的依赖项，您可以提高代码库的可维护性和效率。此外，覆盖依赖项的能力简化了测试，确保您的应用程序保持稳健和可靠。

我们希望这些演示能让您更深入地了解如何在 FastAPI 中创建和使用具有依赖项的可重用组件。FastAPI 的依赖注入系统提供了一种强大的方法来创建干净、模块化且易于测试的可重用组件。通过利用依赖项，您可以编写更易于维护的代码并确保应用程序的组件正确解耦。

写在最后：

1. 至此，小编关于FastAPI后台任务和依赖注入就介绍完毕了。我们分别从概念，实际应用和意义几个方面全面地介绍了这两块内容，希望您能有所启发。下期我们将共同探讨如何测试FastAPI，使用什么样子的技术，如何提高程序质量等等话题，敬请期待