Перейти к содержанию

Фоновые задачи

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

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

К примеру:

  • Отправка писем на почту после выполнения каких-либо действий:
    • Т.к. соединение с почтовым сервером и отправка письма идут достаточно "долго" (несколько секунд), вы можете отправить ответ пользователю, а отправку письма выполнить в фоне.
  • Обработка данных:
    • К примеру, если вы получаете файл, который должен пройти через медленный процесс, вы можете отправить ответ "Accepted" (HTTP 202) и отправить работу с файлом в фон.

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

Сначала импортируйте BackgroundTasks, потом добавьте в функцию параметр с типом BackgroundTasks:

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()


def write_notification(email: str, message=""):
    with open("log.txt", mode="w") as email_file:
        content = f"notification for {email}: {message}"
        email_file.write(content)


@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(write_notification, email, message="some notification")
    return {"message": "Notification sent in the background"}

FastAPI создаст объект класса BackgroundTasks для вас и запишет его в параметр.

Создание функции для фоновой задачи

Создайте функцию, которую хотите запустить в фоне.

Это совершенно обычная функция, которая может принимать параметры.

Она может быть как асинхронной async def, так и обычной def функцией, FastAPI знает, как правильно ее выполнить.

В нашем примере фоновая задача будет вести запись в файл (симулируя отправку письма).

Так как операция записи не использует async и await, мы определим ее как обычную def:

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()


def write_notification(email: str, message=""):
    with open("log.txt", mode="w") as email_file:
        content = f"notification for {email}: {message}"
        email_file.write(content)


@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(write_notification, email, message="some notification")
    return {"message": "Notification sent in the background"}

Добавление фоновой задачи

Внутри функции вызовите метод .add_task() у объекта background tasks и передайте ему функцию, которую хотите выполнить в фоне:

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()


def write_notification(email: str, message=""):
    with open("log.txt", mode="w") as email_file:
        content = f"notification for {email}: {message}"
        email_file.write(content)


@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(write_notification, email, message="some notification")
    return {"message": "Notification sent in the background"}

.add_task() принимает следующие аргументы:

  • Функцию, которая будет выполнена в фоне (write_notification). Обратите внимание, что передается объект функции, без скобок.
  • Любое упорядоченное количество аргументов, которые принимает функция (email).
  • Любое количество именованных аргументов, которые принимает функция (message="some notification").

Встраивание зависимостей

Класс BackgroundTasks также работает с системой встраивания зависимостей, вы можете определить BackgroundTasks на разных уровнях: как параметр функции, как завимость, как подзависимость и так далее.

FastAPI знает, что нужно сделать в каждом случае и как переиспользовать тот же объект BackgroundTasks, так чтобы все фоновые задачи собрались и запустились вместе в фоне:

from fastapi import BackgroundTasks, Depends, FastAPI

app = FastAPI()


def write_log(message: str):
    with open("log.txt", mode="a") as log:
        log.write(message)


def get_query(background_tasks: BackgroundTasks, q: str | None = None):
    if q:
        message = f"found query: {q}\n"
        background_tasks.add_task(write_log, message)
    return q


@app.post("/send-notification/{email}")
async def send_notification(
    email: str, background_tasks: BackgroundTasks, q: str = Depends(get_query)
):
    message = f"message to {email}\n"
    background_tasks.add_task(write_log, message)
    return {"message": "Message sent"}
🤓 Other versions and variants 
from typing import Annotated

from fastapi import BackgroundTasks, Depends, FastAPI

app = FastAPI()


def write_log(message: str):
    with open("log.txt", mode="a") as log:
        log.write(message)


def get_query(background_tasks: BackgroundTasks, q: str | None = None):
    if q:
        message = f"found query: {q}\n"
        background_tasks.add_task(write_log, message)
    return q


@app.post("/send-notification/{email}")
async def send_notification(
    email: str, background_tasks: BackgroundTasks, q: Annotated[str, Depends(get_query)]
):
    message = f"message to {email}\n"
    background_tasks.add_task(write_log, message)
    return {"message": "Message sent"}

