Async Patterns & Best Practices

Master async context managers, iterators, comprehensive error handling, and production patterns

Beyond the Basics

Once you understand async fundamentals, the next step is mastering advanced patterns that make your async code robust, maintainable, and production-ready. This lesson covers async context managers for resource management, async iterators for streaming data, comprehensive error handling strategies, and battle-tested patterns used in real-world systems.

What You'll Master:

Async Context Managers: Properly manage resources with async with
Async Iterators & Generators: Stream data efficiently
Error Handling: Catch and recover from failures gracefully
Production Patterns: Connection pools, circuit breakers, retries
Testing Async Code: Strategies for reliable async tests

Async Context Managers

Async context managers ensure proper setup and cleanup of async resources like database connections, file handles, and network connections.

Using async with

import aiohttp
import asyncio

# Async context manager ensures proper cleanup
async def fetch_data(url):
    # aiohttp session is an async context manager
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            data = await response.json()
            return data
    # Session automatically closed here, even if error occurs

async def main():
    result = await fetch_data('https://api.example.com/data')
    print(result)

asyncio.run(main())

# Why async with?
# - Ensures cleanup happens (like finally block)
# - Works with await in __aenter__ and __aexit__
# - Prevents resource leaks

Creating Custom Async Context Managers

import asyncio
from contextlib import asynccontextmanager

# Method 1: Using __aenter__ and __aexit__
class AsyncDatabaseConnection:
    def __init__(self, host):
        self.host = host
        self.connection = None

    async def __aenter__(self):
        """Called when entering 'async with' block"""
        print(f"Connecting to {self.host}...")
        await asyncio.sleep(1)  # Simulate connection
        self.connection = f"Connected to {self.host}"
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        """Called when exiting 'async with' block"""
        print(f"Closing connection to {self.host}...")
        await asyncio.sleep(0.5)  # Simulate cleanup
        self.connection = None
        return False  # Don't suppress exceptions

# Usage
async def main():
    async with AsyncDatabaseConnection("localhost") as db:
        print(f"Using: {db.connection}")
        # Do work with connection
    # Connection automatically closed

# Method 2: Using @asynccontextmanager decorator
@asynccontextmanager
async def database_connection(host):
    """Simpler way to create async context manager"""
    # Setup
    print(f"Connecting to {host}...")
    connection = await connect_to_database(host)

    try:
        yield connection  # Provide resource
    finally:
        # Cleanup (always runs)
        print(f"Closing connection to {host}...")
        await connection.close()

# Usage
async def query_database():
    async with database_connection("localhost") as conn:
        result = await conn.query("SELECT * FROM users")
        return result

asyncio.run(main())

Real-World Example: Connection Pool

import asyncio
from contextlib import asynccontextmanager

class AsyncConnectionPool:
    def __init__(self, max_connections=10):
        self.max_connections = max_connections
        self.available = []
        self.in_use = set()
        self.semaphore = asyncio.Semaphore(max_connections)

    async def create_connection(self):
        """Create a new connection"""
        await asyncio.sleep(0.1)  # Simulate connection time
        return f"Connection-{id(object())}"

    @asynccontextmanager
    async def acquire(self):
        """Acquire a connection from the pool"""
        async with self.semaphore:
            # Get existing or create new
            if self.available:
                conn = self.available.pop()
            else:
                conn = await self.create_connection()

            self.in_use.add(conn)

            try:
                yield conn
            finally:
                # Return to pool
                self.in_use.remove(conn)
                self.available.append(conn)

# Usage
async def worker(pool, worker_id):
    async with pool.acquire() as conn:
        print(f"Worker {worker_id} using {conn}")
        await asyncio.sleep(1)  # Simulate work
        print(f"Worker {worker_id} done")

async def main():
    pool = AsyncConnectionPool(max_connections=3)

    # 10 workers sharing 3 connections
    await asyncio.gather(*[
        worker(pool, i) for i in range(10)
    ])

asyncio.run(main())

Key benefits:

