Taming Callback Hell: Practical asyncio Patterns for Production Python

import asyncio

async def background_task():
    await asyncio.sleep(1)
    raise ValueError("This error disappears into the void")

async def main():
    # Fire and forget - exception won't surface until GC
    asyncio.create_task(background_task())

    # Main continues, unaware of the failure
    await asyncio.sleep(5)
    print("Main completed - did you notice the error?")

asyncio.run(main())

import asyncio
import logging
import traceback
import sys

logging.basicConfig(level=logging.ERROR)
logger = logging.getLogger(__name__)

def handle_exception(loop: asyncio.AbstractEventLoop, context: dict) -> None:
    """Global handler for unhandled exceptions in asyncio tasks.

    This handler catches:
    - Exceptions from tasks that were never awaited
    - Exceptions from callbacks scheduled with call_soon/call_later
    - Exceptions from signal handlers
    """
    exception = context.get("exception")
    message = context.get("message", "Unhandled exception in event loop")

    # Extract task information if available
    task = context.get("task")
    task_name = task.get_name() if task else "unknown"

    if exception:
        # Log with full traceback for debugging
        tb_str = ''.join(traceback.format_exception(
            type(exception), exception, exception.__traceback__
        ))
        logger.error(
            f"Task '{task_name}' failed: {message}\n"
            f"Exception: {exception}\n"
            f"Traceback:\n{tb_str}"
        )

        # In production, you'd also want to:
        # - Send to error tracking (Sentry, Rollbar, etc.)
        # - Increment error metrics
        # - Potentially trigger alerts for critical failures
    else:
        logger.error(f"Event loop error in task '{task_name}': {message}")

async def main():
    loop = asyncio.get_running_loop()
    loop.set_exception_handler(handle_exception)

    # Now fire-and-forget tasks will log errors immediately
    asyncio.create_task(some_background_work())
    await asyncio.sleep(10)

asyncio.run(main())

import asyncio
import os

# Enable via environment variable (useful for all processes)
os.environ['PYTHONASYNCIODEBUG'] = '1'

# Or enable programmatically for specific runs
async def main():
    # Your application code
    pass

asyncio.run(main(), debug=True)

import asyncio
from typing import List

async def fetch_user(user_id: int) -> dict:
    """Simulate fetching user data from an API."""
    await asyncio.sleep(0.1)  # Simulated network latency
    if user_id == 3:
        raise ValueError(f"User {user_id} not found")
    return {"id": user_id, "name": f"User_{user_id}"}

async def fetch_all_users(user_ids: List[int]) -> List[dict]:
    """Fetch multiple users concurrently with automatic cleanup.

    If any fetch fails, all other fetches are cancelled and
    resources are cleaned up before the exception propagates.
    """
    results = []

    async with asyncio.TaskGroup() as tg:
        tasks = [tg.create_task(fetch_user(uid)) for uid in user_ids]

    # Only reached if ALL tasks succeed
    # If any task raises, the exception propagates here
    return [task.result() for task in tasks]

async def main():
    try:
        users = await fetch_all_users([1, 2, 3, 4, 5])
        print(f"Fetched {len(users)} users")
    except ExceptionGroup as eg:
        # TaskGroup wraps exceptions in ExceptionGroup
        # This allows handling multiple simultaneous failures
        for exc in eg.exceptions:
            print(f"Task failed: {exc}")

asyncio.run(main())

import asyncio

async def might_fail(task_id: int) -> str:
    if task_id % 2 == 0:
        raise ValueError(f"Even task {task_id} failed")
    if task_id % 3 == 0:
        raise RuntimeError(f"Task {task_id} had a runtime error")
    return f"Task {task_id} succeeded"

async def main():
    try:
        async with asyncio.TaskGroup() as tg:
            for i in range(6):
                tg.create_task(might_fail(i))
    except* ValueError as eg:
        print(f"Caught {len(eg.exceptions)} ValueErrors:")
        for exc in eg.exceptions:
            print(f"  - {exc}")
    except* RuntimeError as eg:
        print(f"Caught {len(eg.exceptions)} RuntimeErrors:")
        for exc in eg.exceptions:
            print(f"  - {exc}")

