Asyncio Concurrency Patterns

Asyncio Concurrency Patterns

A comprehensive skill for mastering Python's asyncio library and concurrent programming patterns. This skill covers event loops, coroutines, tasks, futures, synchronization primitives, async context managers, and production-ready patterns for building high-performance asynchronous applications.

When to Use This Skill

Use this skill when:

• Building I/O-bound applications that need to handle many concurrent operations
• Creating web servers, API clients, or websocket applications
• Implementing real-time systems with event-driven architecture
• Optimizing application performance with concurrent request handling
• Managing multiple async operations with proper coordination and error handling
• Building background task processors or job queues
• Implementing async database operations and connection pooling
• Creating chat applications, real-time dashboards, or notification systems
• Handling parallel HTTP requests efficiently
• Managing websocket connections with multiple event sources
• Building microservices with async communication patterns
• Optimizing resource utilization in network applications

Core Concepts

What is Asyncio?

Asyncio is Python's built-in library for writing concurrent code using the async/await syntax. It provides:

• Event Loop: The core of asyncio that schedules and runs asynchronous tasks
• Coroutines: Functions defined with async def that can be paused and resumed
• Tasks: Scheduled coroutines that run concurrently
• Futures: Low-level objects representing results of async operations
• Synchronization Primitives: Locks, semaphores, events for coordination

Event Loop Fundamentals

The event loop is the central execution mechanism in asyncio:

import asyncio

# Get or create an event loop
loop = asyncio.get_event_loop()

# Run a coroutine until complete
loop.run_until_complete(my_coroutine())

# Modern approach (Python 3.7+)
asyncio.run(my_coroutine())

Key Event Loop Concepts:

1. Single-threaded concurrency: One thread, many tasks 2. Cooperative multitasking: Tasks yield control voluntarily 3. I/O multiplexing: Efficient handling of many I/O operations 4. Non-blocking operations: Don't wait for I/O, do other work

Coroutines vs Functions

Regular Function:

def fetch_data():
    # Blocks until complete
    return requests.get('http://api.example.com')

Coroutine:

async def fetch_data():
    # Yields control while waiting
    async with aiohttp.ClientSession() as session:
        async with session.get('http://api.example.com') as resp:
            return await resp.text()

Tasks and Futures

Tasks wrap coroutines and schedule them on the event loop:

# Create a task
task = asyncio.create_task(my_coroutine())

# Task runs in background
# ... do other work ...

# Wait for result
result = await task

Futures represent eventual results:

# Low-level future (rarely used directly)
future = asyncio.Future()

# Set result
future.set_result(42)

# Get result
result = await future

Async Context Managers

Manage resources with async setup/teardown:

class AsyncResource:
    async def __aenter__(self):
        # Async setup
        await self.connect()
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        # Async cleanup
        await self.disconnect()

# Usage
async with AsyncResource() as resource:
    await resource.do_work()

Concurrency Patterns

Pattern 1: Gather - Concurrent Execution

Run multiple coroutines concurrently and wait for all to complete:

import asyncio
import aiohttp

async def fetch(session, url):
    async with session.get(url) as response:
        return await response.text()

async def main():
    async with aiohttp.ClientSession() as session:
        # Run all fetches concurrently
        results = await asyncio.gather(
            fetch(session, 'http://python.org'),
            fetch(session, 'http://docs.python.org'),
            fetch(session, 'http://pypi.org')
        )
        return results

# Results is a list in the same order as inputs
results = asyncio.run(main())

When to use:

• Need all results
• Order matters
• Want to fail fast on first exception (default)
• Can handle partial results with return_exceptions=True

Pattern 2: Wait - Flexible Waiting

More control over how to wait for multiple tasks:

import asyncio

async def task_a():
    await asyncio.sleep(2)
    return 'A'

async def task_b():
    await asyncio.sleep(1)
    return 'B'

async def main():
    tasks = [
        asyncio.create_task(task_a()),
        asyncio.create_task(task_b())
    ]

    # Wait for first to complete
    done, pending = await asyncio.wait(
        tasks,
        return_when=asyncio.FIRST_COMPLETED
    )

    # Get first result
    first_result = done.pop().result()

    # Cancel remaining
    for task in pending:
        task.cancel()

    return first_result

result = asyncio.run(main())  # Returns 'B' after 1 second

Wait strategies:

• FIRST_COMPLETED: Return when first task finishes
• FIRST_EXCEPTION: Return when first task raises exception
• ALL_COMPLETED: Wait for all tasks (default)

Pattern 3: Semaphore - Limit Concurrency

Control maximum number of concurrent operations:

import asyncio
import aiohttp

async def fetch_with_limit(session, url, semaphore):
    async with semaphore:
        # Only N requests run concurrently
        async with session.get(url) as resp:
            return await resp.text()

async def main():
    # Limit to 5 concurrent requests
    semaphore = asyncio.Semaphore(5)

    urls = [f'http://api.example.com/item/{i}' for i in range(100)]

    async with aiohttp.ClientSession() as session:
        tasks = [
            fetch_with_limit(session, url, semaphore)
            for url in urls
        ]
        results = await asyncio.gather(*tasks)

    return results

asyncio.run(main())

When to use:

• Rate limiting API requests
• Controlling database connection usage
• Preventing resource exhaustion
• Respecting external service limits

Pattern 4: Lock - Mutual Exclusion

Ensure only one coroutine accesses a resource at a time:

import asyncio

class SharedCounter:
    def __init__(self):
        self.value = 0
        self.lock = asyncio.Lock()

    async def increment(self):
        async with self.lock:
            # Critical section - only one coroutine at a time
            current = self.value
            await asyncio.sleep(0)  # Simulate async work
            self.value = current + 1

async def worker(counter):
    for _ in range(100):
        await counter.increment()

async def main():
    counter = SharedCounter()

    # Run 10 workers concurrently
    await asyncio.gather(*[worker(counter) for _ in range(10)])

    print(f"Final count: {counter.value}")  # Always 1000

asyncio.run(main())

Pattern 5: Event - Signaling

Coordinate multiple coroutines with events:

import asyncio

async def waiter(event, name):
    print(f'{name} waiting for event')
    await event.wait()
    print(f'{name} received event')

async def setter(event):
    await asyncio.sleep(2)
    print('Setting event')
    event.set()

async def main():
    event = asyncio.Event()

    # Multiple waiters
    await asyncio.gather(
        waiter(event, 'Waiter 1'),
        waiter(event, 'Waiter 2'),
        waiter(event, 'Waiter 3'),
        setter(event)
    )

asyncio.run(main())

Pattern 6: Queue - Producer/Consumer

Coordinate work between producers and consumers:

import asyncio

async def producer(queue, n):
    for i in range(n):
        await asyncio.sleep(0.1)
        await queue.put(f'item-{i}')
        print(f'Produced item-{i}')

    # Signal completion
    await queue.put(None)

async def consumer(queue, name):
    while True:
        item = await queue.get()

        if item is None:
            # Propagate sentinel to other consumers
            await queue.put(None)
            break

        print(f'{name} processing {item}')
        await asyncio.sleep(0.2)
        queue.task_done()

async def main():
    queue = asyncio.Queue()

    # Start producer and consumers
    await asyncio.gather(
        producer(queue, 10),
        consumer(queue, 'Consumer-1'),
        consumer(queue, 'Consumer-2'),
        consumer(queue, 'Consumer-3')
    )

asyncio.run(main())

Task Management

Creating Tasks

Basic Task Creation:

import asyncio

async def background_task():
    await asyncio.sleep(10)
    return 'Done'

async def main():
    # Create task - starts running immediately
    task = asyncio.create_task(background_task())

    # Do other work while task runs
    await asyncio.sleep(1)

    # Wait for result
    result = await task
    return result

asyncio.run(main())

Named Tasks (Python 3.8+):

task = asyncio.create_task(
    background_task(),
    name='my-background-task'
)

print(task.get_name())  # 'my-background-task'

Task Cancellation

Graceful Cancellation:

import asyncio

async def long_running_task():
    try:
        while True:
            await asyncio.sleep(1)
            print('Working...')
    except asyncio.CancelledError:
        print('Task cancelled, cleaning up...')
        # Cleanup logic
        raise  # Re-raise to mark as cancelled

async def main():
    task = asyncio.create_task(long_running_task())

    # Let it run for 3 seconds
    await asyncio.sleep(3)

    # Request cancellation
    task.cancel()

    try:
        await task
    except asyncio.CancelledError:
        print('Task was cancelled')

asyncio.run(main())

Cancellation with Context Manager:

import asyncio
from contextlib import suppress

async def run_with_timeout():
    task = asyncio.create_task(long_running_task())

    try:
        # Wait with timeout
        await asyncio.wait_for(task, timeout=5.0)
    except asyncio.TimeoutError:
        task.cancel()
        with suppress(asyncio.CancelledError):
            await task

Exception Handling in Tasks

Gather with Exception Handling:

import asyncio

async def failing_task(n):
    await asyncio.sleep(n)
    raise ValueError(f'Task {n} failed')

async def successful_task(n):
    await asyncio.sleep(n)
    return f'Task {n} succeeded'

async def main():
    # return_exceptions=True: Returns exceptions instead of raising
    results = await asyncio.gather(
        successful_task(1),
        failing_task(2),
        successful_task(3),
        return_exceptions=True
    )

    for i, result in enumerate(results):
        if isinstance(result, Exception):
            print(f'Task {i} failed: {result}')
        else:
            print(f'Task {i} result: {result}')

asyncio.run(main())

Task Exception Retrieval:

import asyncio

async def main():
    task = asyncio.create_task(failing_task(1))

    # Wait for task
    await asyncio.sleep(2)

    # Check if task failed
    if task.done() and task.exception():
        print(f'Task failed with: {task.exception()}')

asyncio.run(main())

Event Loop Management

Event Loop Policies

Default Event Loop:

import asyncio

async def main():
    # Get running loop
    loop = asyncio.get_running_loop()
    print(f'Loop: {loop}')

asyncio.run(main())

Custom Event Loop:

import asyncio

async def main():
    pass

# Create new event loop
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)

try:
    loop.run_until_complete(main())
finally:
    loop.close()

Event Loop Best Practices:

1. Use `asyncio.run()` for simple programs (Python 3.7+) 2. Avoid creating ClientSession outside event loop 3. Always close loops when done 4. Don't call blocking functions in event loop

Running Blocking Code

Using ThreadPoolExecutor:

import asyncio
import time
from concurrent.futures import ThreadPoolExecutor

def blocking_io():
    # Blocking operation
    time.sleep(2)
    return 'Done'

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

    # Run blocking code in thread pool
    result = await loop.run_in_executor(
        None,  # Use default executor
        blocking_io
    )

    return result

asyncio.run(main())

Custom Executor:

import asyncio
from concurrent.futures import ThreadPoolExecutor

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

    # Custom executor with 4 threads
    with ThreadPoolExecutor(max_workers=4) as executor:
        results = await asyncio.gather(*[
            loop.run_in_executor(executor, blocking_io)
            for _ in range(10)
        ])

    return results

asyncio.run(main())

Loop Callbacks

Schedule Callback:

import asyncio

def callback(arg):
    print(f'Callback called with {arg}')

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

    # Schedule callback
    loop.call_soon(callback, 'immediate')

    # Schedule with delay
    loop.call_later(2, callback, 'delayed')

    # Schedule at specific time
    loop.call_at(loop.time() + 3, callback, 'scheduled')

    await asyncio.sleep(4)

asyncio.run(main())

Async Context Managers

Creating Async Context Managers

Class-Based:

import asyncio

class AsyncDatabaseConnection:
    def __init__(self, host):
        self.host = host
        self.connection = None

    async def __aenter__(self):
        print(f'Connecting to {self.host}')
        await asyncio.sleep(0.1)  # Simulate connection
        self.connection = f'Connection to {self.host}'
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        print(f'Closing connection to {self.host}')
        await asyncio.sleep(0.1)  # Simulate cleanup
        self.connection = None

    async def query(self, sql):
        if not self.connection:
            raise RuntimeError('Not connected')
        await asyncio.sleep(0.05)
        return f'Results for: {sql}'

async def main():
    async with AsyncDatabaseConnection('localhost') as db:
        result = await db.query('SELECT * FROM users')
        print(result)

asyncio.run(main())

Decorator-Based:

import asyncio
from contextlib import asynccontextmanager

@asynccontextmanager
async def async_resource(name):
    # Setup
    print(f'Acquiring {name}')
    await asyncio.sleep(0.1)

    try:
        yield name
    finally:
        # Cleanup
        print(f'Releasing {name}')
        await asyncio.sleep(0.1)

async def main():
    async with async_resource('database') as db:
        print(f'Using {db}')

asyncio.run(main())

Real-World Example: aiohttp ClientSession

import aiohttp
import asyncio

async def fetch(session, url):
    async with session.get(url) as response:
        return await response.text()

