Tasks

Isto pode ser útil para operações que precisam de acontecer após o pedido sem bloquear o cliente (o cliente não precisa esperar que complete) para receber a resposta.

Para associar uma tarefa em segundo plano com uma resposta, a tarefa será executada somente após a resposta ter sido enviada, isto significa que uma tarefa em segundo plano deve ser anexada a uma Resposta.

Exemplo:

• Registrar um utilizador no sistema e enviar um e-mail a confirmar o registro.
• Processar um ficheiro que pode levar "algum tempo". Simplesmente retorna um HTTP 202 e processa o ficheiro em segundo plano.

Usando uma lista

É claro que também há situações em que mais de uma tarefa em segundo plano pode acontecer.

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"],
        )
    ]
)

Através da resposta

Adicionar tarefas através da resposta provavelmente será a maneira que usará com mais frequência e o motivo é que às vezes precisará de algumas informações específicas que só estão disponíveis dentro da sua view.

Usando uma única instância

Da mesma forma que criou uma única tarefa em segundo plano para os handlers, na resposta funciona de maneira semelhante.

Usando uma lista

O mesmo acontece ao executar mais do que uma tarefa em segundo plano e quando mais do que uma operação é necessária.

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"],
        )
    ]
)

Usando o add_task

Outra maneira de adicionar várias tarefas é utilizando a função add_tasks fornecida pelo objecto Tasks.

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"],
        )
    ]
)

O .add_task() recebe como argumentos:

• Uma função de tarefa a ser executada em segundo plano (send_email_notification e write_in_file).
• Qualquer sequência de argumentos que devem ser passados para a função de tarefa na ordem (email, message).
• Quaisquer argumentos de palavra-chave que devem ser passados para a função de tarefa.

Informações técnicas

As classes Task e Tasks derivam diretamente de lilya.background, mas a natureza dos objectos também permite o uso de bibliotecas externas como backgrounder.

Pode usar funções def ou async def ao declarar essas funcionalidades para serem passadas para o Task e o Lilya saberá como lidar com isso.

O objecto Tasks também aceita o parâmetro as_group. Isso permite que o anyio crie um grupo de tarefas e as execute.