python-async-patterns

Python Async Patterns

Modern asynchronous programming patterns with asyncio for Python 3.7+.

Core Concepts

Async programming is for I/O-bound concurrency, not CPU-bound parallelism

Good use cases:

• HTTP requests (API calls, web scraping)
• Database queries
• File I/O
• Network operations

Bad use cases:

• CPU-intensive calculations (use multiprocessing)
• Simple sequential operations

Basic Async/Await

import asyncio

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

async def main():
    result = await fetch_data("https://example.com")
    
    # Run multiple coroutines concurrently
    results = await asyncio.gather(
        fetch_data("url1"),
        fetch_data("url2"),
        fetch_data("url3")
    )

asyncio.run(main())

Concurrent Execution Patterns

asyncio.gather() - Run Multiple Tasks

async def process_urls(urls):
    tasks = [fetch_data(url) for url in urls]
    results = await asyncio.gather(*tasks)
    return results

See concurrent-execution-patterns.md for:

• TaskGroup (3.11+)
• Timeout handling
• Semaphores for rate limiting

Async HTTP with aiohttp

import aiohttp

async def fetch_json(session, url):
    async with session.get(url) as response:
        return await response.json()

async def main():
    async with aiohttp.ClientSession() as session:
        results = await asyncio.gather(
            fetch_json(session, "url1"),
            fetch_json(session, "url2")
        )

See aiohttp-patterns.md for:

• POST requests with JSON
• Error handling
• Connection pooling
• Streaming responses

Async Context Managers

from contextlib import asynccontextmanager

@asynccontextmanager
async def get_session():
    session = aiohttp.ClientSession()
    try:
        yield session
    finally:
        await session.close()