FastAPI如何在控制器作用域外发起BackgroundTask?
问题描述
FastAPI官方文档中介绍了在路由控制器中发起后台任务的方式,示例代码如下:
@app.get("/mypath") async def send_notification(email: str, background_tasks: BackgroundTasks): pass
我想知道是否可以在控制器作用域外发起后台任务(无需在所有函数调用中传递BackgroundTasks对象)。我尝试了以下代码,但没有生效:
from fastapi import BackgroundTasks def print_key(key: str): print(f"test bg task: {key}") def send_stats(key: str): BackgroundTasks().add_task(print_key, key)
解决方案
你手动实例化BackgroundTasks后添加任务不生效,是因为FastAPI的BackgroundTasks必须和当前请求上下文绑定——只有被FastAPI注入到路由函数的实例,才会在请求处理完成后自动执行任务。你自己创建的实例没有关联请求上下文,所以任务不会被触发。
根据你的使用场景,提供两种解决办法:
1. 请求生命周期内(非路由函数)发起任务
如果你的函数仍在请求处理流程中(只是不在路由函数里),可以通过依赖注入获取BackgroundTasks,不用手动传递:
from fastapi import BackgroundTasks, Depends def print_key(key: str): print(f"test bg task: {key}") # 定义依赖获取BackgroundTasks实例 def get_bg_tasks(): return BackgroundTasks() def send_stats(key: str, bg_tasks: BackgroundTasks = Depends(get_bg_tasks)): bg_tasks.add_task(print_key, key) # 路由中调用 @app.get("/trigger") async def trigger_task(key: str, bg_tasks: BackgroundTasks = Depends(get_bg_tasks)): send_stats(key, bg_tasks) return {"status": "task queued"}
这样多层函数调用时,不用手动传递BackgroundTasks,依赖注入会自动关联当前请求的实例,任务会在请求结束后执行。
2. 完全脱离请求上下文发起任务
如果任务不需要和请求挂钩(比如定时任务、独立异步任务),BackgroundTasks就不适用了,因为它依附于请求生命周期。这时可以直接用Python的asyncio创建异步任务,或者用Celery这类专业任务队列:
import asyncio def print_key(key: str): print(f"test bg task: {key}") async def async_print(key: str): print_key(key) def send_stats(key: str): # 启动异步任务 asyncio.create_task(async_print(key))
这种方式的任务独立于请求,直接在后台运行,注意如果在同步函数中使用,需要确保能获取到正确的事件循环。
内容的提问来源于stack exchange,提问作者Carlos Rojas
相关产品推荐
相关产品推荐

