How to run background tasks within a custom APIRoute?

[cmflynn]

First check

• [x ] I added a very descriptive title to this issue.
• [ x] I used the GitHub search to find a similar issue and didn't find it.
• [ x] I searched the FastAPI documentation, with the integrated search.
• [x ] I already searched in Google "How to X in FastAPI" and didn't find any information.
• [ x] I already read and followed all the tutorial in the docs and didn't find an answer.
• [ x] I already checked if it is not related to FastAPI but to Pydantic.
• [x ] I already checked if it is not related to FastAPI but to Swagger UI.
• [x ] I already checked if it is not related to FastAPI but to ReDoc.
• [ x] After submitting this, I commit to one of:
  • Read open issues with questions until I find 2 issues where I can help someone and add a comment to help there.
  • I already hit the "watch" button in this repository to receive notifications and I commit to help at least 2 people that ask questions in the future.
  • Implement a Pull Request for a confirmed bug.

I'm trying to work out how to add background tasks into my custom route handler. I read the part in the docs about using starlette response object when instantiating backround task on my own but am hoping its possible to do it the fastapi native way.

import time
from typing import Callable

from fastapi import APIRouter, FastAPI, Request, Response, BackgroundTasks
from fastapi.routing import APIRoute
from starlette.responses import JSONResponse

def bg_task(msg):
    time.sleep(2)  # simulate blocking behavior
    print(msg)


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

        async def custom_route_handler(request: Request) -> Response:
            bg = BackgroundTasks()
            bg.add_task(print, "pre resp")
            bg.add_task(bg_task, "pre resp sleep")
            before = time.time()
            response: Response = await original_route_handler(request)
            duration = time.time() - before
            response.headers["X-Response-Time"] = str(duration)
            print(f"route duration: {duration}")
            print(f"route response: {response}")
            print(f"route response headers: {response.headers}")
            bg.add_task(print, "post resp")
            bg.add_task(bg_task, "post resp sleep")
            # await bg() ( this works but blocks.. )
            return JSONResponse(response.body.decode(), background=bg) # i would rather return the response directly. trying to avoid building the response myself. 

        return custom_route_handler



app = FastAPI()
router = APIRouter(route_class=TimedRoute)


@app.get("/")
async def not_timed():
    return {"message": "Not timed"}


@router.get("/timed")
async def timed():
    return {"message": "It's the time of my life"}



@app.get("/bg-test", description="Am I Alive?")
async def ping(background_tasks: BackgroundTasks):
    background_tasks.add_task(bg_task, "i wont block")
    return {"ping": "pong"}


app.include_router(router)

[raphaelauv]

I don't get it , write what you would like to have even if the code is not valide

[cmflynn]

I want to be able to use background_tasks as fastapi recommends (path parameter). However in order to use background_tasks without being a path param the docs say to instantiate it yourself via starlette docs(https://www.starlette.io/background/). Here's a code sample of what I would like to happen.

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

        # this will throw "TypeError: custom_route_handler() missing 1 required positional argument: 'background_tasks' "
        async def custom_route_handler(request: Request, background_tasks: BackgroundTasks) -> Response:
            try:
                response: Response = await original_route_handler(request)
            except RequestValidationError as exc:
                body = await request.body()
                detail = {"errors": exc.errors(), "body": body.decode()}
                raise HTTPException(status_code=422, detail=detail)
            background_tasks.add_task(send_message, 'foo bar')
            return response

        return custom_route_handler

I'm specifically tying to avoid having to import startlette.response and returning those responses. I want to be able to return the response that comes out of response: Response = await original_route_handler(request)

[tiangolo]

Why do you need a custom route handler? Wouldn't you be able to do most of this with dependencies or a middleware?

I think it can help if you comment what's your intention, what you are trying to solve, what is the main problem, before thinking about the specific implementation. I think there's a high chance there's a better way to achieve what you need, but I'm not sure yet what you need.

Sorry for the long delay! 🙈 I wanted to personally address each issue/PR and they piled up through time, but now I'm checking each one in order.

[cmflynn]

hey! so I will preface this with I no longer need this solution. but for whats its worth, let me try again. my question was: given a custom api route class such as https://fastapi.tiangolo.com/advanced/custom-request-and-route/#custom-apiroute-class-in-a-router, how would I inject the background_tasks object into the custom_route_handler?

the requirements from what i remember.. sorry its been a while 😬

1. i needed the response object so i could then submit that to a background task to be processed in a non-async way.
2. middleware wasn't an option since at the time middleware could not be applied at the route level. however, looks like this is not the case anymore, and I would probably just use middleware for this now. How to use different middleware for different routes/path #1174 (comment)
3. dependencies didn't work because I need the response object.

[cmflynn]

closing since I think this is resolved via middleware