async def main():
    # ClientSession as async context manager
    async with aiohttp.ClientSession() as session:
        html = await fetch(session, 'http://python.org')
        print(f'Body: {html[:100]}...')

asyncio.run(main())

Why use async context manager for ClientSession?

1. Ensures proper cleanup of connections 2. Prevents resource leaks 3. Manages SSL connections correctly 4. Handles graceful shutdown

Performance Optimization

Profiling Async Code

Basic Timing:

import asyncio
import time

async def slow_operation():
    await asyncio.sleep(1)

async def main():
    start = time.perf_counter()

    await slow_operation()

    elapsed = time.perf_counter() - start
    print(f'Took {elapsed:.2f} seconds')

asyncio.run(main())

Profiling Multiple Operations:

import asyncio
import time

async def timed_task(name, duration):
    start = time.perf_counter()
    await asyncio.sleep(duration)
    elapsed = time.perf_counter() - start
    print(f'{name} took {elapsed:.2f}s')
    return name

async def main():
    await asyncio.gather(
        timed_task('Task 1', 1),
        timed_task('Task 2', 2),
        timed_task('Task 3', 0.5)
    )

asyncio.run(main())

Optimizing Concurrency

Bad - Sequential Execution:

async def slow_approach():
    results = []
    for i in range(10):
        result = await fetch_data(i)
        results.append(result)
    return results
# Takes 10 * fetch_time

Good - Concurrent Execution:

async def fast_approach():
    tasks = [fetch_data(i) for i in range(10)]
    results = await asyncio.gather(*tasks)
    return results
# Takes ~fetch_time

Better - Controlled Concurrency:

async def controlled_approach():
    semaphore = asyncio.Semaphore(5)  # Max 5 concurrent

    async def fetch_with_limit(i):
        async with semaphore:
            return await fetch_data(i)

    tasks = [fetch_with_limit(i) for i in range(10)]
    results = await asyncio.gather(*tasks)
    return results
# Takes ~2 * fetch_time, but respects limits

Avoiding Common Performance Pitfalls

1. Don't create sessions per request:

