FastAPI ignores BackgroundTask in custom APIRoute class

Question

I have a function that I wish to run as background task and that I add on the endpoint shown below. Using the return value I poll on a second endpoint for the status which gets updated in the function I pass as background task.

@app.post("/deploy")
async def deploy_vm(data: DeploymentParameters, background_tasks: BackgroundTasks):
    deployment = Deployment.objects(name=data.vm.name).first()
    if deployment and not data.settings.replace:
        return {"msg": "Deployment Task already in progress", "id": deployment.id_}
    deployment = Deployment(name=data.vm.name, payload=data.model_dump_json())
    deployment.save()
    background_tasks.add_task(perform_syncronous_deployment, data, deployment)
    return {
        "msg": "Deployment Task accepted", 
        "id": deployment.id_, 
        "tasks": deployment.tasks
    }

However the function is never called and the task also doesn't show up on the endpoint I use to poll for status:

@app.get("/deploy/{id_}/status")
async def get_deployment_status(id_: str, background_tasks: BackgroundTasks):
    deployment = Deployment.get_status(id_=id_).first()
    for task in background_tasks.tasks:
        logger.info(task.__dict__)
    return deployment.status.to_json()

The original function uses a concurrent.futures.ThreadPoolExecutor to do some things simultaniously but even if I make it fully synchronous it is not called.

The really weird thing is that it works when I use a plain FastAPI in jupyter (the endpoint and the function are identical) - the only difference is that I use routers in the app instead of registering the endpoints with the app directly.

The polling works but the function is not run and I don't see any error messages. When I set a breakpoint in any endpoint and call the function directly, it works as well.

I tried running the app with a debugger and from the cli.

I'd be gratefull for any hints on what else I can try.

Thank you in advance

Answer 1

The issue was caused by a custom logging router. Start background task from custom api router pointed me in the right direction:

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

        async def custom_route_handler(request: Request) -> Response:
            req_body = await request.body()
            log_request(req_body, request.url.path, request.method)

            response = await original_route_handler(request)
            if response.background:
                response.background.add_task(BackgroundTask(log_response, response.body))
            else:
                response.background = BackgroundTask(log_response, response.body)
            return response

        return custom_route_handler

originally it only assigned the log response task (else part of the if/else statement) to background which replaced all other background tasks.