async-python-patterns

Async Python Patterns

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

When to Invoke

• 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
• Deciding between threading, multiprocessing, and asyncio

Core Concepts

Event Loop

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

Coroutines

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

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

Tasks

Scheduled coroutines that run concurrently on the event loop.

Futures

Low-level objects representing eventual results of async operations.

Async Context Managers

Resources that support async with for proper cleanup.

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

asyncio.run(main())

Fundamental Patterns

Basic Async/Await

import asyncio

async def fetch_data(url: str) -> dict:
    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 with gather()

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]:
    tasks = [fetch_user(uid) for uid in user_ids]
    return await asyncio.gather(*tasks)

asyncio.run(fetch_all_users([1, 2, 3, 4, 5]))