Background Tasks

Как и в Lilya, в Esmerald вы можете определять фоновые задачи для выполнения после возвращения ответа.

Это может быть полезно для тех операций, которые должны происходить после запроса, не блокируя клиента (клиенту не нужно ждать завершения) для получения того же ответа.

Примеры:

• Регистрация пользователя в системе и отправка электронного письма с подтверждением регистрации.
• Обработка файла, которая может занять "некоторое время". Просто верните HTTP 202 и обработайте файл в фоновом режиме.

Как использовать

Как уже упоминалось, Esmerald использует фоновые задачи из Lilya, и вы можете передавать их различными способами:

• Через обработчики
• Через ответ

Через handlers

Это означает передачу фоновой задачи через get, put, post, delete или route.

Существует также два способа передачи через обработчики.

Использование одного экземпляра

Это, вероятно, наиболее распространенный случай, когда вам нужно выполнить одну фоновую задачу при получении запроса, например, отправка уведомления по электронной почте.

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"})

Этот пример довольно простой, но он иллюстрирует, как вы могли бы использовать обработчики для передачи фоновой задачи.

Использование списка

Конечно, есть также ситуации, когда нужно выполнить более одной фоновой задачи.

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"})

Через response

Добавление задач через response, вероятно, будет тем способом, который вы будете использовать чаще всего, иногда вам понадобится какая-то конкретная информация, которая доступна только внутри вашего представления.

Это достигается аналогичным образом, как и в handlers.

Использование одного экземпляра

Таким же образом, как вы создавали одну фоновую задачу для handlers, в response это работает аналогично.

from typing import Dict

from pydantic import BaseModel

from esmerald import BackgroundTask, Request, Response, post


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


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


@post("/register")
async def create_user(data: UserIn, request: Request) -> Response(Dict[str, str]):
    return Response(
        {"message": "Email sent"},
        background=BackgroundTask(
            send_email_notification,
            email=request.user.email,
            message="Thank you for registering.",
        ),
    )

Использование списка

То же самое происходит при выполнении более одной фоновой задачи и когда требуется более одной операции.

from datetime import datetime
from typing import Dict

from pydantic import BaseModel

from esmerald import BackgroundTask, BackgroundTasks, Request, Response, post


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


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)


@post("/register")
async def create_user(data: UserIn, request: Request) -> Response(Dict[str, str]):
    return Response(
        {"message": "Email sent"},
        background=BackgroundTasks(
            tasks=[
                BackgroundTask(
                    send_email_notification,
                    email=request.user.email,
                    message="Thank you for registering.",
                ),
                BackgroundTask(write_in_file, email=request.user.email),
            ]
        ),
    )

Использование add_task

Другой способ добавления нескольких задач - использование функции add_task, предоставляемой объектом BackgroundTasks.

from datetime import datetime
from typing import Dict

from pydantic import BaseModel

from esmerald import BackgroundTasks, Request, Response, post


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


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)


@post("/register")
async def create_user(data: UserIn, request: Request) -> Response(Dict[str, str]):
    background_tasks = BackgroundTasks()
    background_tasks.add_task(
        send_email_notification, email=request.user.email, message="Thank you for registering."
    )
    background_tasks.add_task(write_in_file, email=request.user.email)

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

Функция .add_task() принимает в качестве аргументов:

• Функцию задачи, которая должна быть выполнена в фоновом режиме (send_email_notification и write_in_file).
• Любую последовательность аргументов (*args), которые должны быть переданы функции задачи (email, message).
• Любые именованные аргументы (**kwargs), которые должны быть переданы функции задачи.

Техническая информация

Классы BackgroundTask и BackgroundTasks приходят напрямую из lilya.background. Это означает, что вы можете импортировать их напрямую из Lilya, если хотите.

Esmerald импортирует эти классы и добавляет некоторую дополнительную информацию о типах, но не влияет на общую функциональность ядра.

Вы можете использовать функции def или async def при объявлении этих функций для передачи в BackgroundTask, Esmerald будет знать, как обработать их.

Для получения дополнительной информации об этих объектах вы можете ознакомиться с официальной документацией Lilya по фоновым задачам.

Справочник API

Узнайте больше о BackgroundTask, ознакомившись со Справочником API.