У меня есть программа на Python 3.8, которая запускает сопрограмму, которая работает бесконечно, планируя другие сопрограммы в фоновом режиме через create_task, но не ожидая их. Эти сопрограммы используют AsyncEngine SQLAlchemy для записи некоторых данных в базу данных. Если возникает исключение, обработчик отменяет циклическую сопрограмму, а затем ожидает выполнения оставшихся фоновых задач через сбор. Кажется, все работает нормально.
Урезанный пример:
Код: Выделить всё
async def write_loop():
async_engine: AsyncEngine = get_async_engine()
while True:
data = await get_some_data()
bg_task: Task = asyncio.create_task(write_to_db(async_engine, data))
outstanding_tasks.add(bg_task)
bg_task.add_done_callback(outstanding_tasks.discard)
async def main():
try:
await write_loop()
except:
await asyncio.gather(*outstanding_tasks)
Я использую pytest и pytest-asyncio для реализации некоторых тестовых примеров, реализующих описанное выше. Исключение вызывается посредством некоторых исправлений, поэтому строка с сбором — это последнее, что должно произойти. Каждый тестовый пример украшен @pytest.mark.asyncio, чтобы гарантировать, что ни один из них не использует общий цикл событий. Все они проходят при индивидуальном запуске. При одновременном запуске нескольких операций записи SQLAlchemy завершатся с ошибкой, поскольку невозможно выполнить операцию: выполняется другая операция. Поиск в Google сказал мне, что мне нужно отключить пул соединений в тесте, что, похоже, подразумевает, что разные тесты каким-то образом используют AsyncEngine (или, по крайней мере, базовый пул соединений?), несмотря на то, что каждый из них создает свой собственный.
Сделав это, я больше не получаю вышеуказанную ошибку. Однако теперь, если я запускаю слишком много тестов одновременно, по крайней мере, один тайм-аут (независимо от того, насколько велик я установлен тайм-аут). Я вижу, что точка, в которой он зависает, — это ожидание asyncio.gather(*outstanding_tasks). Если я распечатаю оставшиеся задачи перед вызовом сборки, я увижу, что они находятся в состоянии ожидания и имеют ожидание wait_for=
Подробнее здесь:
https://stackoverflow.com/questions/778 ... sqlalchemy
Мобильная версия