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

Sync vs Async Decision Guide

Before adopting async, consider whether it's the right choice for your use case.

Key Rule: Stay fully sync or fully async within a call path. Mixing creates hidden blocking and complexity.

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.

Syntax:

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

3. Tasks

Scheduled coroutines that run concurrently on the event loop.

4. Futures

Low-level objects representing eventual results of async operations.

5. Async Context Managers

Resources that support async with for proper cleanup.

6. Async Iterators

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

Quick Start

import asyncio

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

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

Fundamental Patterns

Pattern 1: Basic Async/Await

import asyncio

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())

Pattern 2: Concurrent Execution with gather()

import asyncio
from typing import List

async def fetch_user(user_id: int) -> dict:
    """Fetch user data."""
    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

async def main():
    user_ids = [1, 2, 3, 4, 5]
    users = await fetch_all_users(user_ids)
    print(f"Fetched {len(users)} users")

asyncio.run(main())

Pattern 3: Task Creation and Management

import asyncio

async def background_task(name: str, delay: int):
    """Long-running background task."""
    print(f"{name} started")
    await asyncio.sleep(delay)
    print(f"{name} completed")
    return f"Result from {name}"

async def main():
    # Create tasks
    task1 = asyncio.create_task(background_task("Task 1", 2))
    task2 = asyncio.create_task(background_task("Task 2", 1))

    # Do other work
    print("Main: doing other work")
    await asyncio.sleep(0.5)

    # Wait for tasks
    result1 = await task1
    result2 = await task2

    print(f"Results: {result1}, {result2}")

asyncio.run(main())

Pattern 4: Error Handling in Async Code

import asyncio
from typing import List, Optional

async def risky_operation(item_id: int) -> dict:
    """Operation that might fail."""
    await asyncio.sleep(0.1)
    if item_id % 3 == 0:
        raise ValueError(f"Item {item_id} failed")
    return {"id": item_id, "status": "success"}

async def safe_operation(item_id: int) -> Optional[dict]:
    """Wrapper with error handling."""
    try:
        return await risky_operation(item_id)

…