Sometimes backgroundtasks are skipped

[varun-seth]

If there are 2 tasks added in the background in one request, and first one has an error, neither the error is logged nor is the second task run.

Also, if there's an exception in the request after adding the task (like if I raise HTTPException with statuscode=2**) then also the background tasks are skipped. I have tested both of these using fastapi.

How to ensure the background tasks are always run?

[varun-seth]

To solve my case, I created a custom class to manage background tasks, now it is executed more reliably using dependency inject and finally clause.

@api.get("/demo")
def get_demo(background_tasks: PostponedTasks = Depends(get_postponed_tasks, use_cache=False)):
    def task1():
        raise ValueError("Task 1 failed")

    def task2():
        print("Task 2 runs fine")

    background_tasks.add_task(task1)
    background_tasks.add_task(task2)

    return {"status": "ok"}

You can see in the logs that task2 runs fine. This works even if an error is raised in the request.

INFO:     Application startup complete.
DEBUG:    Adding background task: task1
DEBUG:    Adding background task: task2
DEBUG:    Running background task: task1
ERROR:    Task 1 failed
Traceback (most recent call last):
  File "...postponed_tasks.py", line 32, in process
    task(*self.args_list[i], **self.kwargs_list[i])
  File "...main.py", line 160, in task1
    raise ValueError("Task 1 failed")
ValueError: Task 1 failed
DEBUG:    Running background task: task2
Task 2 runs fine
INFO:     127.0.0.1:54728 - "GET /demo HTTP/1.1" 200

Following is the basic code.

from logging import getLogger
from typing import Callable

logger = getLogger(__name__)


class PostponedTasks:
    """
    This runs more reliably as compared to FastAPI's BackgroundTasks
    - It will show proper error messages in the logs
    - It can execute even if the main request raises an exception
    """

    def __init__(self, proceed_on_exception: bool):
        self.proceed_on_exception = proceed_on_exception

        self.tasks: list[Callable] = []
        self.args_list = []
        self.kwargs_list = []

    def add_task(self, func: Callable, *args, **kwargs):
        logger.debug(f"Adding background task: {func.__name__}")
        self.tasks.append(func)
        self.args_list.append(args)
        self.kwargs_list.append(kwargs)

    def process(self):
        error_occurred = False
        for i, task in enumerate(self.tasks):
            try:
                logger.debug(f"Running background task: {task.__name__}")
                task(*self.args_list[i], **self.kwargs_list[i])
            except Exception as e:
                error_occurred = True
                logger.exception(e)

            if error_occurred and not self.proceed_on_exception:
                break

        return not error_occurred


def get_postponed_tasks(proceed_on_exception: bool = True):
    postponed_tasks = PostponedTasks(proceed_on_exception=proceed_on_exception)

    try:
        yield postponed_tasks
    finally:
        postponed_tasks.process()

Edit: The request is not returned until the postponed tasks are all completed. This is not a good approach.

[adriangb]

Can you make a self container reproducible example and submit it to FastAPI?

[varun-seth]

Here's a basic example. The errors are properly logging when middleware is not added. Kindly comment/uncomment the middleware decorator line to see the effect. I will post the same in fastapi discussion.

from fastapi import BackgroundTasks, FastAPI, Request
import time

app = FastAPI()

@app.get("/")
def get_demo(background_tasks: BackgroundTasks):
    def task1():
        raise ValueError("Task 1 failed")

    def task2():
        print("Task 2 runs fine")

    background_tasks.add_task(task1)
    background_tasks.add_task(task2)

    return {"status": "ok"}


@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    return response

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app)

[varun-seth]

fastapi/fastapi#11828