Фонові задачі¶

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

Це корисно для операцій, які потрібно виконати після обробки запиту, але клієнту не обов’язково чекати завершення цієї операції перед отриманням відповіді.

Це включає, наприклад:

Надсилання email-сповіщень після виконання певної дії:
- Підключення до поштового сервера та надсилання листа може займати кілька секунд. Ви можете відразу повернути відповідь, а email-сповіщення надіслати у фоні.
Обробка даних:
- Наприклад, якщо ви отримали файл, який потрібно обробити довготривалим процесом, можна повернути відповідь «Accepted» (HTTP 202) і виконати обробку файлу у фоні.

Використання `BackgroundTasks`¶

Спочатку імпортуйте BackgroundTasks і оголосіть параметр у вашій функції операції шляху з анотацією типу BackgroundTasks:

Python 3.9+

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 обробить її правильно.

У нашому випадку функція записує у файл (імітуючи надсилання email).

І оскільки операція запису не використовує async та await, ми визначаємо функцію як звичайну def:

Python 3.9+

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

Додавання фонової задачі¶

Усередині вашої функції операції шляху, передайте функцію задачі в об'єкт background tasks, використовуючи метод .add_task():

Python 3.9+

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.10+ - non-AnnotatedPython 3.9+ - 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"}

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 про Background Tasks.

Застереження¶

Якщо вам потрібно виконувати складні фонові обчислення, і при цьому нема потреби запускати їх у тому ж процесі (наприклад, не потрібно спільного доступу до пам’яті чи змінних), можливо, варто скористатися більш потужними інструментами, такими як Celery.

Такі інструменти зазвичай потребують складнішої конфігурації та менеджера черги повідомлень/завдань, наприклад, RabbitMQ або Redis. Однак вони дозволяють виконувати фонові задачі в кількох процесах і особливо — на кількох серверах.

Якщо ж вам потрібно отримати доступ до змінних і об’єктів із тієї ж FastAPI-програми або виконувати невеликі фонові завдання (наприклад, надсилати email-сповіщення), достатньо просто використовувати BackgroundTasks.

Підсумок¶

Імпортуйте та використовуйте BackgroundTasks як параметри у функціях операції шляху та залежностях, щоб додавати фонові задачі.

Фонові задачі¶

Використання BackgroundTasks¶

Створення функції задачі¶

Додавання фонової задачі¶

Впровадження залежностей¶

Технічні деталі¶

Застереження¶

Підсумок¶

Використання `BackgroundTasks`¶