Asynchronous Programming and TaskGroups

Updated

September 7, 2026

Asynchronous Programming and TaskGroups

After reading this chapter, you will master asynchronous programming using Python’s asyncio engine, write non-blocking coroutines with async def and await, leverage modern structured concurrency with asyncio.TaskGroup, enforce deterministic deadlines with asyncio.timeout(), and throttle concurrent operations using asyncio.Semaphore.

Mental model

Asynchronous programming is cooperative multitasking on a single thread. Unlike OS threads where the kernel preempts execution at any time, a Python coroutine yields control voluntarily at explicit await boundaries:

Single-Threaded Event Loop Timeline:
  Time ──▶
  Event Loop: [ Coroutine A runs ] ──▶ Hits 'await socket.read()' ──▶ (Yields control)
                     │
                     ▼
              [ Coroutine B runs ] ──▶ Hits 'await socket.write()' ──▶ (Yields control)
                     │
                     ▼
              OS reports A's socket is ready!
                     │
                     ▼
              [ Coroutine A resumes ] ──▶ Finishes

Structured Concurrency with asyncio.TaskGroup

Introduced in Python 3.11, asyncio.TaskGroup replaces brittle patterns like asyncio.gather() with a strict execution contract:

with TaskGroup() as tg:
    tg.create_task(task_1())
    tg.create_task(task_2())
    tg.create_task(task_3())
         │
         ▼ If task_2 raises an exception:
  1. Cancels remaining siblings (task_1 & task_3)
  2. Awaits all sibling cancellations
  3. Bundles failures into an ExceptionGroup

No background tasks are ever left orphaned or leaked into the process.

Minimal example

Save as asyncio_fundamentals.py:

# asyncio_fundamentals.py
import asyncio
import time

async def fetch_cluster_status(node_id: str, delay: float) -> dict[str, str | float]:
    """Simulate non-blocking async network fetch."""
    print(f"  [Start] Polling {node_id} (simulated delay: {delay}s)...")
    await asyncio.sleep(delay)  # Non-blocking async sleep
    print(f"  [Done]  Received response from {node_id}")
    return {"node": node_id, "status": "ONLINE", "rtt_ms": delay * 1000}

async def main() -> None:
    print("--- Polling 3 cluster nodes concurrently via TaskGroup ---")
    t0 = time.perf_counter()
    
    results = []
    # Structured concurrency block
    async with asyncio.TaskGroup() as tg:
        t1 = tg.create_task(fetch_cluster_status("node-01", 0.05))
        t2 = tg.create_task(fetch_cluster_status("node-02", 0.08))
        t3 = tg.create_task(fetch_cluster_status("node-03", 0.03))

    # Reaching here guarantees all tasks completed safely
    results = [t1.result(), t2.result(), t3.result()]
    elapsed = time.perf_counter() - t0

    print(f"\nAll 3 tasks completed in {elapsed:.2f}s (Max delay, NOT sum!):")
    for res in results:
        print(f"  {res['node']} -> {res['status']} ({res['rtt_ms']:.1f}ms)")

if __name__ == "__main__":
    asyncio.run(main())

Run via uv run python asyncio_fundamentals.py:

--- Polling 3 cluster nodes concurrently via TaskGroup ---
  [Start] Polling node-01 (simulated delay: 0.05s)...
  [Start] Polling node-02 (simulated delay: 0.08s)...
  [Start] Polling node-03 (simulated delay: 0.03s)...
  [Done]  Received response from node-03
  [Done]  Received response from node-01
  [Done]  Received response from node-02

All 3 tasks completed in 0.08s (Max delay, NOT sum!):
  node-01 -> ONLINE (50.0ms)
  node-02 -> ONLINE (80.0ms)
  node-03 -> ONLINE (30.0ms)

Worked examples

Case 1: Deterministic Deadlines with asyncio.timeout()

Python 3.11+ provides asyncio.timeout(), an async context manager that cancels the enclosed coroutine if it exceeds the specified budget:

# async_timeout_demo.py
import asyncio

async def slow_database_query() -> str:
    print("  Executing database query (requires 2.0s)...")
    await asyncio.sleep(2.0)
    return "QUERY_RESULT"

async def main() -> None:
    # Set a strict 0.5-second deadline
    try:
        async with asyncio.timeout(0.5):
            result = await slow_database_query()
            print("Result:", result)
    except TimeoutError:
        print("  ALERT: Database query exceeded 0.5s deadline; cancelled cleanly!")

if __name__ == "__main__":
    asyncio.run(main())

Run:

uv run python async_timeout_demo.py

Output:

  Executing database query (requires 2.0s)...
  ALERT: Database query exceeded 0.5s deadline; cancelled cleanly!

Case 2: Throttling Concurrent Requests with asyncio.Semaphore

When crawling or querying an external API, spawning 10,000 tasks simultaneously can exhaust file descriptors or trigger rate-limit bans. A semaphore bounds active concurrency:

# async_semaphore_rate_limit.py
import asyncio
import time

async def worker_with_semaphore(task_id: int, sem: asyncio.Semaphore) -> None:
    async with sem:  # Only 2 workers allowed inside this block concurrently
        print(f"[{time.strftime('%X')}] Worker {task_id} entering critical section")
        await asyncio.sleep(0.05)
        print(f"[{time.strftime('%X')}] Worker {task_id} exiting")

async def main() -> None:
    # Limit concurrency to at most 2 simultaneous executions
    sem = asyncio.Semaphore(2)

    async with asyncio.TaskGroup() as tg:
        for i in range(5):
            tg.create_task(worker_with_semaphore(i + 1, sem))

if __name__ == "__main__":
    asyncio.run(main())

Run:

uv run python async_semaphore_rate_limit.py

Output:

[10:00:00] Worker 1 entering critical section
[10:00:00] Worker 2 entering critical section
[10:00:00] Worker 1 exiting
[10:00:00] Worker 3 entering critical section
[10:00:00] Worker 2 exiting
[10:00:00] Worker 4 entering critical section
...

Pitfalls

Pitfall 1: Calling Synchronous Blocking Code in Coroutines

Calling a blocking function (time.sleep(), requests.get(), or heavy CPU calculations) inside an async function halts the single-threaded event loop, freezing all other concurrent tasks:

# FATAL: Freezes entire event loop!
async def bad_handler():
    time.sleep(5)  # Freezes all users and connections!

# CORRECT: Use non-blocking async primitives
async def good_handler():
    await asyncio.sleep(5)

# OR offload blocking calls to a thread pool:
async def offloaded_handler():
    await asyncio.to_thread(blocking_function)

Pitfall 2: Using Legacy asyncio.gather() Without Shielding

If one task fails in asyncio.gather(), other tasks continue running in the background as unmanaged orphaned coroutines. Always prefer asyncio.TaskGroup().

Exercises

  1. Write an asynchronous function that polls an HTTP mock endpoint and retries up to 3 times with exponential backoff on failure.
  2. Use asyncio.TaskGroup to execute 10 concurrent tasks, and simulate one task raising an exception to verify structured sibling cancellation.
  3. Build a producer-consumer pipeline using asyncio.Queue.
  4. Use asyncio.to_thread() to run a CPU-bound hashing function inside an async application without blocking the event loop.

Further reading

  • Python Standard Library: asyncio documentation.
  • PEP 654: Exception Groups and except*.
  • Yury Selivanov: PEP 492 — Coroutines with async and await syntax.