Background

This is the reference for the BackgroundTask and BackgroundTasks objects where it contains all API informationhow to use it.

Like Starlette and any other Starlette based frameworks, in Esmerald you can define background tasks to run after the returning response.

This can be useful for those operations that need to happen after the request without blocking the client (the client doesn't have to wait to complete) from receiving that same response.

Example:

1. Registering a user in the system and send an email confirming the registration.
2. Processing a file that can take "some time". Simply return a HTTP 202 and process the file in the background.

esmerald.BackgroundTask

BackgroundTask(func, *args, **kwargs)

Bases: BackgroundTask

BackgroundTask as a single instance can be easily achieved.

Example

from pydantic import BaseModel

from esmerald import BackgroundTask, JSONResponse, post


class UserIn(BaseModel):
    email: str
    password: str


async def send_email_notification(message: str):
    '''
    Sends an email notification
    '''
    send_notification(message)


@post(
    "/register",
    background=BackgroundTask(send_email_notification, message="Account created"),
)
async def create_user(data: UserIn) -> JSONResponse:
    JSONResponse({"message": "Created"})

PARAMETER	DESCRIPTION
func	TYPE: Callable[P, Any]
*args	TYPE: args DEFAULT: ()
**kwargs	TYPE: kwargs DEFAULT: {}

Source code in esmerald/backgound.py

def __init__(self, func: Callable[P, Any], *args: P.args, **kwargs: P.kwargs) -> None:
    super().__init__(func, *args, **kwargs)

func instance-attribute

func = func

args instance-attribute

args = args

kwargs instance-attribute

kwargs = kwargs

is_async instance-attribute

is_async = is_async_callable(func)

esmerald.BackgroundTasks

BackgroundTasks(tasks=None)

Bases: BackgroundTasks

Alternatively, the BackgroundTasks can also be used to be passed in.

Example

from datetime import datetime

from pydantic import BaseModel

from esmerald import BackgroundTask, BackgroundTasks, JSONResponse, post


class UserIn(BaseModel):
    email: str
    password: str


async def send_email_notification(message: str):
    '''
    Sends an email notification
    '''
    send_notification(message)


def write_in_file():
    with open("log.txt", mode="w") as log:
        now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        content = f"Notification sent @ {now}"
        log.write(content)


@post(
    "/register",
    background=BackgroundTasks(
        tasks=[
            BackgroundTask(send_email_notification, message="Account created"),
            BackgroundTask(write_in_file),
        ]
    ),
)
async def create_user(data: UserIn) -> JSONResponse:
    JSONResponse({"message": "Created"})

PARAMETER	DESCRIPTION
tasks	TYPE: Optional[List[BackgroundTask]] DEFAULT: None

Source code in esmerald/backgound.py

def __init__(self, tasks: Optional[List[BackgroundTask]] = None):
    super().__init__(tasks=tasks)

func instance-attribute

func = func

args instance-attribute

args = args

kwargs instance-attribute

kwargs = kwargs

is_async instance-attribute

is_async = is_async_callable(func)

tasks instance-attribute

tasks = list(tasks) if tasks else []

add_task

add_task(func, *args, **kwargs)

PARAMETER	DESCRIPTION
func	TYPE: Callable[P, Any]
*args	TYPE: args DEFAULT: ()
**kwargs	TYPE: kwargs DEFAULT: {}

Source code in .venv/lib/python3.8/site-packages/starlette/background.py

def add_task(
    self, func: typing.Callable[P, typing.Any], *args: P.args, **kwargs: P.kwargs
) -> None:
    task = BackgroundTask(func, *args, **kwargs)
    self.tasks.append(task)