如何在 FastAPI 中使用 BackgroundTasks 来异步执行后台任务 

import asyncio
import logging
import timefrom fastapi import FastAPI, BackgroundTasks# 配置日志记录级别
logging.basicConfig(level=logging.INFO)# 初始化FastAPI应用
app = FastAPI(routes=None)# 初始化任务状态字典
task_status = {}def send_mail(n, task_id):"""模拟发送邮件的任务函数。参数:- n: 邮件发送预计耗时的秒数。- task_id: 任务的唯一标识符。"""logging.info(f"开始发送邮件，预计耗时 {n} 秒，任务ID: {task_id}")time.sleep(n)  # 模拟邮件发送耗时logging.info(f"邮件发送完成，耗时 {n} 秒，任务ID: {task_id}")task_status[task_id] = "completed"  # 更新任务状态为完成@app.api_route(path="/index", methods=["GET", "POST"])
async def index(tasks: BackgroundTasks):"""主要的API路由处理函数。参数:- tasks: 用于在后台执行任务的任务管理器。返回:- 一个包含任务ID的字典。"""task_id = str(int(time.time() * 1000))  # 生成一个唯一的任务IDtask_status[task_id] = "running"  # 更新任务状态为运行中tasks.add_task(send_mail, 10, task_id)  # 将邮件发送任务添加到后台任务中print(id(asyncio.get_event_loop()))  # 打印当前事件循环的IDreturn {"index": "index", "task_id": task_id}@app.get("/status/{task_id}")
async def get_task_status(task_id: str):"""获取任务状态的API路由处理函数。参数:- task_id: 任务的唯一标识符。返回:- 一个包含任务ID和状态的字典。"""status = task_status.get(task_id, "not found")  # 获取任务状态，如果不存在则返回"not found"return {"task_id": task_id, "status": status}if __name__ == "__main__":import uvicornimport os# 获取应用模型名称，用于uvicorn运行app_model_name = os.path.basename(__file__).replace(".py", "")print(app_model_name)# 使用uvicorn运行FastAPI应用uvicorn.run(f"{app_model_name}:app", host='0.0.0.0', reload=True)