Python asyncio Deep Dive: Event Loops, Coroutines & Real-World Concurrency

import asyncio
import time

# -------------------------------------------------------------------
# Demonstrates the event loop tick mechanism and cooperative yielding.
# Run this and watch how two coroutines interleave on a SINGLE thread.
# -------------------------------------------------------------------

async def fetch_user_profile(user_id: int) -> dict:
    """Simulates a database round-trip with asyncio.sleep."""
    print(f"[{time.perf_counter():.3f}s] Starting fetch for user {user_id}")

    # 'await' yields control back to the event loop here.
    # The loop immediately runs other ready coroutines while we 'wait'.
    await asyncio.sleep(1.0)  # represents a 1-second DB query

    print(f"[{time.perf_counter():.3f}s] Finished fetch for user {user_id}")
    return {"id": user_id, "name": f"User_{user_id}"}


async def send_welcome_email(user_id: int) -> None:
    """Simulates an outbound SMTP call."""
    print(f"[{time.perf_counter():.3f}s] Sending email to user {user_id}")
    await asyncio.sleep(0.5)  # represents a 500ms SMTP handshake
    print(f"[{time.perf_counter():.3f}s] Email sent to user {user_id}")


async def main() -> None:
    start = time.perf_counter()

    # asyncio.gather schedules BOTH coroutines as Tasks immediately.
    # The event loop runs them concurrently on the single thread.
    profile, _ = await asyncio.gather(
        fetch_user_profile(user_id=42),
        send_welcome_email(user_id=42),
    )

    elapsed = time.perf_counter() - start
    print(f"\nAll done in {elapsed:.3f}s — not 1.5s, because they ran concurrently.")
    print(f"Profile retrieved: {profile}")


if __name__ == "__main__":
    # asyncio.run() creates a NEW event loop, runs main(), then closes the loop.
    # Never call this inside an already-running loop (e.g. Jupyter) — use await directly.
    asyncio.run(main())

import asyncio
import time

# -------------------------------------------------------------------
# Side-by-side comparison: sequential coroutine awaiting vs concurrent Tasks.
# This is the most common performance mistake in async Python code.
# -------------------------------------------------------------------

async def query_database(query_name: str, latency_seconds: float) -> str:
    """Simulates a database query with configurable latency."""
    await asyncio.sleep(latency_seconds)
    return f"Result of '{query_name}'"


async def sequential_approach() -> None:
    """BAD: awaiting coroutines directly runs them one after another."""
    start = time.perf_counter()

    # Each await BLOCKS here until the previous one finishes.
    # Total time = 0.3 + 0.4 + 0.2 = 0.9 seconds
    result_orders = await query_database("orders", latency_seconds=0.3)
    result_users  = await query_database("users",  latency_seconds=0.4)
    result_prefs  = await query_database("prefs",  latency_seconds=0.2)

    elapsed = time.perf_counter() - start
    print(f"Sequential: {elapsed:.3f}s — {result_orders}, {result_users}, {result_prefs}")


async def concurrent_approach() -> None:
    """GOOD: create_task() schedules all three immediately."""
    start = time.perf_counter()

    # All three tasks are scheduled NOW and run concurrently.
    # The event loop interleaves them — total time ≈ max(0.3, 0.4, 0.2) = 0.4 seconds
    task_orders = asyncio.create_task(query_database("orders", latency_seconds=0.3))
    task_users  = asyncio.create_task(query_database("users",  latency_seconds=0.4))
    task_prefs  = asyncio.create_task(query_database("prefs",  latency_seconds=0.2))

    # Now we collect the results — tasks were ALREADY running while we did this
    result_orders = await task_orders
    result_users  = await task_users
    result_prefs  = await task_prefs

    elapsed = time.perf_counter() - start
    print(f"Concurrent:  {elapsed:.3f}s — {result_orders}, {result_users}, {result_prefs}")


async def gather_approach() -> None:
    """BEST for fire-and-collect: asyncio.gather is cleaner than manual tasks."""
    start = time.perf_counter()

    # gather() wraps each coroutine in a Task automatically and collects results
    # in the SAME ORDER as the arguments, regardless of completion order.
    results = await asyncio.gather(
        query_database("orders", latency_seconds=0.3),
        query_database("users",  latency_seconds=0.4),
        query_database("prefs",  latency_seconds=0.2),
    )

    elapsed = time.perf_counter() - start
    print(f"Gather:      {elapsed:.3f}s — {results}")


async def main() -> None:
    await sequential_approach()
    await concurrent_approach()
    await gather_approach()


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

import asyncio
import logging

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
logger = logging.getLogger(__name__)

# -------------------------------------------------------------------
# Production patterns: timeouts, cancellation, and silent failure prevention.
# These patterns separate hobby code from production-grade async services.
# -------------------------------------------------------------------

async def slow_third_party_api(endpoint: str) -> dict:
    """Simulates an API that sometimes takes way too long."""
    await asyncio.sleep(5.0)  # this would block us for 5 seconds
    return {"endpoint": endpoint, "data": "some payload"}


