Background

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

Like Lilya, 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: Task

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 lilya/background.py

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

func instance-attribute

func = enforce_async_callable(func)

args instance-attribute

args = args

kwargs instance-attribute

kwargs = kwargs

esmerald.BackgroundTasks

BackgroundTasks(tasks=None, as_group=False)

Bases: Tasks

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: Sequence[Task] | None DEFAULT: None
as_group	TYPE: bool DEFAULT: False

Source code in lilya/background.py

def __init__(self, tasks: Sequence[Task] | None = None, as_group: bool = False):
    self.tasks = list(tasks) if tasks else []
    self.as_group = as_group

func instance-attribute

func = enforce_async_callable(func)

args instance-attribute

args = args

kwargs instance-attribute

kwargs = kwargs

tasks instance-attribute

tasks = list(tasks) if tasks else []

as_group instance-attribute

as_group = as_group

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 lilya/background.py

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

run_single async

run_single()

Source code in lilya/background.py

async def run_single(self) -> None:
    for task in self.tasks:
        await task()

run_as_group async

run_as_group()

Source code in lilya/background.py

async def run_as_group(self) -> None:
    async with anyio.create_task_group() as group:
        for task in self.tasks:
            group.start_soon(task)