Класс BackgroundTasks
Класс BackgroundTasks применяется для выполнения
задач после того, как FastAPI отправит ответ клиенту.
Это полезно для операций, которые не должны задерживать
ответ: отправка email, запись логов, обработка данных.
Объект класса передаётся в параметр обработчика,
а задачи добавляются методом add_task.
Синтаксис
from fastapi import BackgroundTasks
def handler(background_tasks: BackgroundTasks):
background_tasks.add_task(func, *args, **kwargs)
Пример
Давайте запустим простую фоновую задачу, которая выводит сообщение в консоль:
<+python+>
from fastapi import FastAPI, BackgroundTasks
app = FastAPI()
def write_log(message: str):
with open('log.txt', 'a') as f:
f.write(message + '\n')
@app.get('/')
def index(background_tasks: BackgroundTasks):
background_tasks.add_task(write_log, 'page visited')
return {'text': 'hello'}
<-python+>
Результат выполнения кода:
{"text": "hello"}
Пример
Давайте передадим несколько аргументов в фоновую задачу:
<+python+>
from fastapi import FastAPI, BackgroundTasks
app = FastAPI()
def send_email(to: str, subject: str, body: str):
print(f'Email to {to}: {subject} - {body}')
@app.post('/user')
def create_user(name: str, background_tasks: BackgroundTasks):
background_tasks.add_task(
send_email,
'user@example.com',
'Welcome',
f'Hello, {name}!'
)
return {'name': name}
<-python+>
Результат выполнения кода для запроса
с параметром name=abcde:
{"name": "abcde"}
Пример
Давайте добавим несколько фоновых задач в одном обработчике:
<+python+>
from fastapi import FastAPI, BackgroundTasks
app = FastAPI()
def task_one():
print('task one')
def task_two():
print('task two')
@app.get('/tasks')
def run_tasks(background_tasks: BackgroundTasks):
background_tasks.add_task(task_one)
background_tasks.add_task(task_two)
return {'status': 'scheduled'}
<-python+>
Результат выполнения кода:
{"status": "scheduled"}
Смотрите также
-
класс
BackgroundTasks,
который управляет фоновыми задачами -
метод
add_task,
который добавляет задачу в очередь выполнения