async def fetch_with_timeout(endpoint: str, timeout_seconds: float) -> dict | None:
    """
    Wraps any coroutine with a hard timeout.
    Returns None on timeout instead of crashing the whole request.
    """
    try:
        # wait_for cancels the inner coroutine automatically on timeout
        result = await asyncio.wait_for(
            slow_third_party_api(endpoint),
            timeout=timeout_seconds
        )
        return result
    except asyncio.TimeoutError:
        logger.warning("API call to '%s' timed out after %ss", endpoint, timeout_seconds)
        return None  # caller gets None, not a crash


async def resource_holding_task(resource_name: str) -> str:
    """
    Demonstrates CORRECT cancellation handling.
    Cleans up resources even when cancelled mid-execution.
    """
    logger.info("Acquired resource: %s", resource_name)
    try:
        await asyncio.sleep(10.0)  # long operation
        return f"Finished with {resource_name}"
    except asyncio.CancelledError:
        # Clean up before propagating — this is MANDATORY
        logger.info("Cancelled! Releasing resource: %s", resource_name)
        raise  # ALWAYS re-raise CancelledError — never swallow it
    finally:
        # finally runs whether cancelled OR completed normally
        logger.info("Cleanup complete for: %s", resource_name)


def log_task_exception(task: asyncio.Task) -> None:
    """
    Done callback that surfaces exceptions from fire-and-forget tasks.
    Attach this to any task you don't plan to await directly.
    """
    if task.cancelled():
        return  # cancellation is intentional, not an error
    exc = task.exception()
    if exc is not None:
        # Without this, the exception would be silently swallowed
        logger.error("Task '%s' failed with exception: %r", task.get_name(), exc)


async def main() -> None:
    # --- Pattern 1: Timeout ---
    logger.info("=== Pattern 1: Timeout ===")
    result = await fetch_with_timeout("/api/recommendations", timeout_seconds=2.0)
    logger.info("Result: %s", result)  # None — timed out

    # --- Pattern 2: Cancellation with cleanup ---
    logger.info("\n=== Pattern 2: Cancellation ===")
    db_task = asyncio.create_task(
        resource_holding_task("postgres_connection"),
        name="db-task"
    )
    await asyncio.sleep(0.1)  # let the task start
    db_task.cancel()           # inject CancelledError at the next await inside db_task

    try:
        await db_task  # wait for cancellation to complete fully
    except asyncio.CancelledError:
        logger.info("db_task cancellation confirmed")

    # --- Pattern 3: Fire-and-forget with exception surfacing ---
    logger.info("\n=== Pattern 3: Fire-and-forget safety ===")

    async def buggy_background_job() -> None:
        await asyncio.sleep(0.1)
        raise ValueError("Something went wrong in the background")

    background_task = asyncio.create_task(
        buggy_background_job(),
        name="background-job"
    )
    # Attach callback — fires when task finishes (success OR failure)
    background_task.add_done_callback(log_task_exception)

    await asyncio.sleep(0.5)  # give the background task time to run and fail
    logger.info("Main coroutine continued unaffected")


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

import asyncio
import hashlib
import time
from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor

# -------------------------------------------------------------------
# Bridging asyncio with thread and process pools.
# Use this when you need to call blocking or CPU-heavy code
# without freezing your entire event loop.
# -------------------------------------------------------------------

def compute_password_hash(password: str, iterations: int) -> str:
    """
    CPU-bound work: PBKDF2 with high iteration count.
    This is PURE computation — no I/O — so it belongs in a process pool.
    Running this directly in a coroutine would block the event loop.
    """
    salt = b"static_salt_for_demo"
    key = hashlib.pbkdf2_hmac(
        hash_name="sha256",
        password=password.encode(),
        salt=salt,
        iterations=iterations
    )
    return key.hex()


def blocking_legacy_db_query(user_id: int) -> dict:
    """
    I/O-bound but uses a BLOCKING library (no async driver available).
    Thread pool is appropriate here — threads release the GIL during I/O.
    """
    time.sleep(0.3)  # simulate blocking DB call via synchronous driver
    return {"user_id": user_id, "email": f"user{user_id}@example.com"}


async def handle_user_registration(username: str, raw_password: str) -> dict:
    """
    Realistic registration handler that needs both:
    - a blocking legacy DB query (thread pool)
    - CPU-intensive password hashing (process pool)
    """
    loop = asyncio.get_running_loop()

    # Thread pool: right tool for blocking I/O with sync libraries
    with ThreadPoolExecutor(max_workers=4) as thread_pool:
        user_record = await loop.run_in_executor(
            thread_pool,
            blocking_legacy_db_query,
            42  # user_id
        )

    # Process pool: right tool for CPU-bound work (bypasses the GIL)
    with ProcessPoolExecutor(max_workers=2) as process_pool:
        password_hash = await loop.run_in_executor(
            process_pool,
            compute_password_hash,
            raw_password,
            200_000  # high iteration count for security
        )

    return {
        "username": username,
        "email": user_record["email"],
        "password_hash": password_hash[:16] + "...",  # truncated for display
        "status": "registered"
    }


async def main() -> None:
    start = time.perf_counter()

    # While these run, the event loop is FREE to handle other requests
    result = await handle_user_registration(
        username="alice",
        raw_password="super_secret_passphrase_123"
    )

    elapsed = time.perf_counter() - start
    print(f"Registration complete in {elapsed:.3f}s")
    print(f"Result: {result}")


if __name__ == "__main__":
    # ProcessPoolExecutor requires if __name__ == '__main__' guard on Windows/macOS
    asyncio.run(main())