from typing import Annotated, Union

from fastapi import BackgroundTasks, Depends, FastAPI

app = FastAPI()


def write_log(message: str):
    with open("log.txt", mode="a") as log:
        log.write(message)


def get_query(background_tasks: BackgroundTasks, q: Union[str, None] = None):
    if q:
        message = f"found query: {q}\n"
        background_tasks.add_task(write_log, message)
    return q


@app.post("/send-notification/{email}")
async def send_notification(
    email: str, background_tasks: BackgroundTasks, q: Annotated[str, Depends(get_query)]
):
    message = f"message to {email}\n"
    background_tasks.add_task(write_log, message)
    return {"message": "Message sent"}

from typing import Union

from fastapi import BackgroundTasks, Depends, FastAPI
from typing_extensions import Annotated

app = FastAPI()


def write_log(message: str):
    with open("log.txt", mode="a") as log:
        log.write(message)


def get_query(background_tasks: BackgroundTasks, q: Union[str, None] = None):
    if q:
        message = f"found query: {q}\n"
        background_tasks.add_task(write_log, message)
    return q


@app.post("/send-notification/{email}")
async def send_notification(
    email: str, background_tasks: BackgroundTasks, q: Annotated[str, Depends(get_query)]
):
    message = f"message to {email}\n"
    background_tasks.add_task(write_log, message)
    return {"message": "Message sent"}

Tip

Prefer to use the Annotated version if possible.

from typing import Union

from fastapi import BackgroundTasks, Depends, FastAPI

app = FastAPI()


def write_log(message: str):
    with open("log.txt", mode="a") as log:
        log.write(message)


def get_query(background_tasks: BackgroundTasks, q: Union[str, None] = None):
    if q:
        message = f"found query: {q}\n"
        background_tasks.add_task(write_log, message)
    return q


@app.post("/send-notification/{email}")
async def send_notification(
    email: str, background_tasks: BackgroundTasks, q: str = Depends(get_query)
):
    message = f"message to {email}\n"
    background_tasks.add_task(write_log, message)
    return {"message": "Message sent"}

В этом примере сообщения будут записаны в log.txt после того, как ответ сервера был отправлен.

Если бы в запрос был передан query-параметр q, он бы первыми записался в log.txt фоновой задачей (потому что вызывается в зависимости get_query).

После другая фоновая задача, которая была сгенерирована в функции, запишет сообщение из параметра email.

Технические детали

Класс BackgroundTasks основан на starlette.background.

Он интегрирован в FastAPI, так что вы можете импортировать его прямо из fastapi и избежать случайного импорта BackgroundTask (без s на конце) из starlette.background.

При использовании BackgroundTasks (а не BackgroundTask), вам достаточно только определить параметр функции с типом BackgroundTasks и FastAPI сделает все за вас, также как при использовании объекта Request.

Вы все равно можете использовать BackgroundTask из starlette в FastAPI, но вам придется самостоятельно создавать объект фоновой задачи и вручную обработать Response внутри него.

Вы можете подробнее изучить его в Официальной документации Starlette для BackgroundTasks.

Предостережение

Если вам нужно выполнить тяжелые вычисления в фоне, которым необязательно быть запущенными в одном процессе с приложением FastAPI (к примеру, вам не нужны обрабатываемые переменные или вы не хотите делиться памятью процесса и т.д.), вы можете использовать более серьезные инструменты, такие как Celery.

Их тяжелее настраивать, также им нужен брокер сообщений наподобие RabbitMQ или Redis, но зато они позволяют вам запускать фоновые задачи в нескольких процессах и даже на нескольких серверах.

Но если вам нужен доступ к общим переменным и объектам вашего FastAPI приложения или вам нужно выполнять простые фоновые задачи (наподобие отправки письма из примера) вы можете просто использовать BackgroundTasks.

Резюме

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