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

Task Creation and Management

import asyncio

async def background_task(name: str, delay: int):
    await asyncio.sleep(delay)
    return f"Result from {name}"

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

    # Do other work while tasks run
    await asyncio.sleep(0.5)

    result1 = await task1
    result2 = await task2
    print(f"Results: {result1}, {result2}")

asyncio.run(main())

Error Handling

import asyncio
from typing import Optional

async def safe_operation(item_id: int) -> Optional[dict]:
    try:
        return await risky_operation(item_id)
    except ValueError as e:
        print(f"Error: {e}")
        return None

async def process_items(item_ids):
    tasks = [safe_operation(iid) for iid in item_ids]
    results = await asyncio.gather(*tasks, return_exceptions=True)

    successful = [r for r in results if r is not None and not isinstance(r, Exception)]
    failed = [r for r in results if isinstance(r, Exception)]
    return successful

Timeout Handling

import asyncio

async def with_timeout():
    try:
        result = await asyncio.wait_for(slow_operation(5), timeout=2.0)
    except asyncio.TimeoutError:
        print("Operation timed out")

Semaphore for Rate Limiting

import asyncio
from typing import List

async def api_call(url: str, semaphore: asyncio.Semaphore) -> dict:
    async with semaphore:
        await asyncio.sleep(0.5)  # Simulate API call
        return {"url": url, "status": 200}

async def rate_limited_requests(urls: List[str], max_concurrent: int = 5):
    semaphore = asyncio.Semaphore(max_concurrent)
    tasks = [api_call(url, semaphore) for url in urls]
    return await asyncio.gather(*tasks)

Decision Framework

When to Use asyncio

I/O-bound tasks (network, database, file)
Many concurrent connections
Web servers and API clients
WebSocket applications

When to Use threading

I/O-bound tasks with existing sync libraries
Simple parallelism needs
Interfacing with C extensions that release GIL

When to Use multiprocessing

CPU-bound tasks (computation, data processing)
Need true parallelism (bypasses GIL)
Tasks that don't need shared memory

Common Pitfalls

Forgetting await

# Wrong - returns coroutine object
result = async_function()

# Correct
result = await async_function()

Blocking the Event Loop

# Wrong - blocks event loop
import time
async def bad():
    time.sleep(1)  # Blocks!

# Correct
async def good():
    await asyncio.sleep(1)  # Non-blocking

Not Handling Cancellation

async def cancelable_task():
    try:
        while True:
            await asyncio.sleep(1)
    except asyncio.CancelledError:
        # Perform cleanup
        raise  # Re-raise to propagate

Mixing Sync and Async Code

# Wrong
def sync_function():
    result = await async_function()  # SyntaxError!

# Correct
def sync_function():
    result = asyncio.run(async_function())

Testing Async Code

import asyncio
import pytest

@pytest.mark.asyncio
async def test_async_function():
    result = await fetch_data("https://api.example.com")
    assert result is not None

@pytest.mark.asyncio
async def test_with_timeout():
    with pytest.raises(asyncio.TimeoutError):
        await asyncio.wait_for(slow_operation(5), timeout=1.0)

Best Practices

Use asyncio.run() for entry point (Python 3.7+)
Always await coroutines to execute them
Use gather() for concurrent execution of multiple tasks
Implement proper error handling with try/except
Use timeouts to prevent hanging operations
Pool connections for better performance
Avoid blocking operations in async code - use run_in_executor
Use semaphores for rate limiting
Handle task cancellation properly
Test async code with pytest-asyncio

References

references/async-patterns.md - async context managers, async iterators/generators, producer-consumer pattern, async locks and synchronization, web scraping with aiohttp, async database operations, WebSocket server implementation, connection pools, batch operations, running blocking operations in executors

Resources

Python asyncio documentation: https://docs.python.org/3/library/asyncio.html
aiohttp: Async HTTP client/server
FastAPI: Modern async web framework
asyncpg: Async PostgreSQL driver
motor: Async MongoDB driver

async-python-patterns