Async Python Patterns

Comprehensive guidance for implementing asynchronous Python applications using asyncio, concurrent programming patterns, and async/await for building high-performance, non-blocking systems.

When to Use This Skill

• Building async web APIs (FastAPI, aiohttp, Sanic)
• Implementing concurrent I/O operations (database, file, network)
• Creating web scrapers with concurrent requests
• Developing real-time applications (WebSocket servers, chat systems)
• Processing multiple independent tasks simultaneously
• Building microservices with async communication
• Optimizing I/O-bound workloads
• Implementing async background tasks and queues

Quick Start

import asyncio

async def main():
    print("Hello")
    await asyncio.sleep(1)
    print("World")

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

Basic Async/Await

async def fetch_data(url: str) -> dict:
    """Fetch data from URL asynchronously."""
    await asyncio.sleep(1)  # Simulate I/O
    return {"url": url, "data": "result"}

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

asyncio.run(main())

Concurrent Execution

async def fetch_all_users(user_ids: List[int]) -> List[dict]:
    """Fetch multiple users concurrently."""
    tasks = [fetch_user(uid) for uid in user_ids]
    results = await asyncio.gather(*tasks)
    return results

Core Concepts

1. Event Loop

The event loop is the heart of asyncio, managing and scheduling asynchronous tasks.

Key characteristics:

• Single-threaded cooperative multitasking
• Schedules coroutines for execution
• Handles I/O operations without blocking
• Manages callbacks and futures

2. Coroutines

Functions defined with async def that can be paused and resumed.

async def my_coroutine():
    result = await some_async_operation()
    return result

3. Tasks

Scheduled coroutines that run concurrently on the event loop.

task = asyncio.create_task(background_task())
result = await task

4. Async Context Managers

Resources that support async with for proper cleanup.

async with AsyncDatabaseConnection("postgresql://localhost") as conn:
    result = await conn.query("SELECT * FROM users")

5. Async Iterators

Objects that support async for for iterating over async data sources.

async for item in async_range(1, 10):
    process(item)

See detailed explanations: Core Concepts

Fundamental Patterns

Pattern 1: Task Creation and Management

Create and manage concurrent tasks effectively.

task1 = asyncio.create_task(background_task("Task 1", 2))
task2 = asyncio.create_task(background_task("Task 2", 1))

result1 = await task1
result2 = await task2

See detailed patterns: Basic Patterns

Pattern 2: Error Handling

Handle errors in async code with proper exception handling.

async def safe_operation(item_id: int):
    try:
        return await risky_operation(item_id)
    except ValueError as e:
        print(f"Error: {e}")
        return None

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

See detailed patterns: Error Handling

Pattern 3: Timeout Handling

Execute operations with timeouts to prevent hanging.

try:
    result = await asyncio.wait_for(slow_operation(5), timeout=2.0)
except asyncio.TimeoutError:
    print("Operation timed out")

See detailed patterns: Timeouts and Cancellation

Advanced Patterns

Async Context Managers

Implement async context managers for resource management.

class AsyncDatabaseConnection:
    async def __aenter__(self):
        # Setup logic
        return connection

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        # Cleanup logic
        pass

See detailed implementations: Advanced Patterns

Producer-Consumer Pattern

Implement async queues for producer-consumer workflows.

queue = asyncio.Queue(maxsize=10)

async def producer(queue):
    for item in items:
        await queue.put(item)

async def consumer(queue):
    while True:
        item = await queue.get()
        process(item)
        queue.task_done()

See detailed patterns: Concurrency Patterns

Rate Limiting with Semaphore

Control concurrent access with semaphores.

semaphore = asyncio.Semaphore(5)

async def api_call(url, semaphore):
    async with semaphore:
        return await fetch(url)

See detailed implementations: Synchronization

Real-World Applications

Web Scraping with aiohttp

async def scrape_urls(urls: List[str]):
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        return results

See detailed examples: Real-World Applications

Async Database Operations

async def get_user_data(db, user_id):
    user, orders, profile = await asyncio.gather(
        db.fetch_user(user_id),
        db.fetch_orders(user_id),
        db.fetch_profile(user_id)
    )
    return {"user": user, "orders": orders, "profile": profile}

See detailed patterns: Real-World Applications

Performance Best Practices

1. Use Connection Pools: Reuse connections efficiently
2. Batch Operations: Process items in batches
3. Avoid Blocking: Run CPU-bound work in executors
4. Use Semaphores: Limit concurrent operations
5. Handle Cancellation: Clean up resources properly

See detailed guide: Performance Best Practices

Common Pitfalls

1. Forgetting await: Returns coroutine object, doesn't execute
2. Blocking the Event Loop: Using time.sleep() instead of asyncio.sleep()
3. Not Handling Cancellation: Tasks cancelled without cleanup
4. Mixing Sync and Async: Calling async from sync incorrectly

See detailed solutions: Common Pitfalls

Testing Async Code

import pytest

@pytest.mark.asyncio
async def test_async_function():
    result = await fetch_data("https://api.example.com")
    assert result is not None

@pytest.mark.asyncio
async def test_with_timeout():
    with pytest.raises(asyncio.TimeoutError):
        await asyncio.wait_for(slow_operation(5), timeout=1.0)

See detailed testing guide: Testing

Best Practices Summary

1. Use asyncio.run() for entry point (Python 3.7+)
2. Always await coroutines to execute them
3. Use gather() for concurrent execution
4. Implement proper error handling
5. Use timeouts to prevent hanging operations
6. Pool connections for better performance
7. Avoid blocking operations in async code
8. Use semaphores for rate limiting
9. Handle task cancellation properly
10. Test async code with pytest-asyncio

Resources

• Python asyncio documentation: https://docs.python.org/3/library/asyncio.html
• aiohttp: Async HTTP client/server
• FastAPI: Modern async web framework
• asyncpg: Async PostgreSQL driver
• motor: Async MongoDB driver