asyncio.gather, create_task and TaskGroup: Concurrency in Practice

Three ways to run multiple coroutines concurrently. Each has its own niche.

asyncio.gather() — the simplest

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:
        # All 5 requests run in parallel
        results = await asyncio.gather(*[fetch(client, u) for u in urls])
    print(results)  # [200, 200, 200, 200, 200]

asyncio.run(main())

gather() returns results in the same order as the input coroutines.

asyncio.create_task() — when you need control

async def main():
    # Task is created and IMMEDIATELY starts running
    task1 = asyncio.create_task(fetch(client, url1), name="fetch-1")
    task2 = asyncio.create_task(fetch(client, url2), name="fetch-2")

    # Can cancel while waiting
    await asyncio.sleep(0.1)
    task1.cancel()  # changed my mind

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

asyncio.wait() and as_completed() — for complex scenarios

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

    # Process as they complete — don't wait for the slowest
    async for coro in asyncio.as_completed(tasks):
        result = await coro
        print(f"Done: {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))
    # Reach here only when ALL tasks complete
    # If one fails — all others are automatically cancelled
    print(f"Collected: {len(results)}")

TaskGroup is the right choice for new code. It guarantees that when one task fails the others won’t hang forever.