asyncio.run(main())

import asyncio
from contextlib import asynccontextmanager
from typing import Set, Callable, Coroutine, Any

@asynccontextmanager
async def managed_tasks():
    """Context manager for structured concurrency in Python <3.11.

    All tasks created through the yielded function are guaranteed
    to be cancelled when the context exits, whether normally or
    due to an exception.
    """
    tasks: Set[asyncio.Task] = set()

    def create_task(coro: Coroutine[Any, Any, Any]) -> asyncio.Task:
        task = asyncio.create_task(coro)
        tasks.add(task)
        task.add_done_callback(tasks.discard)
        return task

    try:
        yield create_task
    finally:
        # Cancel all pending tasks on exit
        for task in tasks:
            if not task.done():
                task.cancel()

        # Wait for cancellations to complete
        # return_exceptions=True prevents CancelledError from propagating
        if tasks:
            await asyncio.gather(*tasks, return_exceptions=True)

async def worker(name: str, duration: float) -> str:
    """Example worker that simulates some async work."""
    await asyncio.sleep(duration)
    return f"{name} completed after {duration}s"

async def main():
    async with managed_tasks() as create:
        t1 = create(worker("A", 1.0))
        t2 = create(worker("B", 2.0))
        t3 = create(worker("C", 0.5))

        # Wait for first to complete, others auto-cancel on exit
        done, pending = await asyncio.wait(
            [t1, t2, t3],
            return_when=asyncio.FIRST_COMPLETED
        )
        print(f"First completed: {done.pop().result()}")
    # All remaining tasks cancelled and awaited here

asyncio.run(main())

import asyncio

async def slow_operation() -> str:
    """An operation that takes longer than we want to wait."""
    await asyncio.sleep(10)
    return "completed"

async def main():
    try:
        # wait_for cancels the operation and raises TimeoutError
        result = await asyncio.wait_for(slow_operation(), timeout=2.0)
        print(f"Got result: {result}")
    except TimeoutError:
        print("Operation timed out after 2 seconds")
        # The slow_operation coroutine received CancelledError
        # and had a chance to clean up

asyncio.run(main())

import asyncio

async def fetch_data(url: str) -> dict:
    """Simulate fetching data from an external service."""
    await asyncio.sleep(1)
    return {"url": url, "data": "response payload"}

async def validate_data(data: dict) -> bool:
    """Simulate validating the fetched data."""
    await asyncio.sleep(0.3)
    return True

async def process_data(data: dict) -> dict:
    """Simulate processing the validated data."""
    await asyncio.sleep(0.5)
    return {"processed": True, **data}

async def main():
    try:
        async with asyncio.timeout(2.0):
            # All three operations share the same 2-second budget
            # If fetch takes 1.5s, process only has 0.5s remaining
            data = await fetch_data("https://api.example.com")
            is_valid = await validate_data(data)
            if is_valid:
                result = await process_data(data)
                print(f"Result: {result}")
    except TimeoutError:
        print("Combined operations exceeded 2 second budget")

asyncio.run(main())

import asyncio

async def coordinated_operations():
    """Multiple operations that must complete by the same deadline."""
    loop = asyncio.get_running_loop()
    deadline = loop.time() + 5.0  # 5 seconds from now

    async with asyncio.timeout_at(deadline):
        result1 = await operation_one()

    # Same deadline for the second operation
    async with asyncio.timeout_at(deadline):
        result2 = await operation_two()

    return result1, result2

import asyncio

async def acquire_resource():
    """Simulate acquiring a resource like a database connection."""
    return type('Resource', (), {'close': lambda self: asyncio.sleep(0)})()

async def process_next_item(resource):
    """Simulate processing work."""
    await asyncio.sleep(0.1)

