async-python-patterns

What this skill does

# 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

## Quick Start

```python
import asyncio

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

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

### Basic Async/Await

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

### Concurrent Execution

```python
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
```

## 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.

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

### 3. Tasks

Scheduled coroutines that run concurrently on the event loop.

```python
task = asyncio.create_task(background_task())
result = await task
```

### 4. Async Context Managers

Resources that support `async with` for proper cleanup.

```python
async with AsyncDatabaseConnection("postgresql://localhost") as conn:
    result = await conn.query("SELECT * FROM users")
```

### 5. Async Iterators

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

```python
async for item in async_range(1, 10):
    process(item)
```

**See detailed explanations**: [Core Concepts](references/core-concepts.md)

## Fundamental Patterns

### Pattern 1: Task Creation and Management

Create and manage concurrent tasks effectively.

```python
task1 = asyncio.create_task(background_task("Task 1", 2))
task2 = asyncio.create_task(background_task("Task 2", 1))

result1 = await task1
result2 = await task2
```

**See detailed patterns**: [Basic Patterns](references/basic-patterns.md)

### Pattern 2: Error Handling

Handle errors in async code with proper exception handling.

```python
async def safe_operation(item_id: int):
    try:
        return await risky_operation(item_id)
    except ValueError as e:
        print(f"Error: {e}")
        return None

results = await asyncio.gather(*tasks, return_exceptions=True)
```

**See detailed patterns**: [Error Handling](references/error-handling.md)

### Pattern 3: Timeout Handling

Execute operations with timeouts to prevent hanging.

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

**See detailed patterns**: [Timeouts and Cancellation](references/timeouts-cancellation.md)

## Advanced Patterns

### Async Context Managers

Implement async context managers for resource management.

```python
class AsyncDatabaseConnection:
    async def __aenter__(self):
        # Setup logic
        return connection

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        # Cleanup logic
        pass
```

**See detailed implementations**: [Advanced Patterns](references/advanced-patterns.md)

### Producer-Consumer Pattern

Implement async queues for producer-consumer workflows.

```python
queue = asyncio.Queue(maxsize=10)

async def producer(queue):
    for item in items:
        await queue.put(item)

async def consumer(queue):
    while True:
        item = await queue.get()
        process(item)
        queue.task_done()
```

**See detailed patterns**: [Concurrency Patterns](references/concurrency-patterns.md)

### Rate Limiting with Semaphore

Control concurrent access with semaphores.

```python
semaphore = asyncio.Semaphore(5)

async def api_call(url, semaphore):
    async with semaphore:
        return await fetch(url)
```

**See detailed implementations**: [Synchronization](references/synchronization.md)

## Real-World Applications

### Web Scraping with aiohttp

```python
async def scrape_urls(urls: List[str]):
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        return results
```

**See detailed examples**: [Real-World Applications](references/real-world-apps.md)

### Async Database Operations

```python
async def get_user_data(db, user_id):
    user, orders, profile = await asyncio.gather(
        db.fetch_user(user_id),
        db.fetch_orders(user_id),
        db.fetch_profile(user_id)
    )
    return {"user": user, "orders": orders, "profile": profile}
```

**See detailed patterns**: [Real-World Applications](references/real-world-apps.md)

## Performance Best Practices

1. **Use Connection Pools**: Reuse connections efficiently
2. **Batch Operations**: Process items in batches
3. **Avoid Blocking**: Run CPU-bound work in executors
4. **Use Semaphores**: Limit concurrent operations
5. **Handle Cancellation**: Clean up resources properly

**See detailed guide**: [Performance Best Practices](references/performance.md)

## Common Pitfalls

1. **Forgetting await**: Returns coroutine object, doesn't execute
2. **Blocking the Event Loop**: Using `time.sleep()` instead of `asyncio.sleep()`
3. **Not Handling Cancellation**: Tasks cancelled without cleanup
4. **Mixing Sync and Async**: Calling async from sync incorrectly

**See detailed solutions**: [Common Pitfalls](references/common-pitfalls.md)

## Testing Async Code

```python
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)
```

**See detailed testing guide**: [Testing](references/testing.md)

## Best Practices Summary

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
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 async code with pytest-asyncio

## 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