FastAPI端点调用的函数中如何正确使用Background Tasks？

当前在ServerDetails函数内创建BackgroundTasks对象的方式完全不正确，必须在端点函数getServerDetails内创建（或通过FastAPI依赖注入获取）后传入ServerDetails函数。

原因 #

FastAPI的BackgroundTasks是和请求上下文绑定的机制——只有由FastAPI框架管理的BackgroundTasks实例（比如在端点参数中声明获取的实例），框架才会在响应返回给客户端后自动执行其中添加的任务。如果在普通业务函数里自己实例化BackgroundTasks，FastAPI根本不会追踪这个实例，你添加的notification_func永远不会被执行。

正确写法示例 #

# 端点函数中获取BackgroundTasks实例，传入业务函数
@app.post("/abcd", response_model=abcdResponseModel)
async def getServerDetails(IncomingData: serverModel, background_tasks: BackgroundTasks) -> abcdResponseModel:
    """
    Get Server Details
    """
    result = ServerDetails(IncomingData.dict(), background_tasks)
    # 注意：这里需要确保result的结构匹配abcdResponseModel，当前ServerDetails返回Tuple可能不匹配，需调整
    return result

def ServerDetails(IncomingData: dict, background_tasks: BackgroundTasks) -> Tuple[list, int]:
    background_tasks.add(notification_func, arg1, arg2)
    response = some operation.....
    return response, 500

额外提醒 #

你的ServerDetails返回的是Tuple[list, int]，但端点声明的返回类型是abcdResponseModel，这会导致类型不匹配的报错。需要要么调整ServerDetails的返回值结构以符合响应模型，要么修改端点的返回类型注解，确保两者一致。

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