async def cleanup_on_cancel():
    """Demonstrates proper cancellation handling with cleanup.

    The key insight: you can perform cleanup operations during
    cancellation, but you MUST re-raise CancelledError when done.
    """
    resource = await acquire_resource()
    try:
        while True:
            await process_next_item(resource)
    except asyncio.CancelledError:
        # Perform cleanup, then re-raise
        # This gives you a chance to release resources
        await resource.close()
        print("Resource closed during cancellation")
        raise  # CRITICAL: Always re-raise CancelledError

async def bad_cancellation_handling():
    """DON'T DO THIS - swallowing CancelledError breaks cancellation."""
    try:
        await some_operation()
    except asyncio.CancelledError:
        print("Cancelled, but I'm going to keep running anyway")
        # Missing raise! This task is now a zombie

import asyncio
import signal

async def long_running_worker():
    """A worker that processes items indefinitely."""
    while True:
        await asyncio.sleep(1)
        print("Processing...")

async def graceful_shutdown():
    """Handle SIGTERM/SIGINT for clean shutdown."""
    loop = asyncio.get_running_loop()
    shutdown_event = asyncio.Event()

    def signal_handler():
        print("Shutdown signal received")
        shutdown_event.set()

    # Register signal handlers
    for sig in (signal.SIGTERM, signal.SIGINT):
        loop.add_signal_handler(sig, signal_handler)

    # Start background tasks
    worker_task = asyncio.create_task(long_running_worker())

    # Wait for shutdown signal
    await shutdown_event.wait()

    # Graceful cancellation with timeout
    worker_task.cancel()
    try:
        # Give the task time to clean up
        await asyncio.wait_for(worker_task, timeout=5.0)
    except asyncio.CancelledError:
        pass  # Expected after cancellation
    except TimeoutError:
        print("Worker didn't shut down cleanly within timeout")

    print("Shutdown complete")

import asyncio

async def critical_write(data: dict) -> None:
    """This write must complete even if the parent is cancelled.

    Examples: database commits, message queue acknowledgments,
    audit log entries, billing records.
    """
    await database.commit(data)
    await message_queue.ack()

async def process_message(msg: dict) -> None:
    """Process a message with a critical final step."""
    result = await compute_result(msg)

    # Shield the critical write from cancellation
    # Even if process_message is cancelled, the write continues
    await asyncio.shield(critical_write(result))

[tool.pytest.ini_options]
asyncio_mode = "auto"
asyncio_default_fixture_loop_scope = "function"
filterwarnings = [
    "error::DeprecationWarning",
    "error::PendingDeprecationWarning",
]

import asyncio
import pytest
from unittest.mock import AsyncMock, patch, MagicMock

from myapp.service import UserService

@pytest.fixture
async def user_service():
    """Async fixture that handles connection lifecycle.

    The service connects before the test runs and
    disconnects after, even if the test fails.
    """
    service = UserService()
    await service.connect()
    yield service
    await service.disconnect()

@pytest.fixture
async def mock_database():
    """Fixture providing a mock database connection."""
    mock_db = AsyncMock()
    mock_db.query = AsyncMock(return_value=[{"id": 1, "name": "Test User"}])
    mock_db.execute = AsyncMock(return_value={"rows_affected": 1})
    return mock_db

async def test_fetch_user_success(user_service):
    """Test basic user fetch with real service fixture."""
    user = await user_service.fetch_user(123)
    assert user["id"] == 123
    assert "name" in user

async def test_fetch_user_timeout():
    """Test timeout handling with mocked slow response."""
    service = UserService()

    async def slow_fetch(*args):
        await asyncio.sleep(10)  # Longer than any reasonable timeout
        return {"id": 1}

    with patch.object(service, "_make_request", side_effect=slow_fetch):
        with pytest.raises(TimeoutError):
            await asyncio.wait_for(service.fetch_user(1), timeout=0.1)

import pytest
from unittest.mock import AsyncMock, MagicMock

