Python Async

Async Python Patterns

asyncio and async/await patterns for Python applications.

Quick Start

import asyncio

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

asyncio.run(main())

When To Use

• Building async web APIs (FastAPI, aiohttp)
• Implementing concurrent I/O operations
• Creating web scrapers with concurrent requests
• Developing real-time applications (WebSockets)
• Processing multiple independent tasks simultaneously
• Building microservices with async communication

When NOT To Use

• CPU-bound optimization - use python-performance instead
• Testing async code - use python-testing async module

Modules

This skill uses progressive loading. Content is organized into focused modules:

• See modules/basic-patterns.md - Core async/await, gather(), and task management
• See modules/concurrency-control.md - Semaphores and locks for rate limiting
• See modules/error-handling-timeouts.md - Error handling, timeouts, and cancellation
• See modules/advanced-patterns.md - Context managers, iterators, producer-consumer
• See modules/testing-async.md - Testing with pytest-asyncio
• See modules/real-world-applications.md - Web scraping and database operations
• See modules/pitfalls-best-practices.md - Common mistakes and best practices

Load specific modules based on your needs, or reference all for detailed guidance.

Exit Criteria

• Async patterns applied correctly
• No blocking operations in async code
• Proper error handling implemented
• Rate limiting configured where needed
• Tests pass with pytest-asyncio

Troubleshooting

Common Issues

RuntimeError: no current event loop Use asyncio.run() as the entry point. Avoid get_event_loop() in Python 3.10+.

Blocking call in async context Move sync I/O to asyncio.to_thread() or loop.run_in_executor().

Tests hang indefinitely Ensure pytest-asyncio is installed and test functions are decorated with @pytest.mark.asyncio.

Advanced Async Patterns

Pattern 6: Async Context Managers

Implement __aenter__ and __aexit__ for async resource management.

import asyncio

class AsyncDatabaseConnection:
    def __init__(self, dsn: str):
        self.dsn = dsn
        self.connection = None

    async def __aenter__(self):
        print("Opening connection")
        await asyncio.sleep(0.1)  # Simulate connection
        self.connection = {"dsn": self.dsn, "connected": True}
        return self.connection

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        print("Closing connection")
        await asyncio.sleep(0.1)  # Simulate cleanup
        self.connection = None

async def query_database():
    async with AsyncDatabaseConnection("postgresql://localhost") as conn:
        print(f"Using connection: {conn}")
        await asyncio.sleep(0.2)  # Simulate query
        return {"rows": 10}

Pattern 7: Async Iterators

Use async for to iterate over asynchronous data sources.

from typing import AsyncIterator

async def fetch_pages(url: str, max_pages: int) -> AsyncIterator[dict]:
    for page in range(1, max_pages + 1):
        await asyncio.sleep(0.2)  # Simulate API call
        yield {
            "page": page,
            "url": f"{url}?page={page}",
            "data": [f"item_{page}_{i}" for i in range(5)]
        }

async def consume_pages():
    async for page_data in fetch_pages("https://api.example.com", 3):
        print(f"Page {page_data['page']}: {len(page_data['data'])} items")

Pattern 8: Producer-Consumer

Decouple producers and consumers using async queues.

from asyncio import Queue

async def producer(queue: Queue, producer_id: int, num_items: int):
    for i in range(num_items):
        item = f"Item-{producer_id}-{i}"
        await queue.put(item)
        print(f"Producer {producer_id}: {item}")
        await asyncio.sleep(0.1)
    await queue.put(None)  # Signal completion

async def consumer(queue: Queue, consumer_id: int):
    while True:
        item = await queue.get()
        if item is None:
            queue.task_done()
            break
        print(f"Consumer {consumer_id} processing: {item}")
        await asyncio.sleep(0.2)
        queue.task_done()

Usage Notes

• Context managers validate proper resource cleanup
• Async iterators enable memory-efficient streaming
• Queues decouple producers from consumers
• Always signal completion in producer-consumer patterns

Basic Async Patterns

Core Concepts

1. Event Loop

Single-threaded cooperative multitasking that schedules coroutines.

2. Coroutines

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

3. Tasks

Scheduled coroutines that run concurrently on the event loop.

Pattern 1: 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())

Pattern 2: Concurrent Execution with gather()

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]
    results = await asyncio.gather(*tasks)
    return results

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

Pattern 3: Task Creation and Management

async def background_task(name: str, delay: int):
    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 while tasks run
    print("Main: doing other work")
    await asyncio.sleep(0.5)

    # Wait for tasks
    result1, result2 = await asyncio.gather(task1, task2)
    print(f"Results: {result1}, {result2}")

Concurrency Control

Pattern 9: Semaphore for Rate Limiting

Semaphores limit the number of concurrent operations, essential for API rate limiting and resource management.

import asyncio

async def api_call(url: str, semaphore: asyncio.Semaphore) -> dict:
    async with semaphore:
        print(f"Calling {url}")
        await asyncio.sleep(0.5)
        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]
    results = await asyncio.gather(*tasks)
    return results

Pattern 10: Async Locks

Locks validate exclusive access to shared resources in async code.

class AsyncCounter:
    def __init__(self):
        self.value = 0
        self.lock = asyncio.Lock()

    async def increment(self):
        async with self.lock:
            current = self.value
            await asyncio.sleep(0.01)
            self.value = current + 1

    async def get_value(self) -> int:
        async with self.lock:
            return self.value

Cost Awareness

Python's GIL serializes CPU-bound threads, but contention costs still apply to asyncio locks:

• Uncontended lock: Near-zero cost (no kernel

involvement, just a Python attribute check)

• Contended lock: Coroutine suspension and

