asyncio в Python: event loop, корутини і коли async не допоможе

async def у FastAPI-ендпоінті ще не робить сервіс швидким. Асинхронність у Python — це спосіб не простоювати під час очікування (мережі, диска, бази), а не спосіб рахувати швидше. Розберімо, як це працює, щоб не наступати на типові граблі.

Event loop — один потік, багато задач

asyncio виконує корутини в одному потоці. Коли корутина доходить до await на операції вводу-виводу, вона віддає керування циклу подій, і той запускає іншу задачу, готову продовжити роботу.

import asyncio, time

async def fetch(name: str, delay: float) -> str:
    await asyncio.sleep(delay)      # імітація мережевого запиту
    return name

async def main():
    start = time.perf_counter()
    results = await asyncio.gather(fetch("a", 1), fetch("b", 1), fetch("c", 1))
    print(results, f"{time.perf_counter() - start:.1f}s")   # ['a', 'b', 'c'] 1.0s

asyncio.run(main())

Три «запити» по секунді — за одну секунду, бо очікування відбувалось одночасно.

Головна пастка: блокуючий виклик

@app.get("/report")
async def report():
    data = requests.get("https://api.example.com/data").json()   # ❌ блокує весь цикл
    time.sleep(2)                                                 # ❌ теж
    return data

Поки requests.get чекає відповіді, event loop стоїть — жоден інший запит до сервісу не обробляється. Один повільний зовнішній API — і весь сервіс «завис».

Що робити:

import asyncio, httpx

client = httpx.AsyncClient(timeout=10)

@app.get("/report")
async def report():
    r = await client.get("https://api.example.com/data")
    pdf = await asyncio.to_thread(render_pdf, r.json())   # синхронна бібліотека — в потоці
    return pdf

Порада

У FastAPI звичайні def-ендпоінти виконуються в пулі потоків, а async def — прямо в event loop. Якщо всередині лише синхронні бібліотеки, чесніше написати def, ніж async def з блокуючими викликами.

Структурована конкурентність: TaskGroup

TaskGroup (Python 3.11+) запускає задачі паралельно і гарантує, що жодна не «загубиться»: якщо одна впала, решта скасовуються, а помилка піднімається назовні.

async def load_dashboard(user_id: int):
    async with asyncio.TaskGroup() as tg:
        profile = tg.create_task(get_profile(user_id))
        orders = tg.create_task(get_orders(user_id))
        recs = tg.create_task(get_recommendations(user_id))
    return {"profile": profile.result(), "orders": orders.result(), "recs": recs.result()}

Таймаути

Мережевий виклик без таймауту може чекати вічно:

async with asyncio.timeout(3):
    data = await slow_service()
# TimeoutError через 3 секунди

Обмеження конкурентності

gather на 10 000 URL відкриє 10 000 з'єднань одночасно — і вас заблокує сторонній API або закінчаться файлові дескриптори. Обмежуйте семафором:

sem = asyncio.Semaphore(20)

async def fetch_limited(url: str):
    async with sem:
        r = await client.get(url)
        return r.status_code

results = await asyncio.gather(*(fetch_limited(u) for u in urls), return_exceptions=True)

Не загубіть задачу

asyncio.create_task(send_email(user))   # ❌ посилання ніде не зберігається

Event loop тримає на задачі лише слабке посилання — задачу, на яку ніхто не посилається, збирач сміття може знищити до завершення. Зберігайте посилання (наприклад, у множині, яку очищує add_done_callback) або використовуйте TaskGroup. А для справді фонової роботи (листи, звіти) краще підходить черга завдань.

Коли async не потрібен

• CPU-bound задачі: обробка зображень, ML, парсинг гігабайтів JSON — тут потрібні процеси або окремі сервіси. (Free-threaded збірки Python без GIL уже стабілізуються, але бібліотеки ще наздоганяють.)
• Скрипти, які роблять кілька послідовних запитів — синхронний код простіший і зрозуміліший.
• Команда не знайома з async, а навантаження невелике.