FastAPI 后台任务和依赖注入:使用后台任务进行异步处理

FastAPI 是一个现代、快速（高性能）的 Web 框架，用于基于标准 Python 类型提示使用 Python 3.7+ 构建 API。它的一个强大功能是能够处理后台任务和依赖注入，这使您可以从主请求线程中“卸载”耗时的任务，从而提高应用程序的性能和响应能力。

在这篇博文中，我们将探讨如何在 FastAPI 中使用后台任务，利用其依赖注入系统进行异步处理。我们将通过一些演示来说明这些概念的实际应用。

FastAPI 中的后台任务允许您在主请求-响应周期之外执行任务。这对于耗时或可以延迟的操作特别有用，例如发送电子邮件、处理数据或与第三方服务交互。

设置 FastAPI

在深入研究后台任务之前，让我们先设置一个基本的 FastAPI 项目。确保您已安装 FastAPI 和 Uvicorn：

pip install fastapi uvicorn

创建一个新文件 main.py 并添加以下代码：

from fastapi import FastAPI


app = FastAPI()


@app.get("/")
def read_root():
    return {"message": "Hello World"}

使用 Uvicorn 运行应用程序：

uvicorn main:app --reload

当您导航到 http://127.0.0.1:8000 时，您应该会看到消息 Hello World。

使用后台任务

要在 FastAPI 中使用后台任务，您需要从 fastapi 导入 BackgroundTasks 并定义一个将在后台执行的函数。让我们看一个简单的例子：

from fastapi import BackgroundTasks, FastAPI


app = FastAPI()


def write_log(message: str):
    with open("log.txt", "a") as log_file:
        log_file.write(f"{message}\n")