event-loop rescheduling (~microseconds)

• Semaphore with backpressure: Bounded queue depth

prevents memory growth under burst load

For CPU-bound parallelism, use multiprocessing or concurrent.futures.ProcessPoolExecutor to bypass the GIL entirely. With free-threaded Python (3.13t+), the native concurrency cost hierarchy applies: prefer per-thread state over shared atomics over contended locks.

Usage Notes

• Use semaphores for limiting concurrent operations

(API calls, connections)

• Use locks for protecting shared state
• Always use context managers (async with) for

proper cleanup

• Choose appropriate limits based on resource

constraints

Error Handling and Timeouts

Pattern 4: Error Handling

Handle errors gracefully in concurrent operations using try/except and gather's return_exceptions.

from typing import Optional
import asyncio

async def risky_operation(item_id: int) -> dict:
    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]:
    try:
        return await risky_operation(item_id)
    except ValueError as e:
        print(f"Error: {e}")
        return None

async def process_items(item_ids: list[int]):
    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 and not isinstance(r, Exception)]
    return successful

Pattern 5: Timeout Handling

Prevent operations from hanging indefinitely with timeouts.

async def slow_operation(delay: int) -> str:
    await asyncio.sleep(delay)
    return f"Completed after {delay}s"

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

Handling Cancellation

Properly handle task cancellation to clean up resources.

async def cancelable_task():
    try:
        while True:
            await asyncio.sleep(1)
            print("Working...")
    except asyncio.CancelledError:
        print("Cleaning up...")
        raise  # Re-raise to propagate cancellation

Best Practices

• Use return_exceptions=True in gather() to handle errors without stopping other tasks
• Always implement timeouts for external operations
• Handle CancelledError and re-raise it
• Use context managers for automatic cleanup

Common Pitfalls and Best Practices

Common Pitfalls

1. Forgetting await

# Wrong - returns coroutine object
result = async_function()

# Correct
result = await async_function()

2. Blocking the Event Loop

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

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

3. Not Handling Cancellation

async def cancelable_task():
    try:
        while True:
            await asyncio.sleep(1)
            print("Working...")
    except asyncio.CancelledError:
        print("Cleaning up...")
        raise  # Re-raise to propagate cancellation

4. Creating Event Loops Incorrectly

# Wrong - multiple event loops
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)

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

5. Mixing Sync and Async Code

# Wrong - calling sync blocking function
async def bad():
    requests.get(url)  # Blocking!

# Correct - use async library
async def good():
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()

Best Practices

1. Use asyncio.run() for entry point (Python 3.7+) 2. Always await coroutines to execute them 3. Use gather() for concurrent execution 4. Implement proper error handling with try/except 5. Use timeouts to prevent hanging operations 6. Pool connections for better performance 7. Avoid blocking operations in async code 8. Use semaphores for rate limiting 9. Handle task cancellation properly 10. Test with pytest-asyncio

Performance Tips

• Reuse connections and sessions
• Use connection pools for databases
• Implement backpressure with queues
• Monitor event loop blocking with debug mode
• Profile async code to identify bottlenecks

Debugging

# Enable asyncio debug mode
import asyncio
asyncio.run(main(), debug=True)

Real-World Applications

Web Scraping with aiohttp

Efficiently scrape multiple URLs concurrently.

import aiohttp
import asyncio

async def fetch_url(session: aiohttp.ClientSession, url: str) -> dict:
    try:
        async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as response:
            text = await response.text()
            return {"url": url, "status": response.status, "length": len(text)}
    except Exception as e:
        return {"url": url, "error": str(e)}

async def scrape_urls(urls: list[str]) -> list[dict]:
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, url) for url in urls]
        return await asyncio.gather(*tasks)

Async Database Operations

Fetch related data concurrently for improved performance.

async def get_user_data(db, user_id: int) -> dict:
    # Fetch related data concurrently
    user_task = db.fetch_one(f"SELECT * FROM users WHERE id = {user_id}")
    orders_task = db.execute(f"SELECT * FROM orders WHERE user_id = {user_id}")
    profile_task = db.fetch_one(f"SELECT * FROM profiles WHERE user_id = {user_id}")

    user, orders, profile = await asyncio.gather(user_task, orders_task, profile_task)

    return {"user": user, "orders": orders, "profile": profile}

Best Practices

• Reuse ClientSession for multiple requests
• Implement proper timeouts for HTTP requests
• Handle exceptions for individual operations
• Use connection pooling for database operations
• Fetch related data concurrently instead of sequentially
• Respect rate limits with semaphores

Testing Async Code

pytest-asyncio Setup

Install pytest-asyncio for testing async functions:

pip install pytest-asyncio

Basic Async Tests

import pytest
import asyncio

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

@pytest.mark.asyncio
async def test_concurrent_fetches():
    urls = ["url1", "url2", "url3"]
    tasks = [fetch_data(url) for url in urls]
    results = await asyncio.gather(*tasks)
    assert len(results) == 3

Testing Error Handling

@pytest.mark.asyncio
async def test_error_handling():
    with pytest.raises(ValueError):
        await risky_operation(3)

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

Testing with Mocks

from unittest.mock import AsyncMock

@pytest.mark.asyncio
async def test_with_mock():
    mock_db = AsyncMock()
    mock_db.fetch_one.return_value = {"id": 1, "name": "Test"}

    result = await get_user_data(mock_db, 1)
    assert result["user"]["id"] == 1
    mock_db.fetch_one.assert_called_once()

Best Practices

• Use @pytest.mark.asyncio for async test functions
• Test both success and error paths
• Mock external dependencies with AsyncMock
• Test timeout behavior
• Verify concurrent execution with multiple tasks