You can define background tasks to be run after returning a response.

This is useful for operations that need to happen after a request, but that the client doesn't really have to be waiting for the operation to complete before receiving the response.

This includes, for example:

  • Email notifications sent after performing an action:
    • As connecting to an email server and sending an email tends to be "slow" (several seconds), you can return the response right away and send the email notification in the background.
  • Processing data:
    • For example, let's say you receive a file that must go through a slow process, you can return a response of "Accepted" (HTTP 202) and process it in the background.

Using BackgroundTasks

First, import BackgroundTasks and define a parameter in your path operation function with a type declaration of BackgroundTasks:

 

from fastapi import BackgroundTasks, FastAPI

FastAPI will create the object of type BackgroundTasks for you and pass it as that parameter.

Create a task function

Create a function to be run as the background task.

It is just a standard function that can receive parameters.

It can be an async def or normal def function, FastAPI will know how to handle it correctly.

In this case, the task function will write to a file (simulating sending an email).

And as the write operation doesn't use async and await, we define the function with normal def:

def write_notification(email: str, message=""):
    with open("log.txt", mode="w") as email_file:
        content = f"notification for {email}: {message}"
        email_file.write(content)

Add the background task

Inside of your path operation function, pass your task function to the background tasks object with the method .add_task():

@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(write_notification, email, message="some notification")
    return {"message": "Notification sent in the background"}

.add_task() receives as arguments:

  • A task function to be run in the background (write_notification).
  • Any sequence of arguments that should be passed to the task function in order (email).
  • Any keyword arguments that should be passed to the task function (message="some notification").

Full example

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()


def write_notification(email: str, message=""):
    with open("log.txt", mode="w") as email_file:
        content = f"notification for {email}: {message}"
        email_file.write(content)


@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(write_notification, email, message="some notification")
    return {"message": "Notification sent in the background"}

Dependency Injection

Using BackgroundTasks also works with the dependency injection system, you can declare a parameter of type BackgroundTasks at multiple levels: in a path operation function, in a dependency (dependable), in a sub-dependency, etc.

FastAPI knows what to do in each case and how to re-use the same object, so that all the background tasks are merged together and are run in the background afterwards:

=== "Python 3.6 and above"

from typing import Union

from fastapi import BackgroundTasks, Depends, FastAPI

app = FastAPI()


def write_log(message: str):
    with open("log.txt", mode="a") as log:
        log.write(message)


def get_query(background_tasks: BackgroundTasks, q: Union[str, None] = None):
    if q:
        message = f"found query: {q}\n"
        background_tasks.add_task(write_log, message)
    return q


@app.post("/send-notification/{email}")
async def send_notification(
    email: str, background_tasks: BackgroundTasks, q: str = Depends(get_query)
):
    message = f"message to {email}\n"
    background_tasks.add_task(write_log, message)
    return {"message": "Message sent"}

=== "Python 3.10 and above"

from fastapi import BackgroundTasks, Depends, FastAPI

app = FastAPI()


def write_log(message: str):
    with open("log.txt", mode="a") as log:
        log.write(message)


def get_query(background_tasks: BackgroundTasks, q: str | None = None):
    if q:
        message = f"found query: {q}\n"
        background_tasks.add_task(write_log, message)
    return q


@app.post("/send-notification/{email}")
async def send_notification(
    email: str, background_tasks: BackgroundTasks, q: str = Depends(get_query)
):
    message = f"message to {email}\n"
    background_tasks.add_task(write_log, message)
    return {"message": "Message sent"}

In this example, the messages will be written to the log.txt file after the response is sent.

If there was a query in the request, it will be written to the log in a background task.

And then another background task generated at the path operation function will write a message using the email path parameter.

Related articles

Google adverts for django-cms

install pip install djangocms-ads Provides plugins to add ad slots to the <head> tag and then a plugin to create the advert element in the page content. Plugin templates can be extended using the DJANGOCMS_ADS_TEMPLATES setting

API Generator

Project description API Project Generator Simple API Structure Generator using tecnologies: FastAPI SQLAlchemy aiohttp aiomysql or asyncpg

FastAPI Keycloak

Installation pip install fastapi_keycloak Quickstart In order to just get started, we prepared some containers and configs for you. 1. Configure the Containers

fastapi jwt

FastAPI native extension, easy and simple JWT auth

Sanic GraphQL

install  pip install Sanic-GraphQL Usage Use the GraphQLView view fromsanic_graphql from sanic_graphql import GraphQLView from sanic import Sanic

sanic crud

sanic_crud is a REST API framework for creating a CRUD (Create/Retrieve/Update/Delete) API using Sanic and PeeWee You can use sanic_crud to automatically create an API from your PeeWee models

django admin registration

models.py Tests for various ways of registering models with the admin site. from django.db import models

django admin filters

models.py from django.contrib.auth.models import User from django.contrib.contenttypes.fields import GenericForeignKey, GenericRelation from django.contrib.contenttypes.models import ContentType from django.db import models

django admin custom urls

models.py from functools import update_wrapper from django.contrib import admin from django.db import models from django.http import HttpResponseRedirect from django.urls import reverse