Guarantees cleanup even if exceptions occur
Makes resource management explicit and safe
Prevents resource leaks in long-running applications
Works seamlessly with async operations

Async Iterators & Generators

Async iterators allow you to iterate over data that's fetched asynchronously, perfect for streaming large datasets, paginated APIs, or real-time data feeds.

async for: Consuming Async Iterators

import asyncio

# Async generator function
async def fetch_pages(num_pages):
    """Simulate fetching pages one at a time"""
    for page in range(1, num_pages + 1):
        print(f"Fetching page {page}...")
        await asyncio.sleep(1)  # Simulate API call
        yield {"page": page, "data": f"Content of page {page}"}

# Consuming with async for
async def main():
    async for page in fetch_pages(5):
        print(f"Processing: {page}")
        # Process each page as it arrives
        # Memory efficient - one page at a time

asyncio.run(main())

# Output:
# Fetching page 1...
# Processing: {'page': 1, 'data': 'Content of page 1'}
# Fetching page 2...
# Processing: {'page': 2, 'data': 'Content of page 2'}
# ...

Creating Async Iterators

import asyncio
import aiohttp

# Method 1: Using __aiter__ and __anext__
class AsyncAPIIterator:
    def __init__(self, base_url, max_pages):
        self.base_url = base_url
        self.max_pages = max_pages
        self.current_page = 1

    def __aiter__(self):
        """Return the iterator object (self)"""
        return self

    async def __anext__(self):
        """Return the next item or raise StopAsyncIteration"""
        if self.current_page > self.max_pages:
            raise StopAsyncIteration

        # Fetch page asynchronously
        url = f"{self.base_url}?page={self.current_page}"
        async with aiohttp.ClientSession() as session:
            async with session.get(url) as response:
                data = await response.json()

        self.current_page += 1
        return data

# Usage
async def process_api_data():
    async for page_data in AsyncAPIIterator("https://api.example.com/data", 5):
        print(f"Processing page: {page_data}")

# Method 2: Async Generator (simpler)
async def paginated_api(base_url, max_pages):
    """Async generator - cleaner syntax"""
    for page in range(1, max_pages + 1):
        async with aiohttp.ClientSession() as session:
            async with session.get(f"{base_url}?page={page}") as response:
                data = await response.json()
                yield data

# Usage
async def main():
    async for data in paginated_api("https://api.example.com/data", 5):
        print(data)

asyncio.run(main())

Real-World: Streaming Database Results

import asyncio
import asyncpg

async def stream_large_table(pool, batch_size=1000):
    """Stream millions of rows without loading all into memory"""
    async with pool.acquire() as conn:
        # Use cursor for efficient streaming
        async with conn.transaction():
            cursor = await conn.cursor(
                "SELECT * FROM large_table ORDER BY id"
            )

            while True:
                # Fetch batch
                rows = await cursor.fetch(batch_size)
                if not rows:
                    break

                # Yield batch for processing
                for row in rows:
                    yield row

# Usage
async def process_records():
    pool = await asyncpg.create_pool(
        'postgresql://user:pass@localhost/db'
    )

    processed = 0
    async for record in stream_large_table(pool):
        # Process one record at a time
        await process_record(record)
        processed += 1

        if processed % 10000 == 0:
            print(f"Processed {processed} records...")

    await pool.close()
    print(f"Total processed: {processed}")

asyncio.run(process_records())

# Benefits:
# - Low memory usage (only batch_size in memory)
# - Can process billions of rows
# - Start processing immediately (no wait for full load)

Comprehensive Error Handling

Async code needs robust error handling. Unhandled exceptions in tasks can cause silent failures, and errors in concurrent tasks need special attention.

Try/Except in Async Functions

import asyncio
import aiohttp

