Python Coroutines and asyncio Explained — Internals, Gotchas and Production Patterns

import asyncio
import time

# ---------------------------------------------------------------------------
# This example shows TWO tasks running concurrently on ONE thread.
# Watch the timestamps — both tasks overlap in wall time.
# ---------------------------------------------------------------------------

async def fetch_user_profile(user_id: int) -> dict:
    """Simulates a DB round-trip with a network delay."""
    print(f"[{time.perf_counter():.3f}s] START  fetch_user_profile({user_id})")
    # asyncio.sleep yields control back to the event loop.
    # The loop can run OTHER coroutines while this one 'waits'.
    await asyncio.sleep(1.0)  # pretend this is: await db.fetch_one(query)
    print(f"[{time.perf_counter():.3f}s] FINISH fetch_user_profile({user_id})")
    return {"id": user_id, "name": f"User_{user_id}"}


async def fetch_user_orders(user_id: int) -> list:
    """Simulates a second DB query running at the same time."""
    print(f"[{time.perf_counter():.3f}s] START  fetch_user_orders({user_id})")
    await asyncio.sleep(1.5)  # slightly longer query
    print(f"[{time.perf_counter():.3f}s] FINISH fetch_user_orders({user_id})")
    return [{"order_id": 101, "item": "Widget"}, {"order_id": 102, "item": "Gadget"}]


async def build_user_dashboard(user_id: int) -> None:
    wall_start = time.perf_counter()

    # asyncio.gather schedules BOTH coroutines as Tasks immediately.
    # They run concurrently — the loop switches between them at each 'await'.
    profile, orders = await asyncio.gather(
        fetch_user_profile(user_id),
        fetch_user_orders(user_id),
    )

    wall_elapsed = time.perf_counter() - wall_start
    print(f"\nDashboard ready in {wall_elapsed:.3f}s  (would be 2.5s if sequential)")
    print(f"Profile : {profile}")
    print(f"Orders  : {orders}")


if __name__ == "__main__":
    # asyncio.run() creates a fresh event loop, runs the coroutine to
    # completion, then closes and cleans up the loop. Use this as your
    # single entry point — never nest asyncio.run() calls.
    asyncio.run(build_user_dashboard(user_id=42))

import asyncio
import time


async def slow_api_call(endpoint: str, latency_seconds: float) -> str:
    """Represents any I/O-bound operation with a known latency."""
    await asyncio.sleep(latency_seconds)
    return f"Response from {endpoint}"


async def sequential_approach() -> None:
    """BAD PATTERN: awaiting coroutines one-by-one is just synchronous code."""
    print("--- Sequential approach ---")
    start = time.perf_counter()

    # Each await BLOCKS this coroutine until the call finishes.
    # No concurrency here — we might as well use requests.get().
    result_a = await slow_api_call("/users", latency_seconds=1.0)
    result_b = await slow_api_call("/products", latency_seconds=1.0)
    result_c = await slow_api_call("/inventory", latency_seconds=1.0)

    elapsed = time.perf_counter() - start
    print(f"Results: {result_a}, {result_b}, {result_c}")
    print(f"Time: {elapsed:.2f}s  ← 3 seconds: purely sequential\n")


async def task_based_approach() -> None:
    """GOOD PATTERN: create_task schedules coroutines concurrently."""
    print("--- Task-based approach ---")
    start = time.perf_counter()

    # create_task IMMEDIATELY schedules the coroutine on the event loop.
    # Execution doesn't start until the current coroutine yields (hits an await).
    task_users     = asyncio.create_task(slow_api_call("/users",     latency_seconds=1.0))
    task_products  = asyncio.create_task(slow_api_call("/products",  latency_seconds=1.0))
    task_inventory = asyncio.create_task(slow_api_call("/inventory", latency_seconds=1.0))

    # Now await the tasks. All three are already running concurrently.
    result_a = await task_users
    result_b = await task_products
    result_c = await task_inventory

    elapsed = time.perf_counter() - start
    print(f"Results: {result_a}, {result_b}, {result_c}")
    print(f"Time: {elapsed:.2f}s  ← ~1 second: all tasks ran concurrently\n")


