Как асинхронно запускать задачи обработки данных в быстром API?
У меня есть маршрут в моем быстром приложении API (пример кода ниже). мой корневой маршрут вызывает асинхронную функцию read_root. в этой функции я хочу запросить базу данных, получить некоторый результат и загрузить его в aws в корзину s3. Мне не обязательно ждать завершения операций с базой данных, поскольку обработка данных и их загрузка в s3 может занять некоторое время. Я создал еще одну асинхронную функцию под названием call_database, выполняющую вызов базы данных, обработку и, наконец, загрузку файлов в s3, но вся эта операция является синхронной, т. е. вызов базы данных, получение результатов из базы данных и обработка данных, а затем загрузка в s3.
мой вопрос: может ли функция быть определена как асинхронный процесс таких синхронных операций? и пока они находятся в процессе, можно ли в моем случае вернуть ответ («привет»: «мир») на http-вызов. это завершит этот http-запрос/ответ. но я предполагаю, что функция call_database будет работать до тех пор, пока файлы не будут загружены в s3?
это правильное понимание и реализация?
from typing import Union
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
async def read_root():
await call_database()
return {"Hello": "World"}
async def call_database():
//syncrohonous call to database
//get the result back , and upload it to aws s3
Чтобы заменить выполнение этих операций (вызов базы данных, загрузка s3), вы, вероятно, захотите использовать очереди задач. FastAPI включает поддержку фоновых задач . Другой рекомендуемый подход — использование Celery, проверенной библиотеки для использования очередей задач в Python.
Связанные вопросы по StackOverflow: асинхронное/ожидающее поведение , фоновые задачи FastAPI
@amoralesc — спасибо за ваши идеи. Я пройду по ссылке, которую вы предоставили. концептуально я понимаю, что ключевое слово await будет ждать, пока задача не будет завершена, а затем продолжит остальную часть программы. Итак, есть ли лучший пример, когда нам следует использовать конструкцию async/await?
Конструкции async/await и модуль asyncio в целом лучше всего использовать, когда ваша программа привязана к вводу-выводу, и я думаю, что улучшение производительности более очевидно, когда одновременно выполняется несколько операций ввода-вывода. Обращение к базе данных — это именно операция ввода-вывода (как и загрузка файла в S3, если уж на то пошло). Поскольку вы используете FastAPI, я также предполагаю, что вы используете сервер ASGI, такой как uvicorn + построитель ORM/запросов, который поддерживает асинхронность. В этом сценарии вполне допустимо определить вызовы базы данных как асинхронные функции.
Поскольку операции с базой данных — это то, что вы обычно ожидаете вернуть своему клиенту (будь то возврат запроса, результата операции записи или возможной ошибки при выполнении), здесь лучше всего работает async/await (в отличие от выгрузки их в фоновый режим). исполнитель). Но реальный прирост производительности по сравнению с синхронным кодом можно увидеть только в том случае, если вы его измеряете. Поэтому я рекомендую вам запустить несколько тестов, отправлять несколько запросов в секунду к вашему API и тестировать его самостоятельно. Также ознакомьтесь с документацией FastAPI по асинхронности.
Теперь очень медленные операции (под медленными я подразумеваю >10 секунд), результат которых обычно не ожидается немедленно сообщить клиенту (например, загрузка больших файлов в S3), лучше всего подходят для фонового исполнителя. Именно здесь на помощь приходят фоновые задачи FastAPI или задачи Celery. Затем вы можете выбрать, как сообщить о результате этих операций вашему клиенту, в зависимости от вашего варианта использования: простое обновление базы данных, электронная почта, веб-перехватчик и т. д. Другим примером фоновых задач являются запланированные единицы работы, которые вам необходимо запускать каждый интервал x. времени, например очистка файлов на сервере каждый день в полночь.
В дополнение к ссылкам, представленным на баннере выше, обратите внимание на этот ответ и этот ответ.
Ответы 1
Думаю, вы ищете фоновое задание
from typing import Union
from fastapi import FastAPI, BackgroundTasks
app = FastAPI()
def upload_to_s3(data: dict):
# your upload process here
@app.get("/")
async def read_root(background_tasks: BackgroundTasks):
data_to_upload = await call_database()
background_tasks.add_task(upload_to_s3, data_to_upload)
return {"Hello": "World"}
async def call_database():
//syncrohonous call to database
Вам не хватает декларации для background_tasks.
Поведение
awaitв Python блокирует выполнение сопрограммы (в данном случае функцииread_root) до тех пор, пока ожидаемое не завершится (функцияcall_database). На практике это означает, что ваш текущий код не вернет ответ{"Hello": "World"}до тех пор, пока операция с базой данных не будет завершена.