async def test_database_transaction():
    """Test code that uses async context managers."""
    mock_conn = AsyncMock()
    # __aenter__ returns the connection object for 'as conn' binding
    mock_conn.__aenter__.return_value = mock_conn
    # __aexit__ returns None (no exception suppression)
    mock_conn.__aexit__.return_value = None
    mock_conn.execute = AsyncMock(return_value={"rows_affected": 1})
    mock_conn.commit = AsyncMock()
    mock_conn.rollback = AsyncMock()

    # Use the mock in your test
    async with mock_conn as conn:
        result = await conn.execute("UPDATE users SET active = true")
        assert result["rows_affected"] == 1
        await conn.commit()

    # Verify the context manager was used correctly
    mock_conn.__aenter__.assert_called_once()
    mock_conn.__aexit__.assert_called_once()
    mock_conn.commit.assert_called_once()

async def test_context_manager_error_handling():
    """Test that errors in context managers are handled."""
    mock_conn = AsyncMock()
    mock_conn.__aenter__.return_value = mock_conn
    mock_conn.__aexit__.return_value = None
    mock_conn.execute = AsyncMock(side_effect=RuntimeError("DB error"))
    mock_conn.rollback = AsyncMock()

    with pytest.raises(RuntimeError, match="DB error"):
        async with mock_conn as conn:
            await conn.execute("INVALID SQL")

import asyncio
import pytest

async def test_rate_limiter_rejects_excess():
    """Test rate limiting without actual delays.

    Instead of verifying exact timing, verify that the rate
    limiter blocks when it should and allows when it should.
    """
    from myapp.rate_limiter import RateLimiter

    limiter = RateLimiter(max_requests=2, window_seconds=1.0)

    # First two requests succeed immediately
    assert await limiter.acquire() is True
    assert await limiter.acquire() is True

    # Third request would wait - verify with short timeout
    with pytest.raises(TimeoutError):
        await asyncio.wait_for(limiter.acquire(), timeout=0.01)

async def test_retry_mechanism():
    """Test retry logic with controlled failures."""
    attempt_count = 0

    async def flaky_operation():
        nonlocal attempt_count
        attempt_count += 1
        if attempt_count < 3:
            raise ConnectionError("Temporary failure")
        return "success"

    from myapp.retry import with_retry
    result = await with_retry(flaky_operation, max_attempts=5)

    assert result == "success"
    assert attempt_count == 3  # Succeeded on third attempt

import asyncio
import hashlib
from pathlib import Path

def compute_file_hash(filepath: Path) -> str:
    """CPU-bound synchronous function.

    File I/O and hash computation are both blocking operations
    that would freeze the event loop if called directly.
    """
    hasher = hashlib.sha256()
    with open(filepath, "rb") as f:
        for chunk in iter(lambda: f.read(8192), b""):
            hasher.update(chunk)
    return hasher.hexdigest()

async def process_files(filepaths: list[Path]) -> dict[str, str]:
    """Process multiple files without blocking the event loop.

    Each file hash runs in its own thread, allowing
    concurrent processing without blocking async operations.
    """
    tasks = [
        asyncio.to_thread(compute_file_hash, fp)
        for fp in filepaths
    ]

    results = await asyncio.gather(*tasks)
    return dict(zip(filepaths, results))

async def main():
    files = list(Path("/var/log").glob("*.log"))[:10]
    hashes = await process_files(files)
    for path, hash_value in hashes.items():
        print(f"{path.name}: {hash_value[:16]}...")

asyncio.run(main())

import asyncio
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor

# Create dedicated executors for different workloads
io_executor = ThreadPoolExecutor(
    max_workers=20,
    thread_name_prefix="io_"
)
cpu_executor = ProcessPoolExecutor(
    max_workers=4  # For CPU-bound work, use processes
)

async def legacy_integration():
    """Integrate with synchronous library using explicit executor."""
    loop = asyncio.get_running_loop()

    # Run sync function in thread pool
    result = await loop.run_in_executor(
        io_executor,
        blocking_library_call,
        arg1, arg2
    )
    return result