# BAD - Creates new session each time
async def bad_fetch(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as resp:
            return await resp.text()

# GOOD - Reuse session
async def good_fetch():
    async with aiohttp.ClientSession() as session:
        results = await asyncio.gather(
            session.get('http://example.com/1'),
            session.get('http://example.com/2'),
            session.get('http://example.com/3')
        )
        return results

2. Don't use blocking operations:

import asyncio
import requests  # Blocking library

# BAD - Blocks event loop
async def bad_request():
    response = requests.get('http://example.com')  # BLOCKS!
    return response.text

# GOOD - Use async library
async def good_request():
    async with aiohttp.ClientSession() as session:
        async with session.get('http://example.com') as resp:
            return await resp.text()

# ACCEPTABLE - If must use blocking, use executor
async def acceptable_request():
    loop = asyncio.get_running_loop()
    result = await loop.run_in_executor(
        None,
        lambda: requests.get('http://example.com').text
    )
    return result

3. Proper cleanup with zero-sleep:

async def proper_cleanup():
    async with aiohttp.ClientSession() as session:
        async with session.get('http://example.org/') as resp:
            await resp.read()

    # Zero-sleep to allow underlying connections to close
    await asyncio.sleep(0)

Common Pitfalls

Pitfall 1: Creating ClientSession Outside Event Loop

Problem:

import aiohttp

# BAD - Session created outside event loop
session = aiohttp.ClientSession()

async def fetch(url):
    async with session.get(url) as resp:
        return await resp.text()

Why it's bad:

• Session binds to event loop at creation time
• If loop changes (e.g., uvloop), session becomes invalid
• Can cause program to hang

Solution:

import aiohttp
import asyncio

async def main():
    # Create session inside async function
    async with aiohttp.ClientSession() as session:
        async with session.get('http://python.org') as resp:
            print(await resp.text())

asyncio.run(main())

Pitfall 2: Session as Class Variable

Problem:

class API:
    session = aiohttp.ClientSession()  # BAD - global instance

    async def fetch(self, url):
        async with self.session.get(url) as resp:
            return await resp.text()

Solution:

class API:
    def __init__(self):
        self.session = None

    async def __aenter__(self):
        self.session = aiohttp.ClientSession()
        return self

    async def __aexit__(self, *args):
        await self.session.close()

    async def fetch(self, url):
        async with self.session.get(url) as resp:
            return await resp.text()

# Usage
async def main():
    async with API() as api:
        result = await api.fetch('http://example.com')

Pitfall 3: Forgetting await

Problem:

async def process_data():
    # Forgot await - returns coroutine, doesn't execute!
    result = fetch_data()  # Missing await
    return result

Solution:

async def process_data():
    result = await fetch_data()  # Proper await
    return result

Pitfall 4: Blocking the Event Loop

Problem:

import asyncio
import time

async def bad_sleep():
    time.sleep(5)  # BAD - Blocks entire event loop!

async def main():
    await asyncio.gather(
        bad_sleep(),
        another_task()  # Blocked for 5 seconds
    )

Solution:

import asyncio

async def good_sleep():
    await asyncio.sleep(5)  # GOOD - Yields control

async def main():
    await asyncio.gather(
        good_sleep(),
        another_task()  # Runs concurrently
    )

Pitfall 5: Not Handling Task Cancellation

Problem:

async def bad_task():
    while True:
        await asyncio.sleep(1)
        process_data()
        # No cleanup on cancellation!

Solution:

async def good_task():
    try:
        while True:
            await asyncio.sleep(1)
            process_data()
    except asyncio.CancelledError:
        # Cleanup resources
        cleanup()
        raise  # Re-raise to mark as cancelled

Pitfall 6: Deadlocks with Locks

Problem:

import asyncio

lock1 = asyncio.Lock()
lock2 = asyncio.Lock()

async def task_a():
    async with lock1:
        await asyncio.sleep(0.1)
        async with lock2:  # Deadlock potential
            pass

async def task_b():
    async with lock2:
        await asyncio.sleep(0.1)
        async with lock1:  # Deadlock potential
            pass

Solution:

# Always acquire locks in same order
async def safe_task_a():
    async with lock1:
        async with lock2:
            pass

async def safe_task_b():
    async with lock1:  # Same order
        async with lock2:
            pass

Production Patterns

Pattern 1: Graceful Shutdown

Complete Shutdown Example:

import asyncio
import signal
from contextlib import suppress

class Application:
    def __init__(self):
        self.should_exit = False
        self.tasks = []

    async def worker(self, name):
        try:
            while not self.should_exit:
                print(f'{name} working...')
                await asyncio.sleep(1)
        except asyncio.CancelledError:
            print(f'{name} cancelled, cleaning up...')
            raise

    def handle_signal(self, sig):
        print(f'Received signal {sig}, shutting down...')
        self.should_exit = True

    async def run(self):
        # Setup signal handlers
        loop = asyncio.get_running_loop()
        for sig in (signal.SIGTERM, signal.SIGINT):
            loop.add_signal_handler(
                sig,
                lambda s=sig: self.handle_signal(s)
            )

        # Start workers
        self.tasks = [
            asyncio.create_task(self.worker(f'Worker-{i}'))
            for i in range(3)
        ]

        # Wait for shutdown signal
        while not self.should_exit:
            await asyncio.sleep(0.1)

        # Cancel all tasks
        for task in self.tasks:
            task.cancel()

        # Wait for cancellation to complete
        await asyncio.gather(*self.tasks, return_exceptions=True)

        print('Shutdown complete')

# Run application
app = Application()
asyncio.run(app.run())

Pattern 2: Background Tasks with Application Lifecycle

aiohttp Application with Background Tasks:

import asyncio
from contextlib import suppress
from aiohttp import web

async def listen_to_redis(app):
    """Background task that listens to Redis"""
    # Simulated Redis listening
    try:
        while True:
            # Process messages
            await asyncio.sleep(1)
            print('Processing Redis message...')
    except asyncio.CancelledError:
        print('Redis listener stopped')
        raise

async def background_tasks(app):
    """Cleanup context for managing background tasks"""
    # Startup: Create background task
    app['redis_listener'] = asyncio.create_task(listen_to_redis(app))

    yield  # App is running

    # Cleanup: Cancel background task
    app['redis_listener'].cancel()
    with suppress(asyncio.CancelledError):
        await app['redis_listener']

# Setup application
app = web.Application()
app.cleanup_ctx.append(background_tasks)

Pattern 3: Retry Logic with Exponential Backoff

import asyncio
import aiohttp
from typing import Any, Callable

async def retry_with_backoff(
    coro_func: Callable,
    max_retries: int = 3,
    base_delay: float = 1.0,
    max_delay: float = 60.0,
    *args,
    **kwargs
) -> Any:
    """
    Retry async function with exponential backoff

    Args:
        coro_func: Async function to retry
        max_retries: Maximum number of retries
        base_delay: Initial delay between retries
        max_delay: Maximum delay between retries
    """
    for attempt in range(max_retries):
        try:
            return await coro_func(*args, **kwargs)
        except Exception as e:
            if attempt == max_retries - 1:
                # Last attempt failed
                raise

            # Calculate delay with exponential backoff
            delay = min(base_delay * (2 ** attempt), max_delay)

            print(f'Attempt {attempt + 1} failed: {e}')
            print(f'Retrying in {delay:.1f} seconds...')

            await asyncio.sleep(delay)

# Usage
async def unstable_api_call():
    async with aiohttp.ClientSession() as session:
        async with session.get('http://unstable-api.com') as resp:
            return await resp.json()

async def main():
    result = await retry_with_backoff(
        unstable_api_call,
        max_retries=5,
        base_delay=1.0
    )
    return result

Pattern 4: Circuit Breaker

import asyncio
from datetime import datetime, timedelta
from enum import Enum

class CircuitState(Enum):
    CLOSED = "closed"      # Normal operation
    OPEN = "open"          # Failing, reject requests
    HALF_OPEN = "half_open"  # Testing if recovered

class CircuitBreaker:
    def __init__(
        self,
        failure_threshold: int = 5,
        recovery_timeout: float = 60.0,
        success_threshold: int = 2
    ):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.success_threshold = success_threshold

        self.failure_count = 0
        self.success_count = 0
        self.state = CircuitState.CLOSED
        self.opened_at = None

    async def call(self, coro_func, *args, **kwargs):
        if self.state == CircuitState.OPEN:
            # Check if should try recovery
            if datetime.now() - self.opened_at > timedelta(seconds=self.recovery_timeout):
                self.state = CircuitState.HALF_OPEN
                self.success_count = 0
            else:
                raise Exception('Circuit breaker is OPEN')

        try:
            result = await coro_func(*args, **kwargs)
            self._on_success()
            return result
        except Exception as e:
            self._on_failure()
            raise

    def _on_success(self):
        self.failure_count = 0

        if self.state == CircuitState.HALF_OPEN:
            self.success_count += 1
            if self.success_count >= self.success_threshold:
                self.state = CircuitState.CLOSED
                self.success_count = 0

    def _on_failure(self):
        self.failure_count += 1

        if self.failure_count >= self.failure_threshold:
            self.state = CircuitState.OPEN
            self.opened_at = datetime.now()

# Usage
async def flaky_service():
    # Simulated flaky service
    import random
    await asyncio.sleep(0.1)
    if random.random() < 0.5:
        raise Exception('Service error')
    return 'Success'

async def main():
    breaker = CircuitBreaker(failure_threshold=3, recovery_timeout=5.0)

    for i in range(20):
        try:
            result = await breaker.call(flaky_service)
            print(f'Request {i}: {result} - State: {breaker.state.value}')
        except Exception as e:
            print(f'Request {i}: Failed - State: {breaker.state.value}')

        await asyncio.sleep(0.5)

Pattern 5: WebSocket with Multiple Event Sources

Handling Parallel WebSocket and Background Events:

import asyncio
from aiohttp import web

async def read_subscription(ws, redis):
    """Background task reading from Redis and sending to WebSocket"""
    # Simulated Redis subscription
    channel = await redis.subscribe('channel:1')

    try:
        # Simulate receiving messages
        for i in range(10):
            await asyncio.sleep(1)
            message = f'Redis message {i}'
            await ws.send_str(message)
    finally:
        await redis.unsubscribe('channel:1')

async def websocket_handler(request):
    """WebSocket handler with parallel event sources"""
    ws = web.WebSocketResponse()
    await ws.prepare(request)

    # Create background task for Redis subscription
    redis = request.app['redis']
    task = asyncio.create_task(read_subscription(ws, redis))

    try:
        # Handle incoming WebSocket messages
        async for msg in ws:
            if msg.type == web.WSMsgType.TEXT:
                # Process incoming message
                await ws.send_str(f'Echo: {msg.data}')
            elif msg.type == web.WSMsgType.ERROR:
                print(f'WebSocket error: {ws.exception()}')
    finally:
        # Cleanup: Cancel background task
        task.cancel()

    return ws

Best Practices

Testing Async Code

Using pytest-asyncio:

import pytest
import asyncio

@pytest.mark.asyncio
async def test_async_function():
    result = await async_operation()
    assert result == 'expected'

@pytest.mark.asyncio
async def test_with_fixture(aiohttp_client):
    client = await aiohttp_client(create_app())
    resp = await client.get('/')
    assert resp.status == 200

Manual Event Loop Setup:

import asyncio
import unittest

class TestAsyncCode(unittest.TestCase):
    def setUp(self):
        self.loop = asyncio.new_event_loop()
        asyncio.set_event_loop(self.loop)

    def tearDown(self):
        self.loop.close()

    def test_coroutine(self):
        async def test_impl():
            result = await async_function()
            self.assertEqual(result, 'expected')

        self.loop.run_until_complete(test_impl())

Debugging Async Code

Enable Debug Mode:

import asyncio
import warnings

# Enable asyncio debug mode
asyncio.run(main(), debug=True)

# Or manually
loop = asyncio.get_event_loop()
loop.set_debug(True)
loop.run_until_complete(main())

What debug mode detects:

• Coroutines that were never awaited
• Callbacks taking too long
• Tasks destroyed while pending

Logging Slow Callbacks:

import asyncio
import logging

logging.basicConfig(level=logging.DEBUG)

loop = asyncio.get_event_loop()
loop.slow_callback_duration = 0.1  # 100ms threshold
loop.set_debug(True)

Documentation

Documenting Async Functions:

async def fetch_user_data(user_id: int) -> dict:
    """
    Fetch user data from the database.

    Args:
        user_id: The unique identifier of the user

    Returns:
        Dictionary containing user data

    Raises:
        UserNotFoundError: If user doesn't exist
        DatabaseError: If database connection fails

    Example:
        >>> async def main():
        ...     user = await fetch_user_data(123)
        ...     print(user['name'])

    Note:
        This function must be called within an async context.
        Connection pooling is handled automatically.
    """
    async with get_db_connection() as conn:
        return await conn.fetch_one(
            'SELECT * FROM users WHERE id = $1',
            user_id
        )

Complete Examples

Example 1: Parallel HTTP Requests

import asyncio
import aiohttp
import time

async def fetch(session, url):
    """Fetch a single URL"""
    async with session.get(url) as response:
        return {
            'url': url,
            'status': response.status,
            'length': len(await response.text())
        }

async def fetch_all(urls):
    """Fetch multiple URLs concurrently"""
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        return results

async def main():
    urls = [
        'http://python.org',
        'http://docs.python.org',
        'http://pypi.org',
        'http://github.com/python',
        'http://www.python.org/dev/peps/'
    ]

    start = time.perf_counter()
    results = await fetch_all(urls)
    elapsed = time.perf_counter() - start

    for result in results:
        print(f"{result['url']}: {result['status']} ({result['length']} bytes)")

    print(f"\nFetched {len(urls)} URLs in {elapsed:.2f} seconds")

asyncio.run(main())

Example 2: Rate-Limited API Client

import asyncio
import aiohttp
from typing import List, Dict, Any

class RateLimitedClient:
    def __init__(self, rate_limit: int = 10):
        """
        Args:
            rate_limit: Maximum concurrent requests
        """
        self.semaphore = asyncio.Semaphore(rate_limit)
        self.session = None

    async def __aenter__(self):
        self.session = aiohttp.ClientSession()
        return self

    async def __aexit__(self, *args):
        await self.session.close()
        # Allow connections to close
        await asyncio.sleep(0)

    async def fetch(self, url: str) -> Dict[str, Any]:
        """Fetch URL with rate limiting"""
        async with self.semaphore:
            print(f'Fetching {url}')
            async with self.session.get(url) as resp:
                return {
                    'url': url,
                    'status': resp.status,
                    'data': await resp.json()
                }

    async def fetch_all(self, urls: List[str]) -> List[Dict[str, Any]]:
        """Fetch all URLs with rate limiting"""
        tasks = [self.fetch(url) for url in urls]
        return await asyncio.gather(*tasks, return_exceptions=True)

async def main():
    urls = [f'https://api.github.com/users/{user}'
            for user in ['python', 'django', 'flask', 'requests', 'aiohttp']]

    async with RateLimitedClient(rate_limit=2) as client:
        results = await client.fetch_all(urls)

        for result in results:
            if isinstance(result, Exception):
                print(f'Error: {result}')
            else:
                print(f"User: {result['data'].get('login', 'unknown')}")

asyncio.run(main())

Example 3: Database Connection Pool

import asyncio
from typing import List, Any

class AsyncConnectionPool:
    def __init__(self, size: int = 10):
        self.pool = asyncio.Queue(maxsize=size)
        self.size = size

    async def init(self):
        """Initialize connection pool"""
        for i in range(self.size):
            conn = await self._create_connection(i)
            await self.pool.put(conn)

    async def _create_connection(self, conn_id: int):
        """Create a database connection (simulated)"""
        await asyncio.sleep(0.1)  # Simulate connection time
        return {'id': conn_id, 'connected': True}

    async def acquire(self):
        """Acquire connection from pool"""
        return await self.pool.get()

    async def release(self, conn):
        """Release connection back to pool"""
        await self.pool.put(conn)

    async def execute(self, query: str) -> Any:
        """Execute query using pooled connection"""
        conn = await self.acquire()
        try:
            # Simulate query execution
            await asyncio.sleep(0.05)
            return f"Query '{query}' executed on connection {conn['id']}"
        finally:
            await self.release(conn)

    async def close(self):
        """Close all connections"""
        while not self.pool.empty():
            conn = await self.pool.get()
            # Close connection (simulated)
            conn['connected'] = False

async def worker(pool: AsyncConnectionPool, worker_id: int):
    """Worker that executes queries"""
    for i in range(5):
        result = await pool.execute(f'SELECT * FROM table WHERE id={i}')
        print(f'Worker {worker_id}: {result}')

async def main():
    # Create and initialize pool
    pool = AsyncConnectionPool(size=5)
    await pool.init()

    # Run multiple workers concurrently
    await asyncio.gather(*[
        worker(pool, i) for i in range(10)
    ])

    # Cleanup
    await pool.close()

asyncio.run(main())

Example 4: Real-Time Data Processor

import asyncio
import random
from datetime import datetime

class DataProcessor:
    def __init__(self):
        self.queue = asyncio.Queue()
        self.processed = 0
        self.errors = 0

    async def producer(self, producer_id: int):
        """Produce data items"""
        for i in range(10):
            await asyncio.sleep(random.uniform(0.1, 0.5))
            item = {
                'producer_id': producer_id,
                'item_id': i,
                'timestamp': datetime.now(),
                'data': random.randint(1, 100)
            }
            await self.queue.put(item)
            print(f'Producer {producer_id} generated item {i}')

        # Signal completion
        await self.queue.put(None)

    async def consumer(self, consumer_id: int):
        """Consume and process data items"""
        while True:
            item = await self.queue.get()

            if item is None:
                # Propagate sentinel
                await self.queue.put(None)
                break

            try:
                # Simulate processing
                await asyncio.sleep(random.uniform(0.05, 0.2))

                # Process item
                result = item['data'] * 2
                print(f"Consumer {consumer_id} processed: {item['item_id']} -> {result}")

                self.processed += 1
            except Exception as e:
                print(f'Consumer {consumer_id} error: {e}')
                self.errors += 1
            finally:
                self.queue.task_done()

    async def monitor(self):
        """Monitor processing statistics"""
        while True:
            await asyncio.sleep(2)
            print(f'\n=== Stats: Processed={self.processed}, Errors={self.errors}, Queue={self.queue.qsize()} ===\n')

    async def run(self, num_producers: int = 3, num_consumers: int = 5):
        """Run the data processor"""
        # Start monitor
        monitor_task = asyncio.create_task(self.monitor())

        # Start producers and consumers
        await asyncio.gather(
            *[self.producer(i) for i in range(num_producers)],
            *[self.consumer(i) for i in range(num_consumers)]
        )

        # Cancel monitor
        monitor_task.cancel()

        print(f'\nFinal Stats: Processed={self.processed}, Errors={self.errors}')

async def main():
    processor = DataProcessor()
    await processor.run(num_producers=3, num_consumers=5)

asyncio.run(main())

Example 5: Async File I/O with aiofiles

import asyncio
import aiofiles
from pathlib import Path

async def write_file(path: str, content: str):
    """Write content to file asynchronously"""
    async with aiofiles.open(path, 'w') as f:
        await f.write(content)

async def read_file(path: str) -> str:
    """Read file content asynchronously"""
    async with aiofiles.open(path, 'r') as f:
        return await f.read()

async def process_files(file_paths: list):
    """Process multiple files concurrently"""
    tasks = [read_file(path) for path in file_paths]
    contents = await asyncio.gather(*tasks)

    # Process contents
    results = []
    for path, content in zip(file_paths, contents):
        result = {
            'path': path,
            'lines': len(content.split('\n')),
            'words': len(content.split()),
            'chars': len(content)
        }
        results.append(result)

    return results

async def main():
    # Create test files
    test_files = ['test1.txt', 'test2.txt', 'test3.txt']

    # Write files concurrently
    await asyncio.gather(*[
        write_file(f, f'Content of file {f}\n' * 10)
        for f in test_files
    ])

    # Process files
    results = await process_files(test_files)

    for result in results:
        print(f"{result['path']}: {result['lines']} lines, "
              f"{result['words']} words, {result['chars']} chars")

    # Cleanup
    for f in test_files:
        Path(f).unlink(missing_ok=True)

# asyncio.run(main())  # Uncomment to run (requires aiofiles)

Resources

• Python asyncio Documentation: https://docs.python.org/3/library/asyncio.html
• aiohttp Documentation: https://docs.aiohttp.org/
• Real Python asyncio Guide: https://realpython.com/async-io-python/
• PEP 492 - Coroutines with async and await syntax: https://www.python.org/dev/peps/pep-0492/
• asyncio Cheat Sheet: https://www.pythonsheets.com/notes/python-asyncio.html
• Effective Python: Item 60 - Consider asyncio: https://effectivepython.com/

---

Skill Version: 1.0.0 Last Updated: October 2025 Skill Category: Concurrency, Performance, Async Programming Compatible With: Python 3.7+, aiohttp, asyncio, uvloop

Context7 Documentation Snippets Used

This document lists all Context7 documentation snippets integrated into the asyncio-concurrency-patterns skill.

Library Information

• Library ID: /aio-libs/aiohttp
• Topic: asyncio concurrency event loops coroutines tasks futures async patterns
• Tokens: 8000

Snippets Integrated

1. WebSocket with Parallel Event Sources

Source: aiohttp FAQ Usage: Example 5 in EXAMPLES.md - WebSocket Server with Multiple Event Sources

Demonstrates handling multiple event sources concurrently with WebSocket connections using asyncio.create_task() for background event handling.

async def handler(request):
    ws = web.WebSocketResponse()
    await ws.prepare(request)
    task = asyncio.create_task(
        read_subscription(ws, request.app[redis_key]))
    try:
        async for msg in ws:
            # handle incoming messages
            ...
    finally:
        task.cancel()

2. Danger of Creating ClientSession Outside Event Loop

Source: aiohttp FAQ Usage: SKILL.md - Common Pitfalls section

Critical pattern showing why creating ClientSession at module level causes issues.

Pattern covered: Creating session inside async functions to avoid event loop binding issues.

3. Correctly Creating ClientSession

Source: aiohttp FAQ Usage: Multiple examples throughout SKILL.md and EXAMPLES.md

Standard pattern for proper ClientSession usage:

async def main():
    async with aiohttp.ClientSession() as session:
        async with session.get('http://python.org') as resp:
            print(await resp.text())

asyncio.run(main())

4. Run Task Concurrently with Handler

Source: aiohttp Web Advanced Usage: SKILL.md - Task Management section

Pattern for running background tasks during request handling:

async def handler(request):
    t = asyncio.create_task(get_some_data())
    # Do other work while data is being fetched
    data = await t
    return web.Response(text=data)

5. Event Loop Creation for Tests

Source: aiohttp Testing Usage: EXAMPLES.md - Testing section

Patterns for managing event loops in tests:

with loop_context() as loop:
    # Use the loop
    pass

loop = setup_test_loop()
try:
    # Use loop
finally:
    teardown_test_loop(loop)

6. Iterating Over WebSocket Messages

Source: aiohttp Web Reference Usage: EXAMPLES.md - WebSocket examples

Standard pattern for WebSocket message iteration:

ws = web.WebSocketResponse()
await ws.prepare(request)

async for msg in ws:
    print(msg.data)

7. Background Tasks with Application Lifecycle

Source: aiohttp Web Advanced Usage: SKILL.md - Production Patterns, EXAMPLES.md Example 11

Managing background tasks with cleanup context:

async def background_tasks(app):
    app[redis_listener] = asyncio.create_task(listen_to_redis(app))
    
    yield
    
    app[redis_listener].cancel()
    with contextlib.suppress(asyncio.CancelledError):
        await app[redis_listener]

8. Graceful Shutdown with Zero-Sleep

Source: aiohttp Client Advanced Usage: SKILL.md - Performance Optimization

Proper cleanup pattern for HTTP connections:

async def read_website():
    async with aiohttp.ClientSession() as session:
        async with session.get('http://example.org/') as resp:
            await resp.read()
    # Zero-sleep to allow underlying connections to close
    await asyncio.sleep(0)

9. Request Handler Definition

Source: aiohttp Web Quickstart Usage: SKILL.md - Core Concepts

Basic async handler pattern:

async def handler(request):
    return web.Response()

10. Fetch Content Pattern

Source: aiohttp README Usage: SKILL.md and README.md - Quick Start examples

Standard pattern for fetching web content:

async def main():
    async with aiohttp.ClientSession() as session:
        async with session.get('http://python.org') as response:
            print("Status:", response.status)
            html = await response.text()

asyncio.run(main())

11. WebSocket Receive Methods

Source: aiohttp Web Reference Usage: SKILL.md and EXAMPLES.md - WebSocket patterns

Methods for receiving different message types:

• receive_str() - TEXT messages
• receive_bytes() - BINARY messages
• receive_json() - JSON messages

12. Test Utilities

Source: aiohttp Testing Usage: EXAMPLES.md - Testing section

Testing helper functions:

• unused_port() - Get unused port
• loop_context() - Event loop context manager
• setup_test_loop() - Create test loop
• teardown_test_loop() - Cleanup test loop

13. StreamResponse and Status Setting

Source: aiohttp Web Reference Usage: SKILL.md - Async Context Managers

Pattern for configuring response headers and status:

async def handler(request):
    resp = StreamResponse()
    resp.set_status(404, reason="Not Found")
    await resp.prepare(request)
    return resp

14. ClientSession Usage as Async Context Manager

Source: aiohttp Client Reference Usage: Throughout all examples

Standard async context manager pattern:

resp = await client_session.get(url)
async with resp:
    assert resp.status == 200

15. WebSocket Close Handling

Source: aiohttp CHANGES Usage: EXAMPLES.md - WebSocket examples

Proper handling of WebSocket connection closure:

async for msg in ws:
    if msg.type == web.WSMsgType.CLOSED:
        break

16. ContextVars Example

Source: aiohttp Web Advanced Usage: SKILL.md - Advanced patterns

Context-local variables with asyncio:

from contextvars import ContextVar

VAR = ContextVar('VAR', default='default')

async def handler(request):
    var = VAR.get()
    VAR.set('handler')
    # Modifications isolated to this request

17. Retry Middleware Pattern

Source: aiohttp Client Middleware Cookbook Usage: EXAMPLES.md - Retry Logic example

Retry pattern for handling connection errors:

async def retry_middleware(client, service, **kwargs):
    attempts = kwargs.pop('attempts', 3)
    for attempt in range(attempts):
        try:
            return await client.request(service, **kwargs)
        except aiohttp.ClientConnectionError as e:
            if attempt + 1 == attempts:
                raise
            await asyncio.sleep(0.1 * (attempt + 1))

18. Custom Async Access Logger

Source: aiohttp Logging Usage: SKILL.md - Production Patterns

Creating custom async logger:

from aiohttp.abc import AbstractAsyncAccessLogger

class AccessLogger(AbstractAsyncAccessLogger):
    async def log(self, request, response, time):
        logging_service = request.app['logging_service']
        await logging_service.log(f'{request.remote} '
                                f'"{request.method} {request.path} '
                                f'done in {time}s: {response.status}')

19. Background Task Spawning

Source: aiohttp Web Advanced Usage: SKILL.md - Background Tasks

Using aiojobs for background tasks:

from aiojobs.aiohttp import setup, spawn

async def handler(request):
    await spawn(request, write_data())
    return web.Response()

app = web.Application()
setup(app)

20. Gunicorn Worker Configuration

Source: aiohttp Web Advanced Usage: SKILL.md - Production deployment

Running aiohttp with Gunicorn:

gunicorn my_app_module:my_web_app --bind localhost:8080 \
    --worker-class aiohttp.GunicornWebWorker

Integration Summary

Coverage Areas

1. Event Loop Management: Loop creation, policies, and lifecycle 2. HTTP Client Patterns: Session management, request patterns, cleanup 3. WebSocket Handling: Message iteration, parallel events, cleanup 4. Background Tasks: Lifecycle management, cancellation, cleanup 5. Testing Patterns: Test loop setup, utilities, fixtures 6. Production Patterns: Graceful shutdown, retry logic, logging 7. Performance: Zero-sleep cleanup, connection pooling 8. Error Handling: Exception handling, timeouts, retries

Pattern Distribution

• Core Patterns: 8 snippets (Event loops, sessions, handlers)
• WebSocket Patterns: 4 snippets (Message handling, connections)
• Background Tasks: 3 snippets (Lifecycle, cleanup, spawning)
• Testing: 2 snippets (Loop management, utilities)
• Production: 3 snippets (Logging, deployment, retry)

Code Examples Enhanced

• 20+ examples use Context7 patterns
• All production patterns validated against aiohttp docs
• Best practices aligned with official recommendations
• Common pitfalls based on official FAQ

References

• aiohttp Documentation: https://docs.aiohttp.org/
• aiohttp GitHub: https://github.com/aio-libs/aiohttp
• Context7 Library: /aio-libs/aiohttp

Asyncio Concurrency Patterns - Practical Examples

Comprehensive collection of real-world asyncio examples covering concurrency patterns, error handling, performance optimization, and production-ready code.

Table of Contents

1. HTTP Client Examples 2. WebSocket Examples 3. Database Examples 4. Queue & Task Processing 5. Concurrency Control 6. Error Handling & Retry Logic 7. Background Tasks 8. Testing Examples 9. Performance Examples 10. Production Patterns

---

HTTP Client Examples

Example 1: Basic Concurrent HTTP Requests

Fetch multiple URLs concurrently and compare performance with sequential approach.

import asyncio
import aiohttp
import time

async def fetch_url(session, url):
    """Fetch a single URL and return status and content length"""
    async with session.get(url) as response:
        content = await response.text()
        return {
            'url': url,
            'status': response.status,
            'length': len(content),
            'content_type': response.headers.get('content-type', '')
        }

async def fetch_all_concurrent(urls):
    """Fetch all URLs concurrently"""
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
    return results

async def fetch_all_sequential(urls):
    """Fetch all URLs sequentially (for comparison)"""
    results = []
    async with aiohttp.ClientSession() as session:
        for url in urls:
            result = await fetch_url(session, url)
            results.append(result)
    return results

async def main():
    urls = [
        'http://python.org',
        'http://docs.python.org',
        'http://pypi.org',
        'http://github.com/python',
        'http://www.python.org/dev/peps/'
    ]

    # Concurrent approach
    start = time.perf_counter()
    concurrent_results = await fetch_all_concurrent(urls)
    concurrent_time = time.perf_counter() - start

    print("=== Concurrent Results ===")
    for result in concurrent_results:
        print(f"{result['url']}: {result['status']} ({result['length']} bytes)")
    print(f"Concurrent time: {concurrent_time:.2f}s\n")

    # Sequential approach
    start = time.perf_counter()
    sequential_results = await fetch_all_sequential(urls)
    sequential_time = time.perf_counter() - start

    print("=== Sequential Results ===")
    for result in sequential_results:
        print(f"{result['url']}: {result['status']} ({result['length']} bytes)")
    print(f"Sequential time: {sequential_time:.2f}s")

    print(f"\nSpeedup: {sequential_time / concurrent_time:.2f}x faster")

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

Example 2: Rate-Limited API Client

Production-ready API client with rate limiting and error handling.

import asyncio
import aiohttp
from typing import List, Dict, Any, Optional
from dataclasses import dataclass
from datetime import datetime

@dataclass
class APIResponse:
    url: str
    status: int
    data: Any
    timestamp: datetime
    error: Optional[str] = None

class RateLimitedAPIClient:
    """
    API client with configurable rate limiting and retry logic
    """

    def __init__(
        self,
        base_url: str,
        max_concurrent: int = 10,
        requests_per_second: float = 5.0,
        timeout: int = 30
    ):
        self.base_url = base_url.rstrip('/')
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.rate_limiter = asyncio.Semaphore(int(requests_per_second))
        self.min_interval = 1.0 / requests_per_second
        self.timeout = aiohttp.ClientTimeout(total=timeout)
        self.session: Optional[aiohttp.ClientSession] = None
        self.request_count = 0
        self.error_count = 0

    async def __aenter__(self):
        self.session = aiohttp.ClientSession(timeout=self.timeout)
        return self

    async def __aexit__(self, *args):
        if self.session:
            await self.session.close()
            await asyncio.sleep(0)

    async def _rate_limit(self):
        """Enforce rate limiting"""
        async with self.rate_limiter:
            await asyncio.sleep(self.min_interval)

    async def get(self, endpoint: str, params: Optional[Dict] = None) -> APIResponse:
        """
        Make GET request with rate limiting and error handling
        """
        async with self.semaphore:
            await self._rate_limit()

            url = f'{self.base_url}/{endpoint.lstrip("/")}'
            timestamp = datetime.now()

            try:
                async with self.session.get(url, params=params) as resp:
                    self.request_count += 1

                    if resp.status == 200:
                        data = await resp.json()
                        return APIResponse(
                            url=str(resp.url),
                            status=resp.status,
                            data=data,
                            timestamp=timestamp
                        )
                    else:
                        self.error_count += 1
                        return APIResponse(
                            url=str(resp.url),
                            status=resp.status,
                            data=None,
                            timestamp=timestamp,
                            error=f'HTTP {resp.status}'
                        )

            except asyncio.TimeoutError:
                self.error_count += 1
                return APIResponse(
                    url=url,
                    status=0,
                    data=None,
                    timestamp=timestamp,
                    error='Timeout'
                )
            except Exception as e:
                self.error_count += 1
                return APIResponse(
                    url=url,
                    status=0,
                    data=None,
                    timestamp=timestamp,
                    error=str(e)
                )

    async def get_many(self, endpoints: List[str]) -> List[APIResponse]:
        """Fetch multiple endpoints concurrently"""
        tasks = [self.get(endpoint) for endpoint in endpoints]
        return await asyncio.gather(*tasks)

    def get_stats(self) -> Dict[str, int]:
        """Get client statistics"""
        return {
            'total_requests': self.request_count,
            'total_errors': self.error_count,
            'success_rate': (
                (self.request_count - self.error_count) / self.request_count * 100
                if self.request_count > 0 else 0
            )
        }

async def main():
    # Example: Fetch GitHub user data
    async with RateLimitedAPIClient(
        'https://api.github.com',
        max_concurrent=5,
        requests_per_second=2.0
    ) as client:

        # Fetch multiple users
        users = ['python', 'django', 'flask', 'requests', 'aiohttp']
        endpoints = [f'users/{user}' for user in users]

        print("Fetching user data...")
        results = await client.get_many(endpoints)

        print("\n=== Results ===")
        for result in results:
            if result.error:
                print(f"❌ {result.url}: {result.error}")
            else:
                user_data = result.data
                print(f"✓ {user_data.get('login', 'unknown')}: "
                      f"{user_data.get('public_repos', 0)} repos, "
                      f"{user_data.get('followers', 0)} followers")

        print(f"\n=== Stats ===")
        stats = client.get_stats()
        print(f"Total requests: {stats['total_requests']}")
        print(f"Errors: {stats['total_errors']}")
        print(f"Success rate: {stats['success_rate']:.1f}%")

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

Example 3: Streaming Large Downloads

Handle large file downloads with progress tracking.

import asyncio
import aiohttp
from pathlib import Path
from typing import Optional

class DownloadProgress:
    def __init__(self, total_size: int):
        self.total_size = total_size
        self.downloaded = 0
        self.start_time = asyncio.get_event_loop().time()

    def update(self, chunk_size: int):
        self.downloaded += chunk_size
        elapsed = asyncio.get_event_loop().time() - self.start_time
        speed = self.downloaded / elapsed if elapsed > 0 else 0
        percent = (self.downloaded / self.total_size * 100) if self.total_size > 0 else 0

        print(f"\rProgress: {percent:.1f}% | "
              f"{self.downloaded / 1024 / 1024:.2f} MB / "
              f"{self.total_size / 1024 / 1024:.2f} MB | "
              f"Speed: {speed / 1024 / 1024:.2f} MB/s", end='')

async def download_file(
    url: str,
    destination: Path,
    chunk_size: int = 8192,
    show_progress: bool = True
) -> bool:
    """
    Download file with progress tracking

    Args:
        url: URL to download from
        destination: Path to save file
        chunk_size: Size of chunks to download
        show_progress: Whether to show progress

    Returns:
        True if successful, False otherwise
    """
    try:
        async with aiohttp.ClientSession() as session:
            async with session.get(url) as resp:
                if resp.status != 200:
                    print(f"Error: HTTP {resp.status}")
                    return False

                total_size = int(resp.headers.get('content-length', 0))
                progress = DownloadProgress(total_size) if show_progress else None

                with open(destination, 'wb') as f:
                    async for chunk in resp.content.iter_chunked(chunk_size):
                        f.write(chunk)
                        if progress:
                            progress.update(len(chunk))

                if show_progress:
                    print()  # New line after progress

                return True

    except Exception as e:
        print(f"\nError downloading file: {e}")
        return False

async def download_multiple(downloads: dict):
    """
    Download multiple files concurrently

    Args:
        downloads: Dict mapping URLs to destination paths
    """
    tasks = [
        download_file(url, Path(dest), show_progress=False)
        for url, dest in downloads.items()
    ]

    results = await asyncio.gather(*tasks, return_exceptions=True)

    print("\n=== Download Summary ===")
    for (url, dest), result in zip(downloads.items(), results):
        if isinstance(result, Exception):
            print(f"❌ {url}: {result}")
        elif result:
            print(f"✓ {url} -> {dest}")
        else:
            print(f"❌ {url}: Failed")

async def main():
    # Example: Download a single file
    url = 'http://ipv4.download.thinkbroadband.com/10MB.zip'
    await download_file(url, Path('test_download.zip'))

    # Example: Download multiple files
    # downloads = {
    #     'http://example.com/file1.zip': 'file1.zip',
    #     'http://example.com/file2.zip': 'file2.zip',
    # }
    # await download_multiple(downloads)

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

---

WebSocket Examples

Example 4: WebSocket Client with Reconnection

Robust WebSocket client with automatic reconnection and heartbeat.

import asyncio
import aiohttp
from typing import Optional, Callable
from datetime import datetime

class WebSocketClient:
    """
    WebSocket client with automatic reconnection and heartbeat
    """

    def __init__(
        self,
        url: str,
        heartbeat_interval: float = 30.0,
        reconnect_interval: float = 5.0,
        max_reconnect_attempts: int = 5
    ):
        self.url = url
        self.heartbeat_interval = heartbeat_interval
        self.reconnect_interval = reconnect_interval
        self.max_reconnect_attempts = max_reconnect_attempts

        self.ws: Optional[aiohttp.ClientWebSocketResponse] = None
        self.session: Optional[aiohttp.ClientSession] = None
        self.should_run = False
        self.reconnect_count = 0

        self.on_message: Optional[Callable] = None
        self.on_connect: Optional[Callable] = None
        self.on_disconnect: Optional[Callable] = None

    async def connect(self):
        """Establish WebSocket connection"""
        self.session = aiohttp.ClientSession()
        self.ws = await self.session.ws_connect(self.url)
        self.reconnect_count = 0

        print(f"[{datetime.now()}] Connected to {self.url}")

        if self.on_connect:
            await self.on_connect()

    async def disconnect(self):
        """Close WebSocket connection"""
        if self.ws:
            await self.ws.close()

        if self.session:
            await self.session.close()

        print(f"[{datetime.now()}] Disconnected")

        if self.on_disconnect:
            await self.on_disconnect()

    async def send(self, message: str):
        """Send message to WebSocket"""
        if self.ws and not self.ws.closed:
            await self.ws.send_str(message)

    async def _heartbeat(self):
        """Send periodic heartbeat"""
        while self.should_run:
            try:
                await asyncio.sleep(self.heartbeat_interval)
                if self.ws and not self.ws.closed:
                    await self.ws.ping()
            except Exception as e:
                print(f"Heartbeat error: {e}")

    async def _receive_messages(self):
        """Receive and process messages"""
        try:
            async for msg in self.ws:
                if msg.type == aiohttp.WSMsgType.TEXT:
                    if self.on_message:
                        await self.on_message(msg.data)
                elif msg.type == aiohttp.WSMsgType.ERROR:
                    print(f"WebSocket error: {self.ws.exception()}")
                    break
        except Exception as e:
            print(f"Error receiving messages: {e}")

    async def run(self):
        """Run WebSocket client with reconnection"""
        self.should_run = True

        while self.should_run:
            try:
                await self.connect()

                # Start heartbeat and message receiver
                await asyncio.gather(
                    self._heartbeat(),
                    self._receive_messages()
                )

            except Exception as e:
                print(f"Connection error: {e}")

            finally:
                await self.disconnect()

            # Reconnection logic
            if self.should_run:
                self.reconnect_count += 1

                if self.reconnect_count >= self.max_reconnect_attempts:
                    print("Max reconnection attempts reached")
                    break

                print(f"Reconnecting in {self.reconnect_interval}s... "
                      f"(attempt {self.reconnect_count}/{self.max_reconnect_attempts})")
                await asyncio.sleep(self.reconnect_interval)

    async def stop(self):
        """Stop WebSocket client"""
        self.should_run = False
        await self.disconnect()

# Example usage
async def main():
    client = WebSocketClient('wss://echo.websocket.org')

    # Set up handlers
    async def on_message(message):
        print(f"Received: {message}")

    async def on_connect():
        print("Connected! Sending test message...")
        await client.send("Hello WebSocket!")

    client.on_message = on_message
    client.on_connect = on_connect

    # Run client (with automatic stop after 30 seconds for demo)
    client_task = asyncio.create_task(client.run())

    await asyncio.sleep(30)
    await client.stop()
    await client_task

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

Example 5: WebSocket Server with Multiple Event Sources

Handle WebSocket connections with parallel event sources (based on aiohttp documentation).

import asyncio
from aiohttp import web
from collections import defaultdict
from typing import Set

class WebSocketServer:
    """
    WebSocket server handling multiple event sources
    """

    def __init__(self):
        self.app = web.Application()
        self.websockets: defaultdict[str, Set] = defaultdict(set)

        # Setup routes
        self.app.router.add_get('/ws/{channel}', self.websocket_handler)
        self.app.on_startup.append(self.on_startup)
        self.app.on_cleanup.append(self.on_cleanup)

    async def on_startup(self, app):
        """Initialize background tasks on startup"""
        # Start broadcast task
        app['broadcast_task'] = asyncio.create_task(self.broadcast_loop())
        print("Server started, broadcast task running")

    async def on_cleanup(self, app):
        """Cleanup on shutdown"""
        # Cancel broadcast task
        app['broadcast_task'].cancel()
        with asyncio.suppress(asyncio.CancelledError):
            await app['broadcast_task']

        # Close all websockets
        for channel, ws_set in self.websockets.items():
            for ws in ws_set:
                await ws.close()

        print("Server cleanup complete")

    async def broadcast_loop(self):
        """
        Background task that broadcasts messages to all connected clients
        Simulates reading from Redis, Kafka, etc.
        """
        counter = 0
        try:
            while True:
                await asyncio.sleep(2)
                counter += 1

                # Broadcast to all channels
                message = f"Broadcast message {counter}"
                await self.broadcast_to_channel('general', message)

        except asyncio.CancelledError:
            print("Broadcast task cancelled")
            raise

    async def broadcast_to_channel(self, channel: str, message: str):
        """Send message to all clients in a channel"""
        if channel not in self.websockets:
            return

        dead_sockets = set()

        for ws in self.websockets[channel]:
            try:
                if not ws.closed:
                    await ws.send_str(message)
                else:
                    dead_sockets.add(ws)
            except Exception as e:
                print(f"Error sending to websocket: {e}")
                dead_sockets.add(ws)

        # Remove dead sockets
        self.websockets[channel] -= dead_sockets

    async def websocket_handler(self, request):
        """
        Handle WebSocket connections with parallel event sources
        """
        channel = request.match_info['channel']
        ws = web.WebSocketResponse()
        await ws.prepare(request)

        # Register websocket
        self.websockets[channel].add(ws)
        print(f"Client connected to channel: {channel}")

        # Create background task for this connection (e.g., Redis subscription)
        task = asyncio.create_task(
            self.read_external_events(ws, channel)
        )

        try:
            # Handle incoming messages from client
            async for msg in ws:
                if msg.type == web.WSMsgType.TEXT:
                    # Echo back to sender
                    await ws.send_str(f"Echo: {msg.data}")

                    # Broadcast to all in channel
                    await self.broadcast_to_channel(
                        channel,
                        f"User message: {msg.data}"
                    )

                elif msg.type == web.WSMsgType.ERROR:
                    print(f'WebSocket error: {ws.exception()}')

        finally:
            # Cleanup
            task.cancel()
            with asyncio.suppress(asyncio.CancelledError):
                await task

            self.websockets[channel].discard(ws)
            print(f"Client disconnected from channel: {channel}")

        return ws

    async def read_external_events(self, ws, channel):
        """
        Simulate reading from external event source (Redis, Kafka, etc.)
        and sending to WebSocket
        """
        try:
            counter = 0
            while True:
                await asyncio.sleep(5)
                counter += 1

                # Simulate external event
                message = f"External event {counter} for {channel}"
                if not ws.closed:
                    await ws.send_str(message)

        except asyncio.CancelledError:
            print(f"External event reader cancelled for {channel}")
            raise

    def run(self, host='0.0.0.0', port=8080):
        """Run the server"""
        web.run_app(self.app, host=host, port=port)

# Example usage
if __name__ == '__main__':
    server = WebSocketServer()
    server.run()

---

Database Examples

Example 6: Connection Pool Management

Async database connection pool with proper resource management.

import asyncio
from typing import Any, Optional
from contextlib import asynccontextmanager

class AsyncConnectionPool:
    """
    Async database connection pool
    """

    def __init__(self, size: int = 10, max_overflow: int = 5):
        self.size = size
        self.max_overflow = max_overflow
        self.pool = asyncio.Queue(maxsize=size + max_overflow)
        self.current_size = 0
        self.in_use = 0
        self._lock = asyncio.Lock()

    async def init(self):
        """Initialize connection pool"""
        for i in range(self.size):
            conn = await self._create_connection(i)
            await self.pool.put(conn)
            self.current_size += 1

        print(f"Connection pool initialized with {self.size} connections")

    async def _create_connection(self, conn_id: int):
        """Create a database connection (simulated)"""
        await asyncio.sleep(0.1)  # Simulate connection time
        return {
            'id': conn_id,
            'connected': True,
            'queries_executed': 0,
            'created_at': asyncio.get_event_loop().time()
        }

    async def acquire(self) -> dict:
        """Acquire connection from pool"""
        try:
            # Try to get existing connection
            conn = self.pool.get_nowait()
            self.in_use += 1
            return conn
        except asyncio.QueueEmpty:
            # Pool is empty, create overflow connection if allowed
            async with self._lock:
                if self.current_size < self.size + self.max_overflow:
                    conn = await self._create_connection(self.current_size)
                    self.current_size += 1
                    self.in_use += 1
                    return conn

            # Wait for connection to become available
            conn = await self.pool.get()
            self.in_use += 1
            return conn

    async def release(self, conn: dict):
        """Release connection back to pool"""
        if conn['connected']:
            await self.pool.put(conn)
            self.in_use -= 1

    @asynccontextmanager
    async def connection(self):
        """Context manager for acquiring/releasing connections"""
        conn = await self.acquire()
        try:
            yield conn
        finally:
            await self.release(conn)

    async def execute(self, query: str) -> Any:
        """Execute query using pooled connection"""
        async with self.connection() as conn:
            # Simulate query execution
            await asyncio.sleep(0.05)
            conn['queries_executed'] += 1
            return f"Query '{query}' executed on connection {conn['id']}"

    async def close(self):
        """Close all connections"""
        print("Closing connection pool...")

        while not self.pool.empty():
            try:
                conn = self.pool.get_nowait()
                conn['connected'] = False
            except asyncio.QueueEmpty:
                break

        self.current_size = 0
        print("All connections closed")

    def get_stats(self) -> dict:
        """Get pool statistics"""
        return {
            'total_connections': self.current_size,
            'available': self.pool.qsize(),
            'in_use': self.in_use,
            'max_size': self.size + self.max_overflow
        }

async def worker(pool: AsyncConnectionPool, worker_id: int, num_queries: int):
    """Worker that executes queries"""
    for i in range(num_queries):
        result = await pool.execute(f'SELECT * FROM table_{worker_id} WHERE id={i}')
        print(f'Worker {worker_id}: {result}')
        await asyncio.sleep(0.1)

async def main():
    # Create and initialize pool
    pool = AsyncConnectionPool(size=5, max_overflow=3)
    await pool.init()

    print("\n=== Starting Workers ===")
    # Run multiple workers concurrently
    await asyncio.gather(*[
        worker(pool, i, 5) for i in range(10)
    ])

    print("\n=== Pool Stats ===")
    stats = pool.get_stats()
    for key, value in stats.items():
        print(f"{key}: {value}")

    # Cleanup
    await pool.close()

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

---

Queue & Task Processing

Example 7: Advanced Producer-Consumer Pattern

Multi-producer, multi-consumer with priority queue and monitoring.

import asyncio
import random
from enum import IntEnum
from dataclasses import dataclass, field
from datetime import datetime
from typing import Optional

class Priority(IntEnum):
    LOW = 3
    NORMAL = 2
    HIGH = 1
    URGENT = 0

@dataclass(order=True)
class Task:
    priority: int
    task_id: int = field(compare=False)
    data: str = field(compare=False)
    created_at: datetime = field(default_factory=datetime.now, compare=False)
    processed_at: Optional[datetime] = field(default=None, compare=False)

class TaskProcessor:
    def __init__(
        self,
        num_producers: int = 3,
        num_consumers: int = 5,
        max_queue_size: int = 100
    ):
        self.queue = asyncio.PriorityQueue(maxsize=max_queue_size)
        self.num_producers = num_producers
        self.num_consumers = num_consumers

        self.produced_count = 0
        self.consumed_count = 0
        self.error_count = 0
        self.running = False

    async def producer(self, producer_id: int, num_tasks: int):
        """Produce tasks with random priorities"""
        for i in range(num_tasks):
            # Random delay
            await asyncio.sleep(random.uniform(0.05, 0.2))

            # Create task with random priority
            priority = random.choice(list(Priority))
            task = Task(
                priority=priority.value,
                task_id=self.produced_count,
                data=f'Producer-{producer_id}-Task-{i}'
            )

            await self.queue.put(task)
            self.produced_count += 1

            print(f"📥 Producer {producer_id} created {task.data} "
                  f"(Priority: {priority.name})")

        print(f"Producer {producer_id} finished")

    async def consumer(self, consumer_id: int):
        """Consume and process tasks"""
        while self.running:
            try:
                # Wait for task with timeout
                task = await asyncio.wait_for(
                    self.queue.get(),
                    timeout=1.0
                )

                # Process task
                task.processed_at = datetime.now()
                processing_time = random.uniform(0.1, 0.5)
                await asyncio.sleep(processing_time)

                # Simulate occasional errors
                if random.random() < 0.1:  # 10% error rate
                    raise Exception("Processing error")

                self.consumed_count += 1
                wait_time = (task.processed_at - task.created_at).total_seconds()

                print(f"✅ Consumer {consumer_id} processed {task.data} "
                      f"(waited {wait_time:.2f}s, processed in {processing_time:.2f}s)")

            except asyncio.TimeoutError:
                # No tasks available
                continue
            except Exception as e:
                self.error_count += 1
                print(f"❌ Consumer {consumer_id} error: {e}")

        print(f"Consumer {consumer_id} stopped")

    async def monitor(self):
        """Monitor queue and processing stats"""
        while self.running:
            await asyncio.sleep(2)

            stats = self.get_stats()
            print(f"\n📊 Stats: Produced={stats['produced']}, "
                  f"Consumed={stats['consumed']}, "
                  f"Errors={stats['errors']}, "
                  f"Queue={stats['queue_size']}\n")

    async def run(self, tasks_per_producer: int = 10):
        """Run the task processor"""
        self.running = True

        # Start monitor
        monitor_task = asyncio.create_task(self.monitor())

        # Start all producers
        producer_tasks = [
            asyncio.create_task(self.producer(i, tasks_per_producer))
            for i in range(self.num_producers)
        ]

        # Start all consumers
        consumer_tasks = [
            asyncio.create_task(self.consumer(i))
            for i in range(self.num_consumers)
        ]

        # Wait for all producers to finish
        await asyncio.gather(*producer_tasks)

        # Wait for queue to be empty
        await self.queue.join()

        # Stop consumers and monitor
        self.running = False
        await asyncio.sleep(1.5)  # Give consumers time to stop

        # Cancel any remaining tasks
        monitor_task.cancel()
        for task in consumer_tasks:
            task.cancel()

        print("\n=== Final Stats ===")
        stats = self.get_stats()
        for key, value in stats.items():
            print(f"{key}: {value}")

    def get_stats(self) -> dict:
        """Get processing statistics"""
        return {
            'produced': self.produced_count,
            'consumed': self.consumed_count,
            'errors': self.error_count,
            'queue_size': self.queue.qsize(),
            'success_rate': (
                (self.consumed_count / (self.consumed_count + self.error_count) * 100)
                if (self.consumed_count + self.error_count) > 0
                else 0
            )
        }

async def main():
    processor = TaskProcessor(
        num_producers=3,
        num_consumers=5,
        max_queue_size=50
    )

    await processor.run(tasks_per_producer=10)

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

---

Concurrency Control

Example 8: Semaphore for Resource Limiting

Control access to limited resources with semaphores.

import asyncio
import random
from datetime import datetime

class ResourcePool:
    """
    Manage limited resources with semaphore
    """

    def __init__(self, max_resources: int = 5):
        self.semaphore = asyncio.Semaphore(max_resources)
        self.max_resources = max_resources
        self.active_count = 0
        self.total_acquired = 0

    async def acquire_resource(self, user_id: int, duration: float):
        """
        Acquire resource, use it, then release

        Args:
            user_id: ID of the user acquiring resource
            duration: How long to hold the resource
        """
        print(f"[{datetime.now().strftime('%H:%M:%S')}] "
              f"User {user_id} waiting for resource... "
              f"(Active: {self.active_count}/{self.max_resources})")

        async with self.semaphore:
            self.active_count += 1
            self.total_acquired += 1

            print(f"[{datetime.now().strftime('%H:%M:%S')}] "
                  f"✓ User {user_id} acquired resource "
                  f"(Active: {self.active_count}/{self.max_resources})")

            try:
                # Use resource
                await asyncio.sleep(duration)
                result = f"User {user_id} completed work in {duration:.2f}s"

            finally:
                self.active_count -= 1
                print(f"[{datetime.now().strftime('%H:%M:%S')}] "
                      f"User {user_id} released resource "
                      f"(Active: {self.active_count}/{self.max_resources})")

            return result

async def main():
    # Create pool with 5 resources
    pool = ResourcePool(max_resources=5)

    # Simulate 20 users trying to access resources
    tasks = [
        pool.acquire_resource(
            user_id=i,
            duration=random.uniform(1.0, 3.0)
        )
        for i in range(20)
    ]

    results = await asyncio.gather(*tasks)

    print("\n=== Results ===")
    for result in results:
        print(result)

    print(f"\nTotal acquisitions: {pool.total_acquired}")

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

Example 9: Lock for Shared State

Protect shared state with async locks.

import asyncio
from dataclasses import dataclass, field
from typing import List

@dataclass
class BankAccount:
    """Thread-safe bank account using async lock"""
    balance: float = 0.0
    transactions: List[dict] = field(default_factory=list)
    _lock: asyncio.Lock = field(default_factory=asyncio.Lock)

    async def deposit(self, amount: float, description: str = ""):
        """Deposit money (thread-safe)"""
        async with self._lock:
            # Critical section
            old_balance = self.balance
            await asyncio.sleep(0.01)  # Simulate processing time
            self.balance += amount

            self.transactions.append({
                'type': 'deposit',
                'amount': amount,
                'description': description,
                'old_balance': old_balance,
                'new_balance': self.balance
            })

            print(f"💰 Deposited ${amount:.2f}: ${old_balance:.2f} -> ${self.balance:.2f}")

    async def withdraw(self, amount: float, description: str = ""):
        """Withdraw money (thread-safe)"""
        async with self._lock:
            # Critical section
            if self.balance < amount:
                print(f"❌ Insufficient funds: ${self.balance:.2f} < ${amount:.2f}")
                return False

            old_balance = self.balance
            await asyncio.sleep(0.01)  # Simulate processing time
            self.balance -= amount

            self.transactions.append({
                'type': 'withdrawal',
                'amount': amount,
                'description': description,
                'old_balance': old_balance,
                'new_balance': self.balance
            })

            print(f"💸 Withdrew ${amount:.2f}: ${old_balance:.2f} -> ${self.balance:.2f}")
            return True

    async def get_balance(self) -> float:
        """Get current balance (thread-safe)"""
        async with self._lock:
            return self.balance

async def customer_transactions(account: BankAccount, customer_id: int):
    """Simulate customer making random transactions"""
    import random

    for i in range(5):
        await asyncio.sleep(random.uniform(0.05, 0.2))

        if random.choice([True, False]):
            amount = random.uniform(10, 100)
            await account.deposit(amount, f"Customer {customer_id} deposit {i}")
        else:
            amount = random.uniform(10, 50)
            await account.withdraw(amount, f"Customer {customer_id} withdrawal {i}")

async def main():
    # Create account with initial balance
    account = BankAccount(balance=1000.0)

    print(f"Initial balance: ${account.balance:.2f}\n")

    # Simulate multiple customers accessing account concurrently
    await asyncio.gather(*[
        customer_transactions(account, i)
        for i in range(10)
    ])

    final_balance = await account.get_balance()

    print(f"\n=== Summary ===")
    print(f"Final balance: ${final_balance:.2f}")
    print(f"Total transactions: {len(account.transactions)}")

    deposits = sum(t['amount'] for t in account.transactions if t['type'] == 'deposit')
    withdrawals = sum(t['amount'] for t in account.transactions if t['type'] == 'withdrawal')

    print(f"Total deposits: ${deposits:.2f}")
    print(f"Total withdrawals: ${withdrawals:.2f}")
    print(f"Expected balance: ${1000 + deposits - withdrawals:.2f}")

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

---

Error Handling & Retry Logic

Example 10: Exponential Backoff Retry

Implement retry logic with exponential backoff.

import asyncio
import random
from typing import TypeVar, Callable, Any
from dataclasses import dataclass

T = TypeVar('T')

@dataclass
class RetryStats:
    total_attempts: int = 0
    successful: int = 0
    failed: int = 0
    total_delay: float = 0.0

async def retry_with_exponential_backoff(
    coro_func: Callable[..., Any],
    *args,
    max_retries: int = 3,
    base_delay: float = 1.0,
    max_delay: float = 60.0,
    exponential_base: float = 2.0,
    jitter: bool = True,
    **kwargs
) -> tuple[Any, RetryStats]:
    """
    Retry async function with exponential backoff

    Args:
        coro_func: Async function to retry
        max_retries: Maximum number of retry attempts
        base_delay: Initial delay between retries (seconds)
        max_delay: Maximum delay between retries (seconds)
        exponential_base: Base for exponential backoff
        jitter: Add random jitter to prevent thundering herd

    Returns:
        Tuple of (result, stats)
    """
    stats = RetryStats()

    for attempt in range(max_retries + 1):
        stats.total_attempts += 1

        try:
            result = await coro_func(*args, **kwargs)
            stats.successful += 1
            return result, stats

        except Exception as e:
            if attempt == max_retries:
                # Final attempt failed
                stats.failed += 1
                raise

            # Calculate delay with exponential backoff
            delay = min(base_delay * (exponential_base ** attempt), max_delay)

            # Add jitter
            if jitter:
                delay = delay * (0.5 + random.random() * 0.5)

            stats.total_delay += delay

            print(f"⚠️  Attempt {attempt + 1} failed: {e}")
            print(f"   Retrying in {delay:.2f}s...")

            await asyncio.sleep(delay)

# Example: Flaky API call
async def unstable_api_call(success_rate: float = 0.3) -> dict:
    """Simulate an unstable API that sometimes fails"""
    await asyncio.sleep(0.1)  # Simulate network delay

    if random.random() > success_rate:
        raise ConnectionError("API temporarily unavailable")

    return {"status": "success", "data": "Important data"}

async def main():
    print("=== Testing Retry with Exponential Backoff ===\n")

    # Test with different configurations
    configs = [
        {'max_retries': 3, 'base_delay': 1.0},
        {'max_retries': 5, 'base_delay': 0.5, 'exponential_base': 1.5},
        {'max_retries': 4, 'base_delay': 2.0, 'jitter': False},
    ]

    for i, config in enumerate(configs, 1):
        print(f"--- Configuration {i} ---")
        print(f"Config: {config}\n")

        try:
            result, stats = await retry_with_exponential_backoff(
                unstable_api_call,
                success_rate=0.4,
                **config
            )

            print(f"\n✓ Success!")
            print(f"Result: {result}")
            print(f"Stats: {stats}\n")

        except Exception as e:
            print(f"\n❌ Failed after all retries: {e}\n")

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

---

Background Tasks

Example 11: Application with Background Tasks

Manage background tasks with application lifecycle (aiohttp pattern).

import asyncio
from contextlib import suppress
from datetime import datetime

class Application:
    """
    Application with managed background tasks
    """

    def __init__(self):
        self.tasks = []
        self.should_exit = False
        self.data_processor_count = 0
        self.cleanup_count = 0

    async def data_processor(self):
        """Background task that processes data"""
        print("[DataProcessor] Started")

        try:
            while not self.should_exit:
                # Simulate data processing
                await asyncio.sleep(2)
                self.data_processor_count += 1
                print(f"[DataProcessor] Processed batch {self.data_processor_count}")

        except asyncio.CancelledError:
            print("[DataProcessor] Cancelled, cleaning up...")
            raise

    async def cleanup_task(self):
        """Background task that performs periodic cleanup"""
        print("[CleanupTask] Started")

        try:
            while not self.should_exit:
                await asyncio.sleep(5)
                self.cleanup_count += 1
                print(f"[CleanupTask] Cleanup {self.cleanup_count} completed")

        except asyncio.CancelledError:
            print("[CleanupTask] Cancelled")
            raise

    async def heartbeat(self):
        """Background task that sends heartbeat"""
        print("[Heartbeat] Started")

        try:
            while not self.should_exit:
                await asyncio.sleep(1)
                print(f"[Heartbeat] {datetime.now().strftime('%H:%M:%S')}")

        except asyncio.CancelledError:
            print("[Heartbeat] Cancelled")
            raise

    async def startup(self):
        """Start background tasks"""
        print("\n=== Application Startup ===")

        self.tasks = [
            asyncio.create_task(self.data_processor()),
            asyncio.create_task(self.cleanup_task()),
            asyncio.create_task(self.heartbeat())
        ]

        print(f"Started {len(self.tasks)} background tasks\n")

    async def shutdown(self):
        """Stop background tasks gracefully"""
        print("\n=== Application Shutdown ===")
        self.should_exit = True

        # Cancel all tasks
        for task in self.tasks:
            task.cancel()

        # Wait for all tasks to complete cancellation
        results = await asyncio.gather(*self.tasks, return_exceptions=True)

        # Report cancellation results
        for i, result in enumerate(results):
            if isinstance(result, asyncio.CancelledError):
                print(f"Task {i} cancelled successfully")
            elif isinstance(result, Exception):
                print(f"Task {i} raised exception: {result}")

        print("Shutdown complete\n")

    async def run(self, duration: int = 10):
        """Run application for specified duration"""
        await self.startup()

        # Run for specified duration
        await asyncio.sleep(duration)

        await self.shutdown()

        # Print stats
        print("=== Statistics ===")
        print(f"Data batches processed: {self.data_processor_count}")
        print(f"Cleanups performed: {self.cleanup_count}")

async def main():
    app = Application()
    await app.run(duration=12)

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

---

Testing Examples

Example 12: Testing Async Code with pytest-asyncio

import asyncio
import pytest
from typing import List

# Code to test
class AsyncCache:
    def __init__(self):
        self.cache = {}
        self.lock = asyncio.Lock()

    async def get(self, key: str):
        async with self.lock:
            await asyncio.sleep(0.01)  # Simulate I/O
            return self.cache.get(key)

    async def set(self, key: str, value: any):
        async with self.lock:
            await asyncio.sleep(0.01)  # Simulate I/O
            self.cache[key] = value

    async def delete(self, key: str):
        async with self.lock:
            await asyncio.sleep(0.01)  # Simulate I/O
            return self.cache.pop(key, None)

# Tests
@pytest.fixture
async def cache():
    """Fixture providing a cache instance"""
    return AsyncCache()

@pytest.mark.asyncio
async def test_cache_set_get(cache):
    """Test basic set/get operations"""
    await cache.set('key1', 'value1')
    result = await cache.get('key1')
    assert result == 'value1'

@pytest.mark.asyncio
async def test_cache_get_missing(cache):
    """Test getting non-existent key"""
    result = await cache.get('nonexistent')
    assert result is None

@pytest.mark.asyncio
async def test_cache_delete(cache):
    """Test delete operation"""
    await cache.set('key1', 'value1')
    deleted = await cache.delete('key1')
    assert deleted == 'value1'

    result = await cache.get('key1')
    assert result is None

@pytest.mark.asyncio
async def test_concurrent_access(cache):
    """Test concurrent cache access"""
    async def worker(cache, worker_id, iterations):
        for i in range(iterations):
            await cache.set(f'key_{worker_id}_{i}', f'value_{worker_id}_{i}')
            value = await cache.get(f'key_{worker_id}_{i}')
            assert value == f'value_{worker_id}_{i}'

    # Run 10 workers concurrently
    await asyncio.gather(*[
        worker(cache, i, 10) for i in range(10)
    ])

    # Verify all keys exist
    for worker_id in range(10):
        for i in range(10):
            key = f'key_{worker_id}_{i}'
            value = await cache.get(key)
            assert value == f'value_{worker_id}_{i}'

# Run tests with: pytest -v test_async_cache.py

---

Performance Examples

Example 13: Performance Comparison - Sequential vs Concurrent

import asyncio
import aiohttp
import time
from typing import List

async def fetch_sequential(urls: List[str]) -> List[dict]:
    """Fetch URLs sequentially"""
    results = []

    async with aiohttp.ClientSession() as session:
        for url in urls:
            try:
                async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as resp:
                    results.append({
                        'url': url,
                        'status': resp.status,
                        'size': len(await resp.text())
                    })
            except Exception as e:
                results.append({'url': url, 'error': str(e)})

    return results

async def fetch_concurrent(urls: List[str]) -> List[dict]:
    """Fetch URLs concurrently"""
    async def fetch_one(session, url):
        try:
            async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as resp:
                return {
                    'url': url,
                    'status': resp.status,
                    'size': len(await resp.text())
                }
        except Exception as e:
            return {'url': url, 'error': str(e)}

    async with aiohttp.ClientSession() as session:
        tasks = [fetch_one(session, url) for url in urls]
        results = await asyncio.gather(*tasks)

    return results

async def benchmark():
    """Compare sequential vs concurrent performance"""
    urls = [
        'http://python.org',
        'http://docs.python.org',
        'http://pypi.org',
        'http://github.com/python',
        'http://www.python.org/dev/peps/',
        'http://discuss.python.org',
        'http://peps.python.org',
        'http://wiki.python.org',
    ] * 2  # 16 total requests

    print(f"Benchmarking with {len(urls)} URLs...\n")

    # Sequential
    print("=== Sequential Execution ===")
    start = time.perf_counter()
    seq_results = await fetch_sequential(urls)
    seq_time = time.perf_counter() - start

    successful = sum(1 for r in seq_results if 'error' not in r)
    print(f"Time: {seq_time:.2f}s")
    print(f"Successful: {successful}/{len(urls)}\n")

    # Concurrent
    print("=== Concurrent Execution ===")
    start = time.perf_counter()
    conc_results = await fetch_concurrent(urls)
    conc_time = time.perf_counter() - start

    successful = sum(1 for r in conc_results if 'error' not in r)
    print(f"Time: {conc_time:.2f}s")
    print(f"Successful: {successful}/{len(urls)}\n")

    # Results
    print("=== Results ===")
    print(f"Sequential: {seq_time:.2f}s")
    print(f"Concurrent: {conc_time:.2f}s")
    print(f"Speedup: {seq_time / conc_time:.2f}x faster")
    print(f"Time saved: {seq_time - conc_time:.2f}s")

if __name__ == '__main__':
    asyncio.run(benchmark())

---

Production Patterns

Example 14: Circuit Breaker Pattern

Prevent cascading failures with circuit breaker.

import asyncio
from enum import Enum
from datetime import datetime, timedelta
from typing import Optional, Callable

class CircuitState(Enum):
    CLOSED = "closed"          # Normal operation
    OPEN = "open"              # Failing, reject requests
    HALF_OPEN = "half_open"    # Testing recovery

class CircuitBreaker:
    """
    Circuit breaker for protecting against cascading failures
    """

    def __init__(
        self,
        failure_threshold: int = 5,
        success_threshold: int = 2,
        timeout: float = 60.0,
        expected_exception: type = Exception
    ):
        self.failure_threshold = failure_threshold
        self.success_threshold = success_threshold
        self.timeout = timeout
        self.expected_exception = expected_exception

        self.state = CircuitState.CLOSED
        self.failure_count = 0
        self.success_count = 0
        self.last_failure_time: Optional[datetime] = None

    async def call(self, func: Callable, *args, **kwargs):
        """
        Execute function with circuit breaker protection
        """
        if self.state == CircuitState.OPEN:
            if self._should_attempt_reset():
                self.state = CircuitState.HALF_OPEN
                print(f"🔄 Circuit breaker entering HALF_OPEN state")
            else:
                raise Exception("Circuit breaker is OPEN")

        try:
            result = await func(*args, **kwargs)
            self._on_success()
            return result

        except self.expected_exception as e:
            self._on_failure()
            raise

    def _should_attempt_reset(self) -> bool:
        """Check if enough time has passed to attempt reset"""
        return (
            self.last_failure_time is not None and
            datetime.now() - self.last_failure_time >= timedelta(seconds=self.timeout)
        )

    def _on_success(self):
        """Handle successful call"""
        self.failure_count = 0

        if self.state == CircuitState.HALF_OPEN:
            self.success_count += 1

            if self.success_count >= self.success_threshold:
                self.state = CircuitState.CLOSED
                self.success_count = 0
                print(f"✓ Circuit breaker CLOSED (recovered)")

    def _on_failure(self):
        """Handle failed call"""
        self.failure_count += 1
        self.last_failure_time = datetime.now()

        if self.failure_count >= self.failure_threshold:
            self.state = CircuitState.OPEN
            print(f"⚠️  Circuit breaker OPEN (too many failures)")

    def get_state(self) -> dict:
        """Get current circuit breaker state"""
        return {
            'state': self.state.value,
            'failure_count': self.failure_count,
            'success_count': self.success_count,
            'last_failure': self.last_failure_time
        }

# Example usage
async def flaky_service(failure_rate: float = 0.5):
    """Simulate a flaky service"""
    import random
    await asyncio.sleep(0.1)

    if random.random() < failure_rate:
        raise ConnectionError("Service unavailable")

    return "Success"

async def main():
    breaker = CircuitBreaker(
        failure_threshold=3,
        success_threshold=2,
        timeout=5.0
    )

    print("=== Testing Circuit Breaker ===\n")

    for i in range(30):
        try:
            result = await breaker.call(flaky_service, failure_rate=0.6)
            print(f"Request {i}: {result} - State: {breaker.state.value}")

        except Exception as e:
            print(f"Request {i}: Failed ({e}) - State: {breaker.state.value}")

        await asyncio.sleep(0.5)

    print(f"\n=== Final State ===")
    state = breaker.get_state()
    for key, value in state.items():
        print(f"{key}: {value}")

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

Example 15: Graceful Shutdown

Handle shutdown signals gracefully.

import asyncio
import signal
from contextlib import suppress

class GracefulApplication:
    """
    Application with graceful shutdown handling
    """

    def __init__(self):
        self.shutdown_event = asyncio.Event()
        self.tasks = []

    async def worker(self, name: str):
        """Long-running worker task"""
        print(f"[{name}] Started")

        try:
            counter = 0
            while not self.shutdown_event.is_set():
                await asyncio.sleep(1)
                counter += 1
                print(f"[{name}] Working... ({counter})")

        except asyncio.CancelledError:
            print(f"[{name}] Received cancellation")
            # Cleanup logic here
            await asyncio.sleep(0.5)  # Simulate cleanup
            print(f"[{name}] Cleanup complete")
            raise

    def handle_signal(self, sig):
        """Handle shutdown signals"""
        print(f"\n⚠️  Received signal {signal.Signals(sig).name}")
        print("Initiating graceful shutdown...")
        self.shutdown_event.set()

    async def run(self):
        """Run application with signal handling"""
        # Setup signal handlers
        loop = asyncio.get_running_loop()

        for sig in (signal.SIGTERM, signal.SIGINT):
            loop.add_signal_handler(
                sig,
                lambda s=sig: self.handle_signal(s)
            )

        # Start workers
        print("=== Application Starting ===\n")
        self.tasks = [
            asyncio.create_task(self.worker(f'Worker-{i}'))
            for i in range(3)
        ]

        # Wait for shutdown signal
        await self.shutdown_event.wait()

        # Graceful shutdown
        print("\n=== Graceful Shutdown ===")

        # Give tasks time to finish current work
        print("Waiting for tasks to complete current work...")
        await asyncio.sleep(2)

        # Cancel tasks
        print("Cancelling tasks...")
        for task in self.tasks:
            task.cancel()

        # Wait for all cancellations
        results = await asyncio.gather(*self.tasks, return_exceptions=True)

        # Report results
        for i, result in enumerate(results):
            if isinstance(result, asyncio.CancelledError):
                print(f"✓ Task {i} cancelled successfully")
            elif isinstance(result, Exception):
                print(f"✗ Task {i} raised: {result}")

        print("\n=== Shutdown Complete ===")

async def main():
    app = GracefulApplication()
    await app.run()

if __name__ == '__main__':
    print("Press Ctrl+C to trigger graceful shutdown\n")
    asyncio.run(main())

---

Additional Resources

• Official asyncio Documentation: https://docs.python.org/3/library/asyncio.html
• aiohttp Documentation: https://docs.aiohttp.org/
• Real Python asyncio Tutorial: https://realpython.com/async-io-python/
• pytest-asyncio: https://github.com/pytest-dev/pytest-asyncio

---

Total Examples: 15 comprehensive, production-ready examples Coverage: HTTP clients, WebSockets, databases, queues, concurrency control, error handling, background tasks, testing, performance, and production patterns

Asyncio Concurrency Patterns

Master Python's asyncio library and concurrent programming for building high-performance asynchronous applications.

What is Asyncio?

Asyncio is Python's built-in framework for writing concurrent code using the async/await syntax. It enables you to write programs that handle thousands of I/O operations simultaneously without using threads or multiple processes.

Key Benefits:

• High Concurrency: Handle thousands of concurrent connections with minimal overhead
• Efficient I/O: Non-blocking I/O operations free up the CPU for other tasks
• Simple Syntax: Clean async/await syntax makes asynchronous code readable
• Rich Ecosystem: Libraries like aiohttp, aiofiles, asyncpg for async operations
• Single-threaded: Avoid complexity of thread synchronization and race conditions

Quick Start

Basic Async/Await

import asyncio

async def say_hello():
    print('Hello')
    await asyncio.sleep(1)
    print('World')

# Run the coroutine
asyncio.run(say_hello())

Concurrent Execution

import asyncio

async def task(name, duration):
    print(f'{name} starting')
    await asyncio.sleep(duration)
    print(f'{name} finished')
    return f'{name} result'

async def main():
    # Run three tasks concurrently
    results = await asyncio.gather(
        task('Task 1', 2),
        task('Task 2', 1),
        task('Task 3', 3)
    )
    print(f'All results: {results}')

asyncio.run(main())
# Task 1, 2, 3 all run concurrently - total time ~3 seconds

HTTP Requests with aiohttp

import asyncio
import aiohttp

async def fetch(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()

async def main():
    html = await fetch('http://python.org')
    print(f'Downloaded {len(html)} bytes')

asyncio.run(main())

Core Concepts

Event Loop

The event loop is the heart of asyncio - it schedules and executes asynchronous tasks:

import asyncio

# Modern way (Python 3.7+)
asyncio.run(main())

# Manual loop management (older code)
loop = asyncio.get_event_loop()
loop.run_until_complete(main())
loop.close()

Coroutines

Coroutines are functions defined with async def that can be paused and resumed:

async def my_coroutine():
    # This is a coroutine
    await asyncio.sleep(1)
    return 'done'

# Calling a coroutine returns a coroutine object
coro = my_coroutine()

# You must await it or schedule it
result = await coro  # In async context
# or
result = asyncio.run(coro)  # From synchronous code

Tasks

Tasks wrap coroutines and schedule them to run on the event loop:

async def background_work():
    while True:
        print('Working...')
        await asyncio.sleep(1)

async def main():
    # Create task - starts immediately
    task = asyncio.create_task(background_work())

    # Do other work
    await asyncio.sleep(5)

    # Cancel background task
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print('Task cancelled')

Common Patterns

Pattern 1: Parallel Requests

Fetch multiple URLs concurrently:

import asyncio
import aiohttp

async def fetch_all(urls):
    async with aiohttp.ClientSession() as session:
        tasks = [session.get(url) for url in urls]
        responses = await asyncio.gather(*tasks)

        contents = []
        for resp in responses:
            content = await resp.text()
            contents.append(content)
            await resp.release()

        return contents

urls = ['http://example.com', 'http://python.org', 'http://github.com']
results = asyncio.run(fetch_all(urls))

Pattern 2: Rate Limiting

Control concurrency with semaphores:

import asyncio
import aiohttp

async def fetch_with_limit(session, url, semaphore):
    async with semaphore:  # Only N requests at a time
        async with session.get(url) as resp:
            return await resp.text()

async def main():
    semaphore = asyncio.Semaphore(5)  # Max 5 concurrent requests

    urls = [f'http://api.example.com/item/{i}' for i in range(100)]

    async with aiohttp.ClientSession() as session:
        tasks = [
            fetch_with_limit(session, url, semaphore)
            for url in urls
        ]
        results = await asyncio.gather(*tasks)

    return results

Pattern 3: Producer/Consumer

Coordinate work with queues:

import asyncio

async def producer(queue, n):
    for i in range(n):
        await asyncio.sleep(0.1)
        await queue.put(f'item-{i}')
        print(f'Produced: item-{i}')
    await queue.put(None)  # Signal completion

async def consumer(queue, name):
    while True:
        item = await queue.get()
        if item is None:
            await queue.put(None)  # Propagate to other consumers
            break
        print(f'{name} consumed: {item}')
        await asyncio.sleep(0.2)

async def main():
    queue = asyncio.Queue()
    await asyncio.gather(
        producer(queue, 10),
        consumer(queue, 'Consumer-1'),
        consumer(queue, 'Consumer-2')
    )

asyncio.run(main())

Pattern 4: Timeout Handling

Set timeouts for operations:

import asyncio

async def slow_operation():
    await asyncio.sleep(10)
    return 'done'

async def main():
    try:
        result = await asyncio.wait_for(slow_operation(), timeout=5.0)
    except asyncio.TimeoutError:
        print('Operation timed out')
        result = None

    return result

asyncio.run(main())

Pattern 5: Background Tasks

Run tasks in the background while handling requests:

import asyncio

async def background_task():
    while True:
        print('Background work...')
        await asyncio.sleep(5)

async def main():
    # Start background task
    task = asyncio.create_task(background_task())

    # Main work
    await asyncio.sleep(20)

    # Cleanup
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        pass

asyncio.run(main())

When to Use Asyncio

Good Use Cases:

• Web Servers & APIs: Handle many concurrent connections (aiohttp, FastAPI)
• HTTP Clients: Make many concurrent requests (web scraping, API clients)
• WebSockets: Real-time bidirectional communication
• Database Operations: Async database drivers (asyncpg, motor)
• File I/O: Reading/writing many files concurrently (aiofiles)
• Chat Applications: Handle multiple concurrent connections
• Background Jobs: Process tasks from queues
• Real-time Data: Streaming data, live updates
• Microservices: Service-to-service communication

Not Ideal For:

• CPU-Intensive Work: Use multiprocessing instead
• Blocking Libraries: Must use thread executors
• Simple Scripts: Overhead not worth it for simple tasks
• Legacy Code: If you can't use async libraries

Performance Tips

1. Reuse Sessions

# BAD - Creates new session per request
async def bad():
    for url in urls:
        async with aiohttp.ClientSession() as session:
            async with session.get(url) as resp:
                data = await resp.text()

# GOOD - Reuse session
async def good():
    async with aiohttp.ClientSession() as session:
        for url in urls:
            async with session.get(url) as resp:
                data = await resp.text()

2. Use Gather for Concurrency

# BAD - Sequential execution
async def bad():
    results = []
    for url in urls:
        result = await fetch(url)  # Waits for each
        results.append(result)

# GOOD - Concurrent execution
async def good():
    tasks = [fetch(url) for url in urls]
    results = await asyncio.gather(*tasks)  # All at once

3. Limit Concurrency

# Control resource usage with semaphores
semaphore = asyncio.Semaphore(10)

async def limited_fetch(url):
    async with semaphore:
        return await fetch(url)

4. Proper Cleanup

# Always clean up resources
async def main():
    async with aiohttp.ClientSession() as session:
        # Use session
        pass
    # Session closed automatically

    # Allow connections to close
    await asyncio.sleep(0)

Common Pitfalls

1. Forgetting await

# WRONG - Returns coroutine, doesn't execute
result = async_function()

# RIGHT - Executes and gets result
result = await async_function()

2. Blocking the Event Loop

import time

# WRONG - Blocks entire event loop
async def bad():
    time.sleep(5)  # Everything stops!

# RIGHT - Yields control
async def good():
    await asyncio.sleep(5)  # Other tasks run

3. Creating Sessions Outside Event Loop

# WRONG - Session created before loop exists
session = aiohttp.ClientSession()

async def fetch(url):
    async with session.get(url) as resp:
        return await resp.text()

# RIGHT - Create inside async function
async def fetch(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as resp:
            return await resp.text()

4. Not Handling Cancellation

# WRONG - Doesn't cleanup on cancel
async def bad_task():
    while True:
        await asyncio.sleep(1)
        process()  # Resources may leak

# RIGHT - Handles cancellation
async def good_task():
    try:
        while True:
            await asyncio.sleep(1)
            process()
    except asyncio.CancelledError:
        cleanup()  # Cleanup resources
        raise

Testing Async Code

Using pytest-asyncio

import pytest

@pytest.mark.asyncio
async def test_fetch():
    result = await fetch_data()
    assert result == 'expected'

@pytest.mark.asyncio
async def test_with_fixture(aiohttp_client):
    app = create_app()
    client = await aiohttp_client(app)

    resp = await client.get('/')
    assert resp.status == 200

    data = await resp.json()
    assert data['key'] == 'value'

Manual Testing

import asyncio

def test_async_function():
    async def test_impl():
        result = await my_async_function()
        assert result == 'expected'

    asyncio.run(test_impl())

Debugging

Enable Debug Mode

# Shows warnings about unawaited coroutines
asyncio.run(main(), debug=True)

# Or manually
import asyncio
import logging

logging.basicConfig(level=logging.DEBUG)
loop = asyncio.get_event_loop()
loop.set_debug(True)

What Debug Mode Catches

• Coroutines that were never awaited
• Callbacks that take too long (> 100ms)
• Tasks destroyed while still pending
• Exceptions in callbacks

Real-World Example: API Client

import asyncio
import aiohttp
from typing import List, Dict

class AsyncAPIClient:
    def __init__(self, base_url: str, max_concurrent: int = 10):
        self.base_url = base_url
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.session = None

    async def __aenter__(self):
        self.session = aiohttp.ClientSession()
        return self

    async def __aexit__(self, *args):
        await self.session.close()
        await asyncio.sleep(0)

    async def get(self, endpoint: str) -> Dict:
        """Make GET request with rate limiting"""
        async with self.semaphore:
            url = f'{self.base_url}/{endpoint}'
            async with self.session.get(url) as resp:
                resp.raise_for_status()
                return await resp.json()

    async def get_many(self, endpoints: List[str]) -> List[Dict]:
        """Fetch multiple endpoints concurrently"""
        tasks = [self.get(endpoint) for endpoint in endpoints]
        return await asyncio.gather(*tasks, return_exceptions=True)

# Usage
async def main():
    async with AsyncAPIClient('https://api.example.com', max_concurrent=5) as client:
        # Fetch single endpoint
        user = await client.get('users/123')

        # Fetch multiple endpoints concurrently
        endpoints = [f'users/{i}' for i in range(1, 11)]
        users = await client.get_many(endpoints)

        for user in users:
            if isinstance(user, Exception):
                print(f'Error: {user}')
            else:
                print(f"User: {user.get('name')}")

asyncio.run(main())

Libraries to Know

Core Async Libraries

• aiohttp: Async HTTP client/server framework
• aiofiles: Async file operations
• asyncpg: Async PostgreSQL driver
• motor: Async MongoDB driver
• aiomysql: Async MySQL driver
• aioredis: Async Redis client

Web Frameworks

• FastAPI: Modern async web framework
• Sanic: Async web server
• Quart: Async Flask equivalent
• Starlette: ASGI framework

Testing

• pytest-asyncio: Pytest plugin for async tests
• aioresponses: Mock aiohttp requests
• asynctest: Async mocking utilities

Next Steps

1. Read the Full SKILL.md: Comprehensive guide with all patterns 2. Check EXAMPLES.md: 18+ practical examples with full code 3. Practice: Build a small async project (web scraper, API client) 4. Learn aiohttp: Most common async library for HTTP 5. Study Real Projects: Look at FastAPI, Sanic source code

Resources

• Official asyncio Docs: https://docs.python.org/3/library/asyncio.html
• aiohttp Documentation: https://docs.aiohttp.org/
• Real Python Tutorial: https://realpython.com/async-io-python/
• PEP 492: https://www.python.org/dev/peps/pep-0492/

Quick Reference

# Run async function
asyncio.run(main())

# Create task
task = asyncio.create_task(coro())

# Run concurrently
results = await asyncio.gather(coro1(), coro2(), coro3())

# Timeout
result = await asyncio.wait_for(coro(), timeout=5.0)

# Semaphore (limit concurrency)
sem = asyncio.Semaphore(10)
async with sem:
    await operation()

# Queue (producer/consumer)
queue = asyncio.Queue()
await queue.put(item)
item = await queue.get()

# Lock (mutual exclusion)
lock = asyncio.Lock()
async with lock:
    # critical section
    pass

# Event (signaling)
event = asyncio.Event()
await event.wait()
event.set()

# Sleep (non-blocking)
await asyncio.sleep(1)

---

Get Started: Read SKILL.md for comprehensive coverage and EXAMPLES.md for practical code examples.