揭秘FastAPI后台任务：轻松实现高效异步工作流，解锁高效编程新境界

FastAPI 是一个现代、快速（高性能）的 Web 框架，用于构建 API，与 Starlette 和 Pydantic 集成。它具有异步功能，使得处理大量并发请求成为可能。本文将深入探讨 FastAPI 的后台任务功能，帮助开发者轻松实现高效异步工作流。

后台任务简介

后台任务（Background Tasks）是 FastAPI 中的一个重要特性，允许你在服务器端异步执行长时间运行的任务，而不会阻塞主线程。这些任务可以是在服务器启动时执行的一次性任务，也可以是周期性执行的任务。

安装和配置

要使用后台任务，首先需要确保你的 FastAPI 应用已经安装了 uvicorn 和 fastapi。接下来，你可以使用以下代码来配置后台任务：

from fastapi import FastAPI from fastapi.background import BackgroundTasks app = FastAPI() @app.on_event("startup") async def startup_event(): print("Application started!") @app.on_event("shutdown") async def shutdown_event(): print("Application shutdown!") @app.post("/tasks") async def create_background_task(background_tasks: BackgroundTasks): background_tasks.add_task(long_running_task) return {"message": "Background task added!"} def long_running_task(): print("This is a long-running task!")

在上面的代码中，我们创建了一个简单的 FastAPI 应用，并定义了一个 /tasks 路由来添加后台任务。BackgroundTasks 对象允许我们添加任务，而 long_running_task 函数是一个长时间运行的任务示例。

后台任务类型

FastAPI 支持以下两种后台任务类型：

一次性任务：在服务器启动时执行一次，或者通过 API 调用添加。
周期性任务：定期执行，例如每天或每小时执行一次。

一次性任务

以下是一个添加一次性任务的示例：

from fastapi import FastAPI from fastapi.background import BackgroundTasks app = FastAPI() @app.on_event("startup") async def startup_event(): print("Application started!") @app.post("/tasks") async def create_background_task(background_tasks: BackgroundTasks): background_tasks.add_task(long_running_task) return {"message": "Background task added!"} def long_running_task(): print("This is a long-running task!")

在上面的代码中，我们通过 add_task 方法将 long_running_task 添加到后台任务队列中。

周期性任务

以下是一个添加周期性任务的示例：

from fastapi import FastAPI from fastapi.background import BackgroundTasks from fastapi.responses import JSONResponse from apscheduler.schedulers.background import BackgroundScheduler app = FastAPI() scheduler = BackgroundScheduler() @app.on_event("startup") async def startup_event(): scheduler.add_job(long_running_task, 'interval', hours=1) scheduler.start() print("Scheduler started!") @app.on_event("shutdown") async def shutdown_event(): scheduler.shutdown() def long_running_task(): print("This is a long-running task!")

在上面的代码中，我们使用 apscheduler 库来添加一个周期性任务，该任务每小时执行一次。