@app.post("/log")
def log_message(message: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(write_log, message)
    return {"message": "Log will be written in the background"}

在此示例中，当对 /log 发出 POST 请求时，write_log 函数将安排在后台运行，将提供的消息写入日志文件。

使用后台任务进行依赖注入

FastAPI 的依赖注入系统允许您有效地管理应用程序中的依赖项。您可以使用此系统将依赖项注入到后台任务中。方法如下：

from fastapi import BackgroundTasks, Depends, FastAPI


app = FastAPI()


def get_db():
    # Simulate a database connection
    return {"db_connection": "Connected"}


def write_log(message: str, db: dict):
    with open("log.txt", "a") as log_file:
        log_file.write(f"{message} - DB: {db['db_connection']}\n")


@app.post("/log")
def log_message(message: str, background_tasks: BackgroundTasks, db: dict = Depends(get_db)):
    background_tasks.add_task(write_log, message, db)
    return {"message": "Log will be written in the background"}

在本例中，get_db 函数模拟数据库连接，并使用 FastAPI 的 Depends 将其注入到 write_log 函数中。

演示：在后台发送电子邮件

让我们创建一个更实际的示例，在后台发送电子邮件。我们将使用 smtplib 模块发送电子邮件。

首先，安装必要的依赖项：

pip install python-dotenv

创建一个 .env 文件来存储您的电子邮件凭据：

EMAIL_HOST=smtp.example.com
EMAIL_PORT=587
EMAIL_USER=your-email@example.com
EMAIL_PASS=your-email-password

然后，使用以下代码更新您的 main.py：

import os
import smtplib
from email.mime.text import MIMEText
from fastapi import BackgroundTasks, Depends, FastAPI
from pydantic import BaseModel
from dotenv import load_dotenv


load_dotenv()


app = FastAPI()


class EmailSchema(BaseModel):
    email: str
    subject: str
    message: str


def get_email_client():
    host = os.getenv("EMAIL_HOST")
    port = os.getenv("EMAIL_PORT")
    user = os.getenv("EMAIL_USER")
    password = os.getenv("EMAIL_PASS")
    client = smtplib.SMTP(host, port)
    client.starttls()
    client.login(user, password)
    return client


def send_email(email: EmailSchema, client: smtplib.SMTP):
    msg = MIMEText(email.message)
    msg["Subject"] = email.subject
    msg["From"] = os.getenv("EMAIL_USER")
    msg["To"] = email.email
    client.sendmail(os.getenv("EMAIL_USER"), [email.email], msg.as_string())
    client.quit()


@app.post("/send-email")
def send_email_endpoint(email: EmailSchema, background_tasks: BackgroundTasks, client: smtplib.SMTP = Depends(get_email_client)):
    background_tasks.add_task(send_email, email, client)
    return {"message": "Email will be sent in the background"}

在此示例中，send_email 函数计划在后台运行，发送使用 get_email_client 依赖项提供的 SMTP 客户端发送电子邮件。

还有一些额外的演示，演示了如何在各种场景中使用 FastAPI 中的后台任务和依赖注入：

演示：在后台处理上传的文件

在此演示中，我们将在后台处理上传的文件。我们将通过将文件的内容写入日志来模拟处理。

from fastapi import BackgroundTasks, FastAPI, File, UploadFile


app = FastAPI()


def process_file(file_content: str):
    with open("file_log.txt", "a") as log_file:
        log_file.write(f"Processed content: {file_content}\n")


@app.post("/uploadfile/")
async def upload_file(file: UploadFile, background_tasks: BackgroundTasks):
    content = await file.read()
    background_tasks.add_task(process_file, content.decode("utf-8"))
    return {"message": "File processing will be done in the background"}

在此示例中，当文件上传到 /uploadfile/ 时，process_file 函数会读取其内容并安排在后台进行处理。

演示：在后台生成报告

在此演示中，我们将在后台生成报告。通过创建包含部分内容的文本文件来模拟报告生成。

import time
from fastapi import BackgroundTasks, FastAPI


app = FastAPI()


def generate_report(report_id: str):
    time.sleep(5)  # Simulate a time-consuming report generation
    with open(f"report_{report_id}.txt", "w") as report_file:
        report_file.write(f"Report {report_id} content")


@app.post("/generate-report/{report_id}")
def generate_report_endpoint(report_id: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(generate_report, report_id)
    return {"message": "Report generation will be done in the background"}

在此示例中，当对 /generate-report/{report_id} 发出 POST 请求时，generate_report 函数将安排在后台运行，从而创建报告文件。

演示：在后台刷新 API 令牌

在此演示中，我们将在后台刷新 API 令牌。这对于需要保持令牌最新且不阻塞主请求流的应用程序非常有用。

import requests
from fastapi import BackgroundTasks, Depends, FastAPI


app = FastAPI()


def get_api_config():
    # Simulate fetching API configuration
    return {
        "token_url": "https://example.com/token",
        "client_id": "your-client-id",
        "client_secret": "your-client-secret"
    }


def refresh_api_token(config: dict):
    response = requests.post(config["token_url"], data={
        "client_id": config["client_id"],
        "client_secret": config["client_secret"],
        "grant_type": "client_credentials"
    })
    token_data = response.json()
    with open("token.txt", "w") as token_file:
        token_file.write(token_data["access_token"])


@app.post("/refresh-token")
def refresh_token_endpoint(background_tasks: BackgroundTasks, config: dict = Depends(get_api_config)):
    background_tasks.add_task(refresh_api_token, config)
    return {"message": "API token refresh will be done in the background"}

在此示例中，当对 /refresh-token 发出 POST 请求时，refresh_api_token 函数将安排在后台运行，获取新令牌并将其保存到文件中。

这些演示说明了 FastAPI 中后台任务的不同用例，重点介绍了如何“卸载”各种类型的任务以提高应用程序的响应能力。通过利用 FastAPI 的依赖注入系统，您可以有效地管理依赖项并确保您的后台任务具有正确执行所需的上下文和资源。

请随意调整这些示例以满足您的特定要求，并探索 FastAPI 应用程序中后台任务的其他潜在用例。

FastAPI 中的后台任务和依赖项注入是构建高效且响应迅速的 Web 应用程序的强大工具。通过将耗时的任务转移到后台并有效地管理依赖项，您可以显著提高应用程序的性能。

在这篇博文中，我们介绍了 FastAPI 中后台任务的基础知识、如何将依赖项注入与后台任务结合使用，并提供了在后台发送电子邮件的实际示例。借助这些工具，您可以使用 FastAPI 构建更强大、更可扩展的应用程序。

写在最后：

1. 正如文章开头，后台任务的引入，“卸载”长时间运行的操作，确保主请求的响应周期，提升了用户体验
2. 异步处理任务，某种程度上提高了系统资源的利用率，减少系统的空闲时间也提高了整体吞吐量，即平衡了系统的负载
3. 异步处理也使应用程序更容易扩展，因为它可以分布在多个工作器或者实例上
4. 凡事都有利弊，看你怎么去平衡，过多的使用后台任务，会增加程序的复杂性，比如，需要管理后台任务的生命周期、任务失败或者重试，同时debug也比较麻烦