Tasks

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.

To associate a background task with a response, the task will execute only after the response has been dispatched, this means that a background task must be attached to a Response.

Example:

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

Using a list

Of course there is also the situation where more than one background task needs to happen.

from datetime import datetime

from lilya.apps import Lilya
from lilya.background import Task, Tasks
from lilya.responses import JSONResponse
from lilya.routing import Path


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)


async def create_user() -> JSONResponse:
    background = (
        Tasks(
            tasks=[
                Task(send_email_notification, message="Account created"),
                Task(write_in_file),
            ]
        ),
    )
    JSONResponse({"message": "Created"}, background=background)


app = Lilya(
    routes=[
        Path(
            "/register",
            create_user,
            methods=["POST"],
        )
    ]
)

Via response

Adding tasks via response will be probably the way you will be using more often and the reson being is that sometimes you will need some specific information that is only available inside your view.

Using a single instance

In the same way you created a single background task for the handlers, in the response works in a similar way.

Using a list

The same happens when executing more than one background task and when more than one operation is needed.

from datetime import datetime
from typing import Dict

from lilya.apps import Lilya
from lilya.background import Task, Tasks
from lilya.requests import Request
from lilya.responses import Response
from lilya.routing import Path


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


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


async def create_user(request: Request) -> Response(Dict[str, str]):
    return Response(
        {"message": "Email sent"},
        background=Tasks(
            tasks=[
                Task(
                    send_email_notification,
                    email="example@lilya.dev",
                    message="Thank you for registering.",
                ),
                Task(write_in_file, email="example@lilya.dev"),
            ]
        ),
    )


app = Lilya(
    routes=[
        Path(
            "/register",
            create_user,
            methods=["POST"],
        )
    ]
)

Using the add_task

Another way of adding multiple tasks is by using the add_tasks function provided by the Tasks object.

from datetime import datetime

from lilya.apps import Lilya
from lilya.background import Tasks
from lilya.requests import Request
from lilya.responses import Response
from lilya.routing import Path


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


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


async def create_user(request: Request):
    background_tasks = Tasks(as_group=True)
    background_tasks.add_task(
        send_email_notification,
        email="example@lilya.dev",
        message="Thank you for registering.",
    )
    background_tasks.add_task(write_in_file, email=request.user.email)

    return Response({"message": "Email sent"}, background=background_tasks)


app = Lilya(
    routes=[
        Path(
            "/register",
            create_user,
            methods=["POST"],
        )
    ]
)

The .add_task() receives as arguments:

• A task function to be run in the background (send_email_notification and write_in_file).
• Any sequence of arguments that should be passed to the task function in order (email, message).
• Any keyword arguments that should be passed to the task function.

Technical information

The class Task and Tasks come directly from lilya.background but the nature of the objects also allows the use of external libraries like backgrounder.

You can use def or async def functions when declaring those functionalities to be passed to the Task and Lilya will know how to handle those for you.

The Tasks obejct also accepts the as_group parameter. This enables anyio to create a task group and run them.