async def gather_approach() -> None:
    """CLEANER PATTERN: gather is create_task + await rolled into one call."""
    print("--- gather approach ---")
    start = time.perf_counter()

    # gather wraps each coroutine in a Task internally, then awaits all of them.
    # return_exceptions=True means one failure won't cancel the siblings.
    results = await asyncio.gather(
        slow_api_call("/users",     latency_seconds=1.0),
        slow_api_call("/products",  latency_seconds=1.0),
        slow_api_call("/inventory", latency_seconds=1.0),
        return_exceptions=True,   # production-safe: inspect results for exceptions
    )

    elapsed = time.perf_counter() - start
    for r in results:
        if isinstance(r, Exception):
            print(f"  Task failed: {r}")
        else:
            print(f"  OK: {r}")
    print(f"Time: {elapsed:.2f}s\n")


if __name__ == "__main__":
    asyncio.run(sequential_approach())
    asyncio.run(task_based_approach())
    asyncio.run(gather_approach())

import asyncio
import time


async def database_query(query_name: str, duration: float) -> str:
    """Simulates a DB query. Handles cancellation cleanly."""
    try:
        print(f"  [{query_name}] starting...")
        await asyncio.sleep(duration)
        return f"[{query_name}] result after {duration}s"
    except asyncio.CancelledError:
        # CRITICAL: always do cleanup here (close connections, rollback, etc.)
        # then RE-RAISE — never silently swallow CancelledError.
        print(f"  [{query_name}] cancelled — cleaning up resources")
        raise  # <-- if you omit this, cancellation breaks silently
    finally:
        # finally always runs: use it for resource cleanup regardless of outcome
        print(f"  [{query_name}] finally block ran")


async def demonstrate_wait_for_timeout() -> None:
    """asyncio.wait_for raises TimeoutError when the deadline is exceeded."""
    print("\n=== wait_for timeout demo ===")
    try:
        result = await asyncio.wait_for(
            database_query("slow_analytics", duration=5.0),
            timeout=1.5,  # we'll only wait 1.5 seconds
        )
        print(f"Got: {result}")
    except asyncio.TimeoutError:
        # TimeoutError wraps CancelledError — the inner coroutine DID get cancelled.
        print("Query timed out — returning cached/default data instead")


async def demonstrate_task_group() -> None:
    """TaskGroup (Python 3.11+) — structured concurrency at its best."""
    print("\n=== TaskGroup structured concurrency demo ===")
    results = []
    start = time.perf_counter()

    try:
        async with asyncio.TaskGroup() as task_group:
            # All tasks are tracked by the group.
            # If ANY task raises, the group cancels the rest and re-raises
            # all exceptions together as an ExceptionGroup.
            t1 = task_group.create_task(database_query("user_lookup",   0.8))
            t2 = task_group.create_task(database_query("order_history", 1.0))
            t3 = task_group.create_task(database_query("recommendations", 0.6))
        # All tasks are DONE here — the context manager awaits all of them.
        results = [t1.result(), t2.result(), t3.result()]
    except* ValueError as eg:  # except* syntax handles ExceptionGroup
        print(f"Some tasks failed with ValueError: {eg.exceptions}")

    elapsed = time.perf_counter() - start
    for r in results:
        print(f"  {r}")
    print(f"TaskGroup finished in {elapsed:.2f}s")


async def demonstrate_cancellation_propagation() -> None:
    """Shows that cancelling a parent Task cascades to children."""
    print("\n=== Cancellation propagation demo ===")

    async def parent_workflow() -> None:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(database_query("child_a", 10.0))
            tg.create_task(database_query("child_b", 10.0))

    parent_task = asyncio.create_task(parent_workflow())
    await asyncio.sleep(0.5)   # let children start
    parent_task.cancel()        # cancel the parent

    try:
        await parent_task
    except asyncio.CancelledError:
        print("Parent task was cancelled — both children were also cancelled")


