async-python-patterns

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.

Sync vs Async Decision Guide

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

Use Case	Recommended Approach
Many concurrent network/DB calls	asyncio
CPU-bound computation	multiprocessing or thread pool
Mixed I/O + CPU	Offload CPU work with asyncio.to_thread()
Simple scripts, few connections	Sync (simpler, easier to debug)
Web APIs with high concurrency	Async frameworks (FastAPI, aiohttp)

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