async/await v Pythonu: asyncio srozumitelně a prakticky

Asynchronní programování v Pythonu umožňuje jednomu vláknu obsluhovat stovky souběžných operací – typicky síťových požadavků. Není to ale kouzelné zrychlení všeho. Pojďme si ukázat, kdy async dává smysl a jak ho psát správně.

Kdy async pomůže a kdy ne

Typ úlohy	Pomůže asyncio?
Mnoho HTTP požadavků, websockety, chatboti	Ano, výrazně
Čekání na databázi s async ovladačem	Ano
Výpočty (zpracování obrázků, matematika)	Ne – použijte multiprocessing
Jeden požadavek za druhým bez souběhu	Ne, jen přidá složitost

Asyncio je kooperativní: korutina se vzdá řízení jen v místě await. Pokud kód nikde nečeká, nic se souběžně neděje.

Základní pojmy

import asyncio

async def pozdrav(jmeno: str, zpozdeni: float) -> str:
    await asyncio.sleep(zpozdeni)
    return f"Ahoj, {jmeno}"

async def main():
    vysledek = await pozdrav("Petře", 1)
    print(vysledek)

asyncio.run(main())

• async def definuje korutinovou funkci. Její zavolání nic nespustí, jen vrátí objekt korutiny.
• await korutinu spustí a počká na výsledek.
• asyncio.run() vytvoří smyčku událostí, spustí v ní hlavní korutinu a na konci smyčku uklidí.

Souběh: gather

Takhle se čeká postupně (celkem 3 sekundy):

async def main():
    await pozdrav("A", 1)
    await pozdrav("B", 1)
    await pozdrav("C", 1)

Takhle souběžně (celkem 1 sekunda):

async def main():
    vysledky = await asyncio.gather(
        pozdrav("A", 1),
        pozdrav("B", 1),
        pozdrav("C", 1),
    )
    print(vysledky)  # pořadí odpovídá pořadí argumentů

TaskGroup (Python 3.11+)

Modernější a bezpečnější alternativa ke gather je asyncio.TaskGroup. Když jedna úloha selže, ostatní se zruší a výjimky se vyhodí jako ExceptionGroup:

async def main():
    async with asyncio.TaskGroup() as tg:
        t1 = tg.create_task(pozdrav("A", 1))
        t2 = tg.create_task(pozdrav("B", 2))
    print(t1.result(), t2.result())

U gather bez return_exceptions=True se první výjimka propaguje, ale ostatní úlohy běží dál „na pozadí“ – to bývá zdroj záhadných chyb.

Praktický příklad: stahování mnoha URL

import asyncio
import httpx

URLS = [f"https://httpbin.org/delay/1?i={i}" for i in range(20)]

async def stahni(client: httpx.AsyncClient, sem: asyncio.Semaphore, url: str) -> int:
    async with sem:
        r = await client.get(url, timeout=10)
        return r.status_code

async def main():
    sem = asyncio.Semaphore(5)  # max 5 požadavků najednou
    async with httpx.AsyncClient() as client:
        kody = await asyncio.gather(*(stahni(client, sem, u) for u in URLS))
    print(kody)

asyncio.run(main())

Klíčové body:

• Jeden sdílený AsyncClient – znovu používá spojení. Vytvářet klienta pro každý požadavek je pomalé.
• Semafor omezuje souběh. Bez něj byste na server poslali všechny požadavky naráz.
• Timeout vždy. Jinak jedna zaseknutá odpověď drží celou dávku.

Timeouty pro celé bloky

Od Pythonu 3.11 existuje asyncio.timeout:

async def main():
    try:
        async with asyncio.timeout(5):
            await nejaka_pomala_operace()
    except TimeoutError:
        print("Nestihlo se to")

Ve starších verzích použijte asyncio.wait_for(korutina, timeout=5).

Nejčastější chyby

1. Blokující volání uvnitř async funkce

async def spatne():
    time.sleep(2)           # zablokuje celou smyčku!
    requests.get(url)       # taky blokuje

async def dobre():
    await asyncio.sleep(2)
    async with httpx.AsyncClient() as c:
        await c.get(url)

Pokud musíte použít blokující knihovnu, pošlete ji do vlákna:

data = await asyncio.to_thread(blokujici_funkce, arg)

2. Zapomenutý await

async def main():
    pozdrav("A", 1)  # RuntimeWarning: coroutine was never awaited

Korutina se vůbec nespustí. Python vypíše varování, ale program pokračuje.

3. Úloha, na kterou nikdo nedrží referenci

asyncio.create_task(prace())  # může být uklizena garbage collectorem

Uložte si referenci (např. do množiny) nebo použijte TaskGroup.

4. asyncio.run() uvnitř běžící smyčky – například v Jupyteru, kde smyčka už běží. Tam stačí rovnou await main().

Ladění

Spusťte program s debug režimem asyncio:

PYTHONASYNCIODEBUG=1 python app.py
# nebo
python -X dev app.py

Asyncio pak hlásí korutiny, které nebyly awaitnuty, a operace, které blokovaly smyčku déle než 100 ms – přesně ty chyby, které se jinak hledají nejhůř.

Shrnutí

• Async se vyplatí pro I/O se spoustou čekání, ne pro výpočty.
• Souběh vzniká přes gather, TaskGroup nebo create_task, ne samotným await.
• Omezujte souběh semaforem a vždy nastavujte timeouty.
• Nepoužívejte blokující knihovny v async def, případně je obalte asyncio.to_thread.