FastAPI如何通过自定义APIRoute为全部路由统一创建BackgroundTasks后台任务

核心逻辑：通过自定义APIRoute统一创建BackgroundTasks实例挂载到请求上下文，业务层直接调用添加任务，最后统一绑定到响应对象即可，FastAPI会自动执行所有后台任务。

步骤1：修改自定义APIRoute实现 #

# router.py
from fastapi import APIRoute, BackgroundTasks
from fastapi.requests import Request
from fastapi.responses import Response
from typing import Callable

class MyRoute(APIRoute):
    def get_route_handler(self) -> Callable:
        original_route_handler = super().get_route_handler()

        async def custom_route_handler(request: Request) -> Response:
            # 统一创建后台任务实例，挂载到请求上下文
            request.state.background_tasks = BackgroundTasks()
            # 执行原路由业务逻辑
            response: Response = await original_route_handler(request)
            # 把所有后台任务绑定到响应，FastAPI会自动执行
            response.background = request.state.background_tasks
            return response

        return custom_route_handler

步骤2：业务层添加后台任务 #

首先需要给应用全局绑定自定义路由类，所有路由自动复用这套逻辑：

# app.py
from fastapi import FastAPI, Request
from router import MyRoute
from core import users

app = FastAPI()
# 全局应用自定义路由类
app.router.route_class = MyRoute

# 模拟后台任务函数
def write_notification(email: str, message: str):
    print(f"给{email}发送通知：{message}")

@app.post("/send-notification/{email}")
async def send_notification(email: str, request: Request):
    # 路由内直接添加后台任务
    request.state.background_tasks.add_task(write_notification, email, message="请求受理成功通知")
    # 调用其他模块逻辑，传入后台任务实例即可跨模块添加
    users.process(email, request.state.background_tasks)
    return {"message": "操作成功"}

工具模块内的使用示例：

# core/users.py
from fastapi import BackgroundTasks

def process(email: str, background_tasks: BackgroundTasks):
    # 业务处理逻辑
    # 直接调用add_task添加后台任务
    background_tasks.add_task(write_notification, email, message="用户处理完成通知")

补充说明 #

• 该方案无需在每个路由函数的参数列表声明BackgroundTasks，所有后台任务统一在自定义路由层托管，降低重复代码
• 如果你的路由需要返回自定义响应对象，只要是继承自FastAPI的Response类，都支持绑定background属性
• 后台任务会在响应返回给客户端之后串行执行，不会阻塞请求响应流程

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