asyncio.gather, create_task и TaskGroup: конкурентность на практике

Три способа запустить несколько coroutines конкурентно. У каждого своя ниша.

asyncio.gather() — самый простой

import asyncio, httpx

async def fetch(client, url):
    r = await client.get(url)
    return r.status_code

async def main():
    urls = ["https://httpbin.org/get"] * 5
    async with httpx.AsyncClient() as client:
        # Все 5 запросов идут параллельно
        results = await asyncio.gather(*[fetch(client, u) for u in urls])
    print(results)  # [200, 200, 200, 200, 200]

asyncio.run(main())

gather() возвращает список результатов в том же порядке что и входные coroutines.

asyncio.create_task() — когда нужен контроль

async def main():
    # Task создаётся и СРАЗУ начинает выполняться
    task1 = asyncio.create_task(fetch(client, url1), name="fetch-1")
    task2 = asyncio.create_task(fetch(client, url2), name="fetch-2")

    # Можно отменить пока ждёшь
    await asyncio.sleep(0.1)
    task1.cancel()  # если передумал

    try:
        result = await task2
    except asyncio.CancelledError:
        pass

asyncio.wait() и as_completed() — для сложных сценариев

async def main():
    tasks = {asyncio.create_task(fetch(c, u)) for u in urls}

    # Обрабатываем по мере готовности — не ждём самого медленного
    async for coro in asyncio.as_completed(tasks):
        result = await coro
        print(f"Готово: {result}")

TaskGroup (Python 3.11+) — structured concurrency

async def main():
    results = []
    async with asyncio.TaskGroup() as tg:
        for url in urls:
            tg.create_task(fetch_and_save(url, results))
    # Сюда попадаем только когда ВСЕ задачи завершились
    # Если хоть одна упала — все остальные автоматически отменяются
    print(f"Собрано: {len(results)}")

TaskGroup — правильный выбор для нового кода. Он гарантирует что при ошибке в одной задаче остальные не зависнут навсегда.