如何在FastAPI的非装饰器函数中添加Background Tasks

当然可以，但你提供的代码存在问题——some_function里的background_tasks和email都是未定义的变量，直接运行会抛出NameError。

要在普通函数中使用BackgroundTasks，核心是把路由函数中通过依赖注入拿到的BackgroundTasks实例，以及所需的参数传递给普通函数，具体修改如下：

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()


def write_notification(email: str, message=""):
    with open("log.txt", mode="w") as email_file:
        content = f"notification for {email}: {message}"
        email_file.write(content)

# 修改some_function，接收background_tasks和email作为参数
def some_function(background_tasks: BackgroundTasks, email: str):
    background_tasks.add_task(write_notification, email, message="some notification")

@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
    # 调用时传入实例和参数
    some_function(background_tasks, email)
    return {"message": "Notification sent in the background"}

如果你的业务逻辑涉及多层函数调用，只需要逐层传递BackgroundTasks实例即可。另外，也可以通过FastAPI的Depends机制在普通函数中声明依赖，但对于简单场景，直接传递实例是最直观且高效的方式。

内容的提问来源于stack exchange，提问作者EnergyBoy