async def parallel_cpu_work(items: list) -> list:
    """CPU-bound work benefits from ProcessPoolExecutor."""
    loop = asyncio.get_running_loop()

    # Each item processed in a separate process
    tasks = [
        loop.run_in_executor(cpu_executor, cpu_intensive_function, item)
        for item in items
    ]
    return await asyncio.gather(*tasks)

import asyncio
import requests  # Synchronous HTTP library - DO NOT USE IN ASYNC CODE

async def bad_fetch(url: str) -> str:
    # WRONG: blocks the event loop for every request
    # All other concurrent operations freeze during this call
    response = requests.get(url)
    return response.text

async def acceptable_fetch(url: str) -> str:
    # ACCEPTABLE: offload to thread if you must use sync library
    # Still has thread overhead and pool contention
    response = await asyncio.to_thread(requests.get, url)
    return response.text

# BEST: use an async HTTP library designed for asyncio
import aiohttp

async def best_fetch(url: str) -> str:
    # RIGHT: fully async, no blocking, efficient connection pooling
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()

import asyncio
import logging
import uuid
from contextvars import ContextVar
from typing import Optional

# Context variables for request tracking
request_id: ContextVar[str] = ContextVar("request_id", default="no-request")
user_id: ContextVar[Optional[int]] = ContextVar("user_id", default=None)

class RequestContextFilter(logging.Filter):
    """Inject request context into all log records.

    This filter runs for every log message and adds context
    variables as record attributes for the formatter.
    """

    def filter(self, record: logging.LogRecord) -> bool:
        record.request_id = request_id.get()
        record.user_id = user_id.get() or "anonymous"
        return True

# Configure logging with context
handler = logging.StreamHandler()
handler.setFormatter(logging.Formatter(
    "%(asctime)s [%(request_id)s] user=%(user_id)s %(levelname)s: %(message)s"
))
handler.addFilter(RequestContextFilter())

logger = logging.getLogger(__name__)
logger.addHandler(handler)
logger.setLevel(logging.INFO)

async def handle_request(data: dict) -> dict:
    """Process a request with context-aware logging.

    All log messages within this function and any functions
    it calls will include the request ID automatically.
    """
    # Set context for this request
    token = request_id.set(str(uuid.uuid4())[:8])
    user_token = user_id.set(data.get("user_id"))

    try:
        logger.info(f"Processing request: {data.get('action')}")

        # These nested calls inherit the context
        result = await validate_input(data)
        result = await process_data(result)

        logger.info(f"Request completed successfully")
        return result
    except Exception as e:
        logger.error(f"Request failed: {e}")
        raise
    finally:
        # Reset context when request completes
        request_id.reset(token)
        user_id.reset(user_token)

import asyncio
import logging

# Enable asyncio debug logging
logging.basicConfig(level=logging.DEBUG)
asyncio_logger = logging.getLogger("asyncio")
asyncio_logger.setLevel(logging.WARNING)  # Only show problems

async def main():
    loop = asyncio.get_running_loop()

    # Customize slow callback threshold (default is 0.1 seconds)
    loop.slow_callback_duration = 0.05  # 50ms - more aggressive detection

    # Your application code here
    await run_server()

# Run with debug mode enabled
asyncio.run(main(), debug=True)

from opentelemetry import trace
from opentelemetry.instrumentation.asyncio import AsyncioInstrumentor

# Enable asyncio instrumentation for automatic span propagation
AsyncioInstrumentor().instrument()

tracer = trace.get_tracer(__name__)

async def traced_operation(item_id: int) -> dict:
    """Operation with explicit span for detailed tracing."""
    with tracer.start_as_current_span("process_item") as span:
        span.set_attribute("item.id", item_id)

        result = await fetch_item(item_id)
        span.set_attribute("item.status", result["status"])

        if result["status"] == "error":
            span.set_status(trace.Status(trace.StatusCode.ERROR))

        return result