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

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

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

Например:

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

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

Сначала импортируйте BackgroundTasks и объявите параметр в вашей функции‑обработчике пути с типом BackgroundTasks:

Python 3.8+

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:

Python 3.8+

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():

Python 3.8+

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 на нескольких уровнях: в функции‑обработчике пути, в зависимости (dependable), в подзависимости и т. д.

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

Python 3.10+

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

🤓 Other versions and variants

Python 3.9+Python 3.8+Python 3.10+ - non-AnnotatedPython 3.8+ - non-Annotated

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

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), она будет записана в лог фоновой задачей.

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

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

Класс BackgroundTasks приходит напрямую из starlette.background.

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

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

По‑прежнему можно использовать один BackgroundTask в FastAPI, но тогда вам нужно создать объект в своём коде и вернуть Starlette Response, включающий его.

Подробнее см. в официальной документации Starlette по фоновым задачам.

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

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

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

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

Резюме¶

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

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

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

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

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

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

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

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

Резюме¶

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