You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

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

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.07.22 07:30:23