async-python-patterns

Async Python Patterns

Expert 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
• Optimizing I/O-bound workloads requiring parallelism
• Implementing async background tasks and task queues

Core Patterns

1. Basic Async/Await

Foundation for all async operations:

import asyncio

async def fetch_data(url: str) -> dict:
    """Fetch data 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())

Key concepts:

• async def defines coroutines (pausable functions)
• await yields control back to event loop
• asyncio.run() is the entry point (Python 3.7+)
• Single-threaded cooperative multitasking

2. Concurrent Execution with gather()

Execute multiple operations simultaneously:

import asyncio
from typing import List

async def fetch_user(user_id: int) -> dict:
    await asyncio.sleep(0.5)
    return {"id": user_id, "name": f"User {user_id}"}

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

# Speed: Sequential = 5s, Concurrent = 0.5s for 10 users

When to use:

• Independent operations that can run in parallel
• I/O-bound tasks (API calls, database queries)
• Need all results before proceeding
• Use return_exceptions=True to handle partial failures