How to use Background Tasks inside a function that is called by a FastAPI endpoint?

Question

I have the below FastAPI endpoint:

@app.post("/abcd", response_model=abcdResponseModel)
async def getServerDetails(IncomingData: serverModel) -> abcdResponseModel:
    """
    Get Server Details
    """
    result = ServerDetails(IncomingData.dict())
    return result

The above is calling the function below

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

Is this the way to do this, or do I need to create a BackgroundTasks() object inside getServerDetails endpoitn and pass it to the ServerDetails function?

Answer 1

BackgroundTasks work if you define a parameter in your endpoint with a type declaration of BackgroundTasks, which will then be added to the returned Response object by FastAPI. Just creating an instance of it inside your custom function or endpoint, e.g., tasks = BackgroundTasks(), it wouldn't work as expected, unless you add tasks to the Response object on your own.

Hence, if you didn't want to add the tasks inside your endpoint, but rather inside a custom function that is called from within your endpoint, you would still need to define the parameter as mentioned above, and then pass it to your custom function.

Also, as per the documentation:

By only using BackgroundTasks (and not BackgroundTask), it's then possible to use it as a path operation function parameter and have FastAPI handle the rest for you, just like when using the Request object directly.

It's still possible to use BackgroundTask alone in FastAPI, but you have to create the object in your code and return a Starlette Response including it.

Example

from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

def send_notification(message):
    print(f'Notification will be sent over. Message: {message}')


def add_some_tasks(background_tasks: BackgroundTasks):
    background_tasks.add_task(send_notification, message="Success")
 
 
@app.get("/{msg}")
def send_msg(msg: str, background_tasks: BackgroundTasks):
    add_some_tasks(background_tasks)
    return {"message": "Request has been successfully submitted."}

Alternative Options

As mentioned in this answer and the documentation excerpt earlier, one could also add the background task(s) directly to the Response object in the following ways. Note that JSONResponse could also be replaced by any other type of class inheriting from Response, such as PlainTextResponse.

Option 1

from fastapi import BackgroundTasks

@app.get("/{msg}")
def send_msg(msg: str):
    content = {"message": "Request has been successfully submitted."}
    tasks = BackgroundTasks()
    tasks.add_task(send_notification, message="Success")
    return JSONResponse(content, background=tasks)

Hence, to perform the above inside your custom function instead, you could follow the example below:

from fastapi import BackgroundTasks

def add_some_tasks():
    tasks = BackgroundTasks()
    tasks.add_task(send_notification, message="Success")
    return tasks
    

@app.get("/{msg}")
def send_msg(msg: str):
    content = {"message": "Request has been successfully submitted."}
    tasks = add_some_tasks()
    return JSONResponse(content, background=tasks)

Option 2

Simply, a variation of the previous option.

from fastapi import BackgroundTasks
from fastapi.responses import JSONResponse

@app.get("/{msg}")
def send_msg(msg: str):
    content = {"message": "Request has been successfully submitted."}
    tasks = BackgroundTasks()
    tasks.add_task(send_notification, message="Success")
    response = JSONResponse(content)
    response.background = tasks
    return response

Again, if you need to add the tasks inside a separate function than the endpoint, please follow the example provided in Option 1.

---

Just for the sake of completeness, to add a single task only, you could alternatively use Starlette's BackgroundTask (this applies to both options provided above):

from starlette.background import BackgroundTask

@app.get("/{msg}")
def send_msg(msg: str):
    # ...
    response.background = BackgroundTask(send_notification, message="Success")
    return response