async def fetch_with_error_handling(url):
    """Proper error handling in async function"""
    try:
        async with aiohttp.ClientSession() as session:
            async with session.get(url, timeout=10) as response:
                response.raise_for_status()  # Raise for 4xx/5xx
                return await response.json()

    except asyncio.TimeoutError:
        print(f"Timeout fetching {url}")
        return None

    except aiohttp.ClientError as e:
        print(f"HTTP error for {url}: {e}")
        return None

    except Exception as e:
        print(f"Unexpected error for {url}: {e}")
        return None

    finally:
        # Cleanup code (always runs)
        print(f"Finished attempt for {url}")

# Usage
async def main():
    result = await fetch_with_error_handling("https://api.example.com/data")
    if result:
        print(f"Success: {result}")
    else:
        print("Failed to fetch data")

asyncio.run(main())

Handling Errors in Concurrent Tasks

import asyncio

async def risky_task(task_id):
    """Task that might fail"""
    await asyncio.sleep(1)
    if task_id == 2:
        raise ValueError(f"Task {task_id} failed!")
    return f"Task {task_id} succeeded"

# Method 1: gather() with return_exceptions
async def main_gather():
    results = await asyncio.gather(
        risky_task(1),
        risky_task(2),  # This will fail
        risky_task(3),
        return_exceptions=True  # Don't crash, return exceptions
    )

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

# Method 2: Wrapping tasks individually
async def safe_task(task_id):
    """Wrapper with error handling"""
    try:
        return await risky_task(task_id)
    except Exception as e:
        print(f"Caught error in task {task_id}: {e}")
        return None  # Return default value

async def main_wrapped():
    results = await asyncio.gather(
        safe_task(1),
        safe_task(2),  # Error caught internally
        safe_task(3)
    )
    print(f"All results: {results}")

# Method 3: Using create_task with error handling
async def main_tasks():
    tasks = [
        asyncio.create_task(safe_task(i))
        for i in range(1, 4)
    ]

    results = await asyncio.gather(*tasks)
    print(f"Results: {results}")

asyncio.run(main_gather())

Retry Pattern with Exponential Backoff

import asyncio
import random

async def retry_with_backoff(
    coro_func,
    max_retries=3,
    base_delay=1,
    max_delay=60,
    exponential_base=2
):
    """Retry with exponential backoff"""
    for attempt in range(max_retries):
        try:
            return await coro_func()

        except Exception as e:
            if attempt == max_retries - 1:
                # Last attempt, re-raise
                raise

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

            # Add jitter to prevent thundering herd
            jitter = random.uniform(0, delay * 0.1)
            total_delay = delay + jitter

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

            await asyncio.sleep(total_delay)

# Usage
async def unreliable_api_call():
    """Simulates unreliable API"""
    if random.random() < 0.7:
        raise Exception("API temporarily unavailable")
    return {"data": "success"}

async def main():
    try:
        result = await retry_with_backoff(
            unreliable_api_call,
            max_retries=5,
            base_delay=1
        )
        print(f"Success: {result}")
    except Exception as e:
        print(f"Failed after all retries: {e}")

asyncio.run(main())

Production-Ready Patterns

Timeout Context Manager

Prevent operations from hanging indefinitely.

import asyncio

async def operation_with_timeout():
    try:
        # Timeout after 5 seconds
        async with asyncio.timeout(5):
            result = await slow_operation()
            return result
    except asyncio.TimeoutError:
        print("Operation timed out!")
        return None

# Alternative: wait_for
async def alternative_timeout():
    try:
        result = await asyncio.wait_for(
            slow_operation(),
            timeout=5.0
        )
        return result
    except asyncio.TimeoutError:
        print("Operation timed out!")
        return None

# Nested timeouts
async def nested_timeouts():
    try:
        # Outer timeout: 10 seconds
        async with asyncio.timeout(10):
            # Inner timeout: 5 seconds
            async with asyncio.timeout(5):
                result = await slow_operation()

            # This has 5 seconds remaining
            await another_operation()
    except asyncio.TimeoutError:
        print("Operation exceeded timeout")

asyncio.run(operation_with_timeout())

Testing Async Code

Testing async code requires special considerations and tools.

import asyncio
import pytest