if __name__ == "__main__":
    asyncio.run(demonstrate_wait_for_timeout())
    asyncio.run(demonstrate_task_group())
    asyncio.run(demonstrate_cancellation_propagation())

import asyncio
import time
import httpx          # async HTTP client — pip install httpx
import requests       # synchronous HTTP client — pip install requests
from concurrent.futures import ThreadPoolExecutor


# ---------------------------------------------------------------------------
# Simulating CPU-heavy work (e.g. image processing, parsing, encryption)
# ---------------------------------------------------------------------------

def cpu_bound_resize(image_id: int) -> str:
    """Pretend this is PIL image processing — pure Python, GIL-bound."""
    # Simulate work with a tight loop — this BLOCKS the event loop if called
    # directly from a coroutine.
    result = sum(i * i for i in range(500_000))  # ~50ms on typical hardware
    return f"image_{image_id}_resized (checksum={result % 9999})"


async def bad_blocking_example() -> None:
    """This coroutine BLOCKS the event loop for ~50ms per call."""
    print("[BAD]  Calling CPU-bound work directly in a coroutine...")
    start = time.perf_counter()

    # These run SEQUENTIALLY and BLOCK the loop — no other coroutine can run.
    results = [cpu_bound_resize(img_id) for img_id in range(5)]

    elapsed = time.perf_counter() - start
    print(f"[BAD]  Done in {elapsed:.3f}s — event loop was blocked the whole time")


async def good_thread_executor_example() -> None:
    """Offload blocking work to a thread pool — loop stays free."""
    print("[GOOD] Offloading CPU-bound work to thread pool...")
    loop = asyncio.get_running_loop()
    start = time.perf_counter()

    # run_in_executor submits work to a ThreadPoolExecutor.
    # The event loop is FREE while the thread runs — other coroutines proceed.
    with ThreadPoolExecutor(max_workers=4) as pool:
        tasks = [
            loop.run_in_executor(pool, cpu_bound_resize, img_id)
            for img_id in range(5)
        ]
        results = await asyncio.gather(*tasks)

    elapsed = time.perf_counter() - start
    print(f"[GOOD] Done in {elapsed:.3f}s — ran 4 tasks in parallel via threads")
    for r in results:
        print(f"  {r}")


async def good_asyncio_to_thread_example() -> None:
    """asyncio.to_thread (Python 3.9+) — cleaner syntax for the same pattern."""
    print("\n[BEST] Using asyncio.to_thread for legacy sync functions...")
    start = time.perf_counter()

    # to_thread wraps the sync function in the default executor automatically.
    # This is the idiomatic way to call synchronous code from async code.
    tasks = [
        asyncio.to_thread(cpu_bound_resize, img_id)
        for img_id in range(5)
    ]
    results = await asyncio.gather(*tasks)

    elapsed = time.perf_counter() - start
    print(f"[BEST] Done in {elapsed:.3f}s")
    for r in results:
        print(f"  {r}")


async def demonstrate_slow_callback_detection() -> None:
    """Enable asyncio debug mode to catch blocking calls automatically."""
    loop = asyncio.get_running_loop()
    loop.slow_callback_duration = 0.05  # warn if any callback takes > 50ms
    # In production: set PYTHONASYNCIODEBUG=1 environment variable OR
    # asyncio.run(main(), debug=True)
    # Python will log: 'Executing <Task ...> took 0.123 seconds'
    print("Slow callback detection threshold set to 50ms")


if __name__ == "__main__":
    asyncio.run(bad_blocking_example())
    asyncio.run(good_thread_executor_example())
    asyncio.run(good_asyncio_to_thread_example())
    asyncio.run(demonstrate_slow_callback_detection())