# Using pytest-asyncio
@pytest.mark.asyncio
async def test_async_function():
    """Test async function with pytest"""
    result = await fetch_data("https://api.example.com")
    assert result["status"] == "success"

# Testing with mocks
from unittest.mock import AsyncMock, patch

@pytest.mark.asyncio
async def test_with_mock():
    """Test with mocked async calls"""
    mock_response = AsyncMock()
    mock_response.json.return_value = {"data": "test"}

    with patch('aiohttp.ClientSession.get', return_value=mock_response):
        result = await fetch_data("https://api.example.com")
        assert result["data"] == "test"

# Testing timeouts
@pytest.mark.asyncio
async def test_timeout():
    """Test that operation times out"""
    with pytest.raises(asyncio.TimeoutError):
        await asyncio.wait_for(slow_operation(), timeout=1.0)

# Testing concurrent operations
@pytest.mark.asyncio
async def test_concurrent_execution():
    """Test multiple concurrent operations"""
    start = asyncio.get_event_loop().time()

    results = await asyncio.gather(
        task(1), task(2), task(3)
    )

    elapsed = asyncio.get_event_loop().time() - start

    assert len(results) == 3
    assert elapsed < 2.0  # Should run concurrently

# Fixture for async setup/teardown
@pytest.fixture
async def db_connection():
    """Async fixture"""
    conn = await create_connection()
    yield conn
    await conn.close()

@pytest.mark.asyncio
async def test_with_fixture(db_connection):
    """Use async fixture"""
    result = await db_connection.query("SELECT 1")
    assert result is not None

Async Best Practices

✓ Do This

Use async context managers for resources
Always handle exceptions in tasks
Use return_exceptions=True in gather()
Implement timeouts for all I/O operations
Use async iterators for streaming data
Implement retry logic with backoff
Use circuit breakers for external services
Limit concurrency with semaphores
Log errors with context
Test async code thoroughly

✗ Avoid This

Letting exceptions crash tasks silently
No timeouts on I/O operations
Unbounded concurrency
Mixing blocking and async code
Not using async context managers
Ignoring backpressure
No retry logic for transient failures
Creating too many tasks at once
Not testing edge cases
Swallowing errors without logging

Key Takeaways

Async context managers ensure cleanup - use async with for resources
Async iterators enable streaming - process data as it arrives
Error handling is critical - use return_exceptions and try/except
Implement retry logic - exponential backoff for resilience
Use circuit breakers - prevent cascading failures
Timeouts prevent hangs - always set reasonable timeouts
Worker pools limit concurrency - prevent resource exhaustion
Test async code properly - use pytest-asyncio and mocks

Practice Exercises

Exercise 1: Async Database Pool

Build an async connection pool for database connections with proper context manager support. Implement connection timeout, max pool size, and connection recycling. Test with concurrent queries.

Exercise 2: Resilient API Client

Create an async API client with circuit breaker, retry logic with exponential backoff, rate limiting using semaphore, and comprehensive error handling. Should gracefully handle timeouts, network errors, and API rate limits.

Exercise 3: Streaming Data Processor

Build an async data processor that streams records from a paginated API using async iterators, processes them in batches using a worker pool, and handles errors without stopping the entire pipeline. Include metrics (records processed, errors encountered).

Additional Resources

asyncio docs: docs.python.org/3/library/asyncio.html
aiohttp: docs.aiohttp.org - async HTTP client/server
pytest-asyncio: Testing async code
tenacity: Retry library for Python
Design Patterns: "Designing Data-Intensive Applications" by Martin Kleppmann
Circuit Breakers: martinfowler.com/bliki/CircuitBreaker.html

What's Next?

You've mastered async patterns! Now let's explore threading and concurrency to handle parallel execution with threads.

Threading Basics - Create and manage threads for concurrent execution
Thread Synchronization - Locks, semaphores, and thread-safe code
GIL Understanding - Learn about Python's Global Interpreter Lock

Back to "Async Fundamentals"

Python AdvancedLesson 2

Next: "Threading & Concurrency"