asyncio skill

---
name: asyncio
description: Python asyncio - Modern concurrent programming with async/await, event loops, tasks, coroutines, primitives, aiohttp, and FastAPI async patterns
version: 1.0.0
category: toolchain
author: Claude MPM Team
license: MIT
progressive_disclosure:
  entry_point:
    summary: "Async/await concurrency: coroutines, tasks, event loops, async HTTP (aiohttp), async DB, FastAPI async, locks, queues, error handling"
    when_to_use: "Concurrent I/O operations, HTTP clients/servers, database access, WebSocket handling, parallel API calls, background tasks, real-time systems"
    quick_start: "1. Define async def functions 2. Await async operations 3. Run with asyncio.run() 4. Use asyncio.gather() for parallel tasks 5. Handle errors with try/except"
context_limit: 700
tags:
  - asyncio
  - async
  - concurrency
  - coroutines
  - aiohttp
  - fastapi
  - async-database
  - event-loop
  - tasks
requires_tools: []
---

# Python asyncio - Async/Await Concurrency

## Overview

Python's asyncio library enables writing concurrent code using async/await syntax. It's ideal for I/O-bound operations like HTTP requests, database queries, file operations, and WebSocket connections. asyncio provides non-blocking execution without the complexity of threading or multiprocessing.

**Key Features**:
- async/await syntax for readable concurrent code
- Event loop for managing concurrent operations
- Tasks for running multiple coroutines concurrently
- Primitives: locks, semaphores, events, queues
- HTTP client/server with aiohttp
- Database async support (asyncpg, aiomysql, motor)
- FastAPI async endpoints
- WebSocket support
- Background task management

**Installation**:
```bash
# asyncio is built-in (Python 3.7+)

# Async HTTP client
pip install aiohttp

# Async HTTP requests (alternative)
pip install httpx

# Async database drivers
pip install asyncpg aiomysql motor  # PostgreSQL, MySQL, MongoDB

# FastAPI with async support
pip install fastapi uvicorn[standard]

# Async testing
pip install pytest-asyncio
```

## Basic Async/Await Patterns

### 1. Simple Async Function

```python
import asyncio

async def hello():
    """Basic async function (coroutine)."""
    print("Hello")
    await asyncio.sleep(1)  # Async sleep (non-blocking)
    print("World")
    return "Done"

# Run async function
result = asyncio.run(hello())
print(result)  # "Done"
```

**Key Points**:
- `async def` defines a coroutine function
- `await` suspends execution until awaitable completes
- `asyncio.run()` is the entry point for async programs
- Coroutines must be awaited or scheduled

### 2. Multiple Concurrent Tasks

```python
import asyncio
import time

async def task(name, duration):
    """Simulate async task."""
    print(f"{name}: Starting (duration: {duration}s)")
    await asyncio.sleep(duration)
    print(f"{name}: Complete")
    return f"{name} result"

async def run_concurrent():
    """Run multiple tasks concurrently."""
    start = time.time()

    # Sequential (slow) - 6 seconds total
    # result1 = await task("Task 1", 3)
    # result2 = await task("Task 2", 2)
    # result3 = await task("Task 3", 1)

    # Concurrent (fast) - 3 seconds total
    results = await asyncio.gather(
        task("Task 1", 3),
        task("Task 2", 2),
        task("Task 3", 1)
    )

    elapsed = time.time() - start
    print(f"Total time: {elapsed:.2f}s")
    print(f"Results: {results}")

asyncio.run(run_concurrent())
# Output: Total time: 3.00s (tasks ran concurrently)
```

### 3. Task Creation and Management

```python
import asyncio

async def background_task(name):
    """Long-running background task."""
    for i in range(5):
        print(f"{name}: iteration {i}")
        await asyncio.sleep(1)
    return f"{name} complete"

async def main():
    # Create task (starts immediately)
    task1 = asyncio.create_task(background_task("Task-1"))
    task2 = asyncio.create_task(background_task("Task-2"))

    # Do other work while tasks run
    print("Main: doing other work")
    await asyncio.sleep(2)

    # Wait for tasks to complete
    result1 = await task1
    result2 = await task2

    print(f"Results: {result1}, {result2}")

asyncio.run(main())
```

### 4. Error Handling in Async Code

```python
import asyncio

async def risky_operation(fail=False):
    """Operation that might fail."""
    await asyncio.sleep(1)
    if fail:
        raise ValueError("Operation failed")
    return "Success"

async def handle_errors():
    # Individual try/except
    try:
        result = await risky_operation(fail=True)
    except ValueError as e:
        print(f"Caught error: {e}")
        result = "Fallback value"

    # Gather with error handling
    results = await asyncio.gather(
        risky_operation(fail=False),
        risky_operation(fail=True),
        risky_operation(fail=False),
        return_exceptions=True  # Return exceptions instead of raising
    )

    for i, result in enumerate(results):
        if isinstance(result, Exception):
            print(f"Task {i} failed: {result}")
        else:
            print(f"Task {i} succeeded: {result}")

asyncio.run(handle_errors())
```

## Event Loop Fundamentals

### 1. Event Loop Lifecycle

```python
import asyncio

# Modern approach (Python 3.7+)
async def main():
    print("Main coroutine")
    await asyncio.sleep(1)

asyncio.run(main())  # Creates loop, runs main, closes loop

# Manual loop management (advanced use cases)
async def manual_example():
    loop = asyncio.get_event_loop()

    # Schedule coroutine
    task = loop.create_task(some_coroutine())

    # Schedule callback
    loop.call_later(5, callback_function)

    # Run until complete
    result = await task

    return result

# Get current event loop
async def get_current_loop():
    loop = asyncio.get_running_loop()
    print(f"Loop: {loop}")

    # Schedule callback in event loop
    loop.call_soon(lambda: print("Callback executed"))

    await asyncio.sleep(0)  # Let callback execute
```

### 2. Loop Scheduling and Callbacks

```python
import asyncio
from datetime import datetime

def callback(name, loop):
    """Callback function (not async)."""
    print(f"{datetime.now()}: {name} callback executed")

    # Stop loop after callback
    # loop.stop()

async def schedule_callbacks():
    loop = asyncio.get_running_loop()

    # Schedule immediate callback
    loop.call_soon(callback, "Immediate", loop)

    # Schedule callback after delay
    loop.call_later(2, callback, "Delayed 2s", loop)

    # Schedule callback at specific time
    loop.call_at(loop.time() + 3, callback, "Delayed 3s", loop)

    # Wait for callbacks to execute
    await asyncio.sleep(5)

asyncio.run(schedule_callbacks())
```

### 3. Running Blocking Code

```python
import asyncio
import time

def blocking_io():
    """CPU-intensive or blocking I/O operation."""
    print("Blocking operation started")
    time.sleep(2)  # Blocks thread
    print("Blocking operation complete")
    return "Blocking result"

async def run_in_executor():
    """Run blocking code in thread pool."""
    loop = asyncio.get_running_loop()

    # Run in default executor (thread pool)
    result = await loop.run_in_executor(
        None,  # Use default executor
        blocking_io
    )

    print(f"Result: {result}")

# Run blocking operations concurrently
async def concurrent_blocking():
    loop = asyncio.get_running_loop()

    # These run in thread pool, don't block event loop
    results = await asyncio.gather(
        loop.run_in_executor(None, blocking_io),
        loop.run_in_executor(None, blocking_io),
        loop.run_in_executor(None, blocking_io)
    )

    print(f"All results: {results}")

asyncio.run(concurrent_blocking())
```

## Asyncio Primitives

### 1. Locks for Mutual Exclusion

```python
import asyncio

# Shared resource
counter = 0
lock = asyncio.Lock()

async def increment_with_lock(name):
    """Increment counter with lock protection."""
    global counter

    async with lock:
        # Critical section - only one task at a time
        print(f"{name}: acquired lock")
        current = counter
        await asyncio.sleep(0.1)  # Simulate processing
        counter = current + 1
        print(f"{name}: released lock, counter={counter}")

async def increment_without_lock(name):
    """Increment without lock - race condition!"""
    global counter

    current = counter
    await asyncio.sleep(0.1)  # Race condition window
    counter = current + 1
    print(f"{name}: counter={counter}")

async def test_locks():
    global counter

    # Without lock (race condition)
    counter = 0
    await asyncio.gather(
        increment_without_lock("Task-1"),
        increment_without_lock("Task-2"),
        increment_without_lock("Task-3")
    )
    print(f"Without lock: {counter}")  # Often wrong (< 3)

    # With lock (correct)
    counter = 0
    await asyncio.gather(
        increment_with_lock("Task-1"),
        increment_with_lock("Task-2"),
        increment_with_lock("Task-3")
    )
    print(f"With lock: {counter}")  # Always 3

asyncio.run(test_locks())
```

### 2. Semaphores for Resource Limiting

```python
import asyncio

# Limit concurrent operations
semaphore = asyncio.Semaphore(2)  # Max 2 concurrent

async def limited_operation(name):
    """Operation limited by semaphore."""
    print(f"{name}: waiting for semaphore")

    async with semaphore:
        print(f"{name}: acquired semaphore")
        await asyncio.sleep(2)  # Simulate work
        print(f"{name}: releasing semaphore")

async def test_semaphore():
    # Create 5 tasks, but only 2 run concurrently
    await asyncio.gather(
        limited_operation("Task-1"),
        limited_operation("Task-2"),
        limited_operation("Task-3"),
        limited_operation("Task-4"),
        limited_operation("Task-5")
    )

asyncio.run(test_semaphore())
# Only 2 tasks hold semaphore at any time
```

### 3. Events for Signaling

```python
import asyncio

event = asyncio.Event()

async def waiter(name):
    """Wait for event to be set."""
    print(f"{name}: waiting for event")
    await event.wait()  # Block until event is set
    print(f"{name}: event received!")

async def setter():
    """Set event after delay."""
    await asyncio.sleep(2)
    print("Setter: setting event")
    event.set()  # Wake up all waiters

async def test_event():
    # Create waiters
    await asyncio.gather(
        waiter("Waiter-1"),
        waiter("Waiter-2"),
        waiter("Waiter-3"),
        setter()
    )

asyncio.run(test_event())
```

### 4. Queues for Task Distribution

```python
import asyncio
import random

async def producer(queue, name):
    """Produce items and add to queue."""
    for i in range(5):
        item = f"{name}-item-{i}"
        await queue.put(item)
        print(f"{name}: produced {item}")
        await asyncio.sleep(random.uniform(0.1, 0.5))

    # Signal completion
    await queue.put(None)

async def consumer(queue, name):
    """Consume items from queue."""
    while True:
        item = await queue.get()  # Block until item available

        if item is None:  # Shutdown signal
            queue.task_done()
            break

        print(f"{name}: consumed {item}")
        await asyncio.sleep(random.uniform(0.2, 0.8))
        queue.task_done()

async def test_queue():
    queue = asyncio.Queue(maxsize=10)

    # Create producers and consumers
    await asyncio.gather(
        producer(queue, "Producer-1"),
        producer(queue, "Producer-2"),
        consumer(queue, "Consumer-1"),
        consumer(queue, "Consumer-2"),
        consumer(queue, "Consumer-3")
    )

    # Wait for all items to be processed
    await queue.join()
    print("All tasks complete")

asyncio.run(test_queue())
```

### 5. Condition Variables

```python
import asyncio

condition = asyncio.Condition()
items = []

async def consumer(name):
    """Wait for items to be available."""
    async with condition:
        # Wait until items are available
        await condition.wait_for(lambda: len(items) > 0)

        item = items.pop(0)
        print(f"{name}: consumed {item}")

async def producer(name):
    """Add items and notify consumers."""
    async with condition:
        item = f"{name}-item"
        items.append(item)
        print(f"{name}: produced {item}")

        # Notify one waiting consumer
        condition.notify(n=1)
        # Or notify all: condition.notify_all()

async def test_condition():
    await asyncio.gather(
        consumer("Consumer-1"),
        consumer("Consumer-2"),
        producer("Producer-1"),
        producer("Producer-2")
    )

asyncio.run(test_condition())
```

## Async HTTP with aiohttp

### 1. Basic HTTP Client

```python
import asyncio
import aiohttp

async def fetch_url(session, url):
    """Fetch single URL."""
    async with session.get(url) as response:
        status = response.status
        text = await response.text()
        return {"url": url, "status": status, "length": len(text)}

async def fetch_multiple_urls():
    """Fetch multiple URLs concurrently."""
    urls = [
        "https://httpbin.org/delay/1",
        "https://httpbin.org/delay/2",
        "https://httpbin.org/json",
    ]

    async with aiohttp.ClientSession() as session:
        # Concurrent requests
        tasks = [fetch_url(session, url) for url in urls]
        results = await asyncio.gather(*tasks)

        for result in results:
            print(f"{result['url']}: {result['status']} ({result['length']} bytes)")

asyncio.run(fetch_multiple_urls())
```

### 2. HTTP Client with Error Handling

```python
import asyncio
import aiohttp
from typing import Dict, Any

async def fetch_with_retry(
    session: aiohttp.ClientSession,
    url: str,
    max_retries: int = 3
) -> Dict[str, Any]:
    """Fetch URL with retry logic."""
    for attempt in range(max_retries):
        try:
            async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as response:
                response.raise_for_status()  # Raise for 4xx/5xx
                data = await response.json()
                return {"success": True, "data": data}

        except aiohttp.ClientError as e:
            print(f"Attempt {attempt + 1} failed: {e}")
            if attempt == max_retries - 1:
                return {"success": False, "error": str(e)}

            # Exponential backoff
            await asyncio.sleep(2 ** attempt)

        except asyncio.TimeoutError:
            print(f"Attempt {attempt + 1} timed out")
            if attempt == max_retries - 1:
                return {"success": False, "error": "Timeout"}

            await asyncio.sleep(2 ** attempt)

async def parallel_api_calls():
    """Make parallel API calls with error handling."""
    urls = [
        "https://httpbin.org/json",
        "https://httpbin.org/status/500",  # Will fail
        "https://httpbin.org/delay/1",
    ]

    async with aiohttp.ClientSession() as session:
        results = await asyncio.gather(
            *[fetch_with_retry(session, url) for url in urls],
            return_exceptions=True  # Don't stop on errors
        )

        for url, result in zip(urls, results):
            if isinstance(result, Exception):
                print(f"{url}: Exception - {result}")
            elif result["success"]:
                print(f"{url}: Success")
            else:
                print(f"{url}: Failed - {result['error']}")

asyncio.run(parallel_api_calls())
```

### 3. HTTP Server with aiohttp

```python
from aiohttp import web
import asyncio

async def handle_hello(request):
    """Simple GET handler."""
    name = request.query.get("name", "World")
    return web.json_response({"message": f"Hello, {name}!"})

async def handle_post(request):
    """POST handler with JSON body."""
    data = await request.json()

    # Simulate async processing
    await asyncio.sleep(1)

    return web.json_response({
        "received": data,
        "status": "processed"
    })

async def handle_stream(request):
    """Streaming response."""
    response = web.StreamResponse()
    await response.prepare(request)

    for i in range(10):
        await response.write(f"Chunk {i}\n".encode())
        await asyncio.sleep(0.5)

    await response.write_eof()
    return response

# Create application
app = web.Application()
app.router.add_get("/hello", handle_hello)
app.router.add_post("/process", handle_post)
app.router.add_get("/stream", handle_stream)

# Run server
if __name__ == "__main__":
    web.run_app(app, host="0.0.0.0", port=8080)
```

### 4. WebSocket Client

```python
import asyncio
import aiohttp

async def websocket_client():
    """Connect to WebSocket server."""
    url = "wss://echo.websocket.org"

    async with aiohttp.ClientSession() as session:
        async with session.ws_connect(url) as ws:
            # Send messages
            await ws.send_str("Hello WebSocket")
            await ws.send_json({"type": "greeting", "data": "test"})

            # Receive messages
            async for msg in ws:
                if msg.type == aiohttp.WSMsgType.TEXT:
                    print(f"Received: {msg.data}")

                    if msg.data == "close":
                        await ws.close()
                        break

                elif msg.type == aiohttp.WSMsgType.ERROR:
                    print(f"Error: {ws.exception()}")
                    break

asyncio.run(websocket_client())
```

## Async Database Operations

### 1. PostgreSQL with asyncpg

```python
import asyncio
import asyncpg

async def database_operations():
    """Async PostgreSQL operations."""
    # Create connection pool
    pool = await asyncpg.create_pool(
        host="localhost",
        database="mydb",
        user="user",
        password="password",
        min_size=5,
        max_size=20
    )

    try:
        # Acquire connection from pool
        async with pool.acquire() as conn:
            # Execute query
            rows = await conn.fetch(
                "SELECT id, name, email FROM users WHERE active = $1",
                True
            )

            for row in rows:
                print(f"User: {row['name']} ({row['email']})")

            # Insert data
            await conn.execute(
                "INSERT INTO users (name, email) VALUES ($1, $2)",
                "Alice", "[email protected]"
            )

            # Transaction
            async with conn.transaction():
                await conn.execute("UPDATE users SET active = $1 WHERE id = $2", False, 1)
                await conn.execute("INSERT INTO audit_log (action) VALUES ($1)", "deactivate")

    finally:
        await pool.close()

asyncio.run(database_operations())
```

### 2. MongoDB with motor

```python
import asyncio
from motor.motor_asyncio import AsyncIOMotorClient

async def mongodb_operations():
    """Async MongoDB operations."""
    # Create client
    client = AsyncIOMotorClient("mongodb://localhost:27017")
    db = client.mydb
    collection = db.users

    try:
        # Insert document
        result = await collection.insert_one({
            "name": "Alice",
            "email": "[email protected]",
            "age": 30
        })
        print(f"Inserted ID: {result.inserted_id}")

        # Find documents
        cursor = collection.find({"age": {"$gte": 25}})
        async for document in cursor:
            print(f"User: {document['name']}")

        # Update document
        await collection.update_one(
            {"name": "Alice"},
            {"$set": {"age": 31}}
        )

        # Aggregation pipeline
        pipeline = [
            {"$match": {"age": {"$gte": 25}}},
            {"$group": {"_id": None, "avg_age": {"$avg": "$age"}}}
        ]
        async for result in collection.aggregate(pipeline):
            print(f"Average age: {result['avg_age']}")

    finally:
        client.close()

asyncio.run(mongodb_operations())
```

### 3. Connection Pool Pattern

```python
import asyncio
import asyncpg
from typing import Optional

class DatabasePool:
    """Async database connection pool manager."""

    def __init__(self, dsn: str):
        self.dsn = dsn
        self.pool: Optional[asyncpg.Pool] = None

    async def connect(self):
        """Create connection pool."""
        self.pool = await asyncpg.create_pool(self.dsn, min_size=5, max_size=20)

    async def close(self):
        """Close connection pool."""
        if self.pool:
            await self.pool.close()

    async def execute(self, query: str, *args):
        """Execute query."""
        async with self.pool.acquire() as conn:
            return await conn.execute(query, *args)

    async def fetch(self, query: str, *args):
        """Fetch multiple rows."""
        async with self.pool.acquire() as conn:
            return await conn.fetch(query, *args)

    async def fetchrow(self, query: str, *args):
        """Fetch single row."""
        async with self.pool.acquire() as conn:
            return await conn.fetchrow(query, *args)

# Usage
async def use_pool():
    db = DatabasePool("postgresql://user:pass@localhost/mydb")
    await db.connect()

    try:
        # Execute operations
        rows = await db.fetch("SELECT * FROM users")
        for row in rows:
            print(row)
    finally:
        await db.close()

asyncio.run(use_pool())
```

## FastAPI Async Patterns

### 1. Async Endpoints

```python
from fastapi import FastAPI, HTTPException
import asyncio
import httpx

app = FastAPI()

@app.get("/")
async def root():
    """Simple async endpoint."""
    return {"message": "Hello World"}

@app.get("/delay/{seconds}")
async def delayed_response(seconds: int):
    """Endpoint with async delay."""
    await asyncio.sleep(seconds)
    return {"message": f"Waited {seconds} seconds"}

@app.get("/fetch")
async def fetch_external():
    """Fetch data from external API."""
    async with httpx.AsyncClient() as client:
        response = await client.get("https://httpbin.org/json")
        return response.json()

@app.get("/parallel")
async def parallel_requests():
    """Make parallel API calls."""
    async with httpx.AsyncClient() as client:
        responses = await asyncio.gather(
            client.get("https://httpbin.org/delay/1"),
            client.get("https://httpbin.org/delay/2"),
            client.get("https://httpbin.org/json")
        )

        return {
            "results": [r.json() for r in responses]
        }
```

### 2. Background Tasks

```python
from fastapi import FastAPI, BackgroundTasks
import asyncio

app = FastAPI()

async def send_email(email: str, message: str):
    """Simulate sending email."""
    print(f"Sending email to {email}")
    await asyncio.sleep(5)  # Simulate slow email service
    print(f"Email sent to {email}: {message}")

@app.post("/send-notification")
async def send_notification(
    email: str,
    message: str,
    background_tasks: BackgroundTasks
):
    """Send notification in background."""
    # Add task to background
    background_tasks.add_task(send_email, email, message)

    # Return immediately
    return {"status": "notification queued"}

# Alternative: manual task creation
@app.post("/send-notification-manual")
async def send_notification_manual(email: str, message: str):
    """Create background task manually."""
    asyncio.create_task(send_email(email, message))
    return {"status": "notification queued"}
```

### 3. Async Dependencies

```python
from fastapi import FastAPI, Depends
import asyncpg

app = FastAPI()

# Database pool (global)
db_pool = None

async def get_db():
    """Dependency: database connection."""
    async with db_pool.acquire() as conn:
        yield conn

@app.on_event("startup")
async def startup():
    """Initialize database pool on startup."""
    global db_pool
    db_pool = await asyncpg.create_pool(
        "postgresql://user:pass@localhost/mydb"
    )

@app.on_event("shutdown")
async def shutdown():
    """Close database pool on shutdown."""
    await db_pool.close()

@app.get("/users/{user_id}")
async def get_user(user_id: int, conn=Depends(get_db)):
    """Get user with async database dependency."""
    user = await conn.fetchrow(
        "SELECT * FROM users WHERE id = $1",
        user_id
    )

    if not user:
        raise HTTPException(status_code=404, detail="User not found")

    return dict(user)
```

### 4. WebSocket Endpoints

```python
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from typing import List
import asyncio

app = FastAPI()

# Active connections
active_connections: List[WebSocket] = []

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    """WebSocket endpoint."""
    await websocket.accept()
    active_connections.append(websocket)

    try:
        while True:
            # Receive message
            data = await websocket.receive_text()

            # Broadcast to all connections
            for connection in active_connections:
                await connection.send_text(f"Broadcast: {data}")

    except WebSocketDisconnect:
        active_connections.remove(websocket)
        print("Client disconnected")

# Background task to send periodic updates
async def broadcast_updates():
    """Send periodic updates to all clients."""
    while True:
        await asyncio.sleep(5)

        for connection in active_connections:
            try:
                await connection.send_text("Periodic update")
            except:
                active_connections.remove(connection)

@app.on_event("startup")
async def startup():
    """Start background update task."""
    asyncio.create_task(broadcast_updates())
```

## Common Patterns and Best Practices

### 1. Timeout Handling

```python
import asyncio

async def slow_operation():
    """Slow operation."""
    await asyncio.sleep(10)
    return "Result"

async def with_timeout():
    """Run operation with timeout."""
    try:
        result = await asyncio.wait_for(slow_operation(), timeout=5.0)
        print(f"Result: {result}")

    except asyncio.TimeoutError:
        print("Operation timed out")

asyncio.run(with_timeout())
```

### 2. Cancellation Handling

```python
import asyncio

async def cancellable_task():
    """Task that can be cancelled."""
    try:
        for i in range(10):
            print(f"Working: {i}")
            await asyncio.sleep(1)

        return "Complete"

    except asyncio.CancelledError:
        print("Task was cancelled")
        # Cleanup
        raise  # Re-raise to propagate cancellation

async def cancel_example():
    """Example of task cancellation."""
    task = asyncio.create_task(cancellable_task())

    # Let it run for a bit
    await asyncio.sleep(3)

    # Cancel the task
    task.cancel()

    try:
        await task
    except asyncio.CancelledError:
        print("Confirmed: task was cancelled")

asyncio.run(cancel_example())
```

### 3. Resource Cleanup with Context Managers

```python
import asyncio

class AsyncResource:
    """Async context manager for resource management."""

    async def __aenter__(self):
        """Async setup."""
        print("Acquiring resource")
        await asyncio.sleep(1)  # Simulate async setup
        self.connection = "connected"
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        """Async cleanup."""
        print("Releasing resource")
        await asyncio.sleep(1)  # Simulate async cleanup
        self.connection = None

async def use_resource():
    """Use async resource."""
    async with AsyncResource() as resource:
        print(f"Using resource: {resource.connection}")
        await asyncio.sleep(1)
    # Resource automatically cleaned up

asyncio.run(use_resource())
```

### 4. Debouncing and Throttling

```python
import asyncio
from datetime import datetime

class Debouncer:
    """Debounce async function calls."""

    def __init__(self, delay: float):
        self.delay = delay
        self.task = None

    async def call(self, func, *args, **kwargs):
        """Debounced function call."""
        # Cancel previous task
        if self.task:
            self.task.cancel()

        # Create new task
        async def delayed_call():
            await asyncio.sleep(self.delay)
            await func(*args, **kwargs)

        self.task = asyncio.create_task(delayed_call())

async def api_call(query: str):
    """Simulated API call."""
    print(f"{datetime.now()}: API call with query: {query}")

async def debounce_example():
    """Example of debouncing."""
    debouncer = Debouncer(delay=1.0)

    # Rapid calls - only last one executes
    await debouncer.call(api_call, "query1")
    await asyncio.sleep(0.1)
    await debouncer.call(api_call, "query2")
    await asyncio.sleep(0.1)
    await debouncer.call(api_call, "query3")

    # Wait for debounced call
    await asyncio.sleep(2)

asyncio.run(debounce_example())
# Output: Only "query3" API call executes
```

### 5. Rate Limiting

```python
import asyncio
from datetime import datetime

class RateLimiter:
    """Limit rate of async operations."""

    def __init__(self, max_calls: int, period: float):
        self.max_calls = max_calls
        self.period = period
        self.semaphore = asyncio.Semaphore(max_calls)
        self.calls = []

    async def __aenter__(self):
        """Acquire rate limit slot."""
        await self.semaphore.acquire()

        now = asyncio.get_event_loop().time()

        # Remove old calls outside period
        self.calls = [t for t in self.calls if now - t < self.period]

        if len(self.calls) >= self.max_calls:
            # Wait until oldest call expires
            sleep_time = self.period - (now - self.calls[0])
            if sleep_time > 0:
                await asyncio.sleep(sleep_time)

        self.calls.append(now)
        return self

    async def __aexit__(self, *args):
        """Release semaphore."""
        self.semaphore.release()

async def rate_limited_operation(limiter, name):
    """Operation with rate limiting."""
    async with limiter:
        print(f"{datetime.now()}: {name}")
        await asyncio.sleep(0.1)

async def rate_limit_example():
    """Example of rate limiting."""
    # Max 3 calls per 2 seconds
    limiter = RateLimiter(max_calls=3, period=2.0)

    # Try to make 6 calls
    await asyncio.gather(*[
        rate_limited_operation(limiter, f"Call-{i}")
        for i in range(6)
    ])

asyncio.run(rate_limit_example())
```

## Debugging Async Code

### 1. Enable Debug Mode

```python
import asyncio
import logging

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

# Or set environment variable:
# PYTHONASYNCIODEBUG=1 python script.py

# Configure logging
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)

async def debug_example():
    logger.debug("Starting operation")
    await asyncio.sleep(1)
    logger.debug("Operation complete")
```

### 2. Detect Blocking Code

```python
import asyncio
import time

async def problematic_code():
    """Code with blocking operation."""
    print("Starting")

    # BAD: Blocking sleep
    time.sleep(2)  # This blocks the event loop!

    print("Complete")

# Run with debug mode to detect blocking
asyncio.run(problematic_code(), debug=True)
# Warning: Executing <Task> took 2.001 seconds
```

### 3. Track Pending Tasks

```python
import asyncio

async def track_tasks():
    """Track all pending tasks."""
    # Get all tasks
    tasks = asyncio.all_tasks()

    print(f"Total tasks: {len(tasks)}")
    for task in tasks:
        print(f"  - {task.get_name()}: {task}")

        # Check if task is done
        if task.done():
            try:
                result = task.result()
                print(f"    Result: {result}")
            except Exception as e:
                print(f"    Exception: {e}")

# Create some tasks
async def main():
    task1 = asyncio.create_task(asyncio.sleep(5), name="sleep-task")
    task2 = asyncio.create_task(track_tasks(), name="tracking")

    await task2
    task1.cancel()

asyncio.run(main())
```

## Testing Async Code

### 1. pytest-asyncio Setup

```python
# test_async.py
import pytest
import asyncio

# Mark test as async
@pytest.mark.asyncio
async def test_async_function():
    """Test async function."""
    result = await some_async_function()
    assert result == "expected"

@pytest.mark.asyncio
async def test_async_http():
    """Test async HTTP client."""
    async with aiohttp.ClientSession() as session:
        async with session.get("https://httpbin.org/json") as response:
            assert response.status == 200
            data = await response.json()
            assert "slideshow" in data

# Async fixture
@pytest.fixture
async def async_client():
    """Async test fixture."""
    client = await create_async_client()
    yield client
    await client.close()

@pytest.mark.asyncio
async def test_with_fixture(async_client):
    """Test using async fixture."""
    result = await async_client.fetch_data()
    assert result is not None
```

### 2. Mocking Async Functions

```python
import pytest
from unittest.mock import AsyncMock, patch

@pytest.mark.asyncio
async def test_with_mock():
    """Test with async mock."""
    # Create async mock
    mock_func = AsyncMock(return_value="mocked result")

    result = await mock_func()
    assert result == "mocked result"
    mock_func.assert_called_once()

@pytest.mark.asyncio
@patch("module.async_function", new_callable=AsyncMock)
async def test_with_patch(mock_async):
    """Test with patched async function."""
    mock_async.return_value = {"status": "success"}

    result = await some_function_that_calls_async()

    assert result["status"] == "success"
    mock_async.assert_called_once()
```

## Performance Optimization

### 1. Use asyncio.gather() for Parallelism

```python
import asyncio
import time

async def slow_task(n):
    await asyncio.sleep(1)
    return n * 2

async def optimized():
    """Parallel execution."""
    start = time.time()

    # Sequential (slow) - 5 seconds
    # results = []
    # for i in range(5):
    #     result = await slow_task(i)
    #     results.append(result)

    # Parallel (fast) - 1 second
    results = await asyncio.gather(*[slow_task(i) for i in range(5)])

    elapsed = time.time() - start
    print(f"Time: {elapsed:.2f}s, Results: {results}")

asyncio.run(optimized())
```

### 2. Connection Pooling

```python
import asyncio
import aiohttp

# BAD: Create new session for each request
async def bad_pattern():
    for i in range(10):
        async with aiohttp.ClientSession() as session:
            async with session.get("https://httpbin.org/json") as response:
                await response.json()

# GOOD: Reuse session with connection pool
async def good_pattern():
    async with aiohttp.ClientSession() as session:
        tasks = [
            session.get("https://httpbin.org/json")
            for i in range(10)
        ]
        responses = await asyncio.gather(*tasks)
        for response in responses:
            await response.json()
```

### 3. Avoid Blocking Operations

```python
import asyncio

# BAD: Blocking I/O in async function
async def bad_file_read():
    with open("large_file.txt") as f:  # Blocks event loop!
        data = f.read()
    return data

# GOOD: Use async file I/O or run in executor
async def good_file_read():
    loop = asyncio.get_running_loop()

    # Run blocking operation in thread pool
    data = await loop.run_in_executor(
        None,
        lambda: open("large_file.txt").read()
    )
    return data

# BETTER: Use aiofiles for async file I/O
import aiofiles

async def better_file_read():
    async with aiofiles.open("large_file.txt") as f:
        data = await f.read()
    return data
```

## Common Pitfalls

### ❌ Anti-Pattern 1: Not Awaiting Coroutines

```python
# WRONG
async def bad():
    result = async_function()  # Returns coroutine, doesn't execute!
    print(result)  # Prints: <coroutine object>

# CORRECT
async def good():
    result = await async_function()  # Actually executes
    print(result)
```

### ❌ Anti-Pattern 2: Blocking the Event Loop

```python
# WRONG
import time

async def bad():
    time.sleep(5)  # Blocks entire event loop!

# CORRECT
async def good():
    await asyncio.sleep(5)  # Non-blocking
```

### ❌ Anti-Pattern 3: Not Handling Cancellation

```python
# WRONG
async def bad():
    await asyncio.sleep(10)
    # No cleanup if cancelled

# CORRECT
async def good():
    try:
        await asyncio.sleep(10)
    except asyncio.CancelledError:
        # Cleanup resources
        await cleanup()
        raise  # Re-raise to propagate
```

### ❌ Anti-Pattern 4: Creating Event Loop Incorrectly

```python
# WRONG (Python 3.7+)
loop = asyncio.get_event_loop()
loop.run_until_complete(main())

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

### ❌ Anti-Pattern 5: Not Closing Resources

```python
# WRONG
async def bad():
    session = aiohttp.ClientSession()
    response = await session.get(url)
    # Session never closed - resource leak!

# CORRECT
async def good():
    async with aiohttp.ClientSession() as session:
        response = await session.get(url)
        # Session automatically closed
```

## Best Practices

1. **Use asyncio.run()** for entry point (Python 3.7+)
2. **Always await coroutines** - don't forget await
3. **Use async context managers** for resource cleanup
4. **Connection pooling** for HTTP and database clients
5. **Handle CancelledError** for graceful shutdown
6. **Use asyncio.gather()** for parallel operations
7. **Avoid blocking operations** in async functions
8. **Use timeouts** to prevent hanging operations
9. **Debug mode** during development to catch issues
10. **Test async code** with pytest-asyncio

## Quick Reference

### Common Commands

```bash
# Run async script
python script.py

# Run with debug mode
PYTHONASYNCIODEBUG=1 python script.py

# Run tests
pytest -v --asyncio-mode=auto

# Install async dependencies
pip install aiohttp asyncpg motor pytest-asyncio
```

### Essential Imports

```python
import asyncio
import aiohttp
import asyncpg
from typing import List, Dict, Any
```

## Resources

- **Official Documentation**: https://docs.python.org/3/library/asyncio.html
- **aiohttp**: https://docs.aiohttp.org/
- **asyncpg**: https://magicstack.github.io/asyncpg/
- **FastAPI Async**: https://fastapi.tiangolo.com/async/
- **pytest-asyncio**: https://pytest-asyncio.readthedocs.io/

## Related Skills

When using asyncio, consider these complementary skills:

- **fastapi-local-dev**: FastAPI async server patterns and production deployment
- **pytest**: Testing async code with pytest-asyncio and fixtures
- **systematic-debugging**: Debugging async race conditions and deadlocks

### Quick FastAPI Async Patterns (Inlined for Standalone Use)

```python
# FastAPI async endpoint pattern
from fastapi import FastAPI, Depends
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker

app = FastAPI()

# Async database setup
engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db")
AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

async def get_db():
    async with AsyncSessionLocal() as session:
        yield session

@app.get("/users/{user_id}")
async def get_user(user_id: int, db: AsyncSession = Depends(get_db)):
    # Async database query
    result = await db.execute(select(User).where(User.id == user_id))
    user = result.scalar_one_or_none()
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

# Background tasks with asyncio
@app.post("/send-email")
async def send_email_endpoint(email: EmailSchema):
    # Non-blocking background task
    asyncio.create_task(send_email_async(email))
    return {"status": "email queued"}
```

### Quick pytest-asyncio Patterns (Inlined for Standalone Use)

```python
# Testing async functions with pytest
import pytest
import pytest_asyncio
from httpx import AsyncClient

# Async test fixture
@pytest_asyncio.fixture
async def async_client():
    async with AsyncClient(app=app, base_url="http://test") as client:
        yield client

# Async test function
@pytest.mark.asyncio
async def test_get_user(async_client):
    response = await async_client.get("/users/1")
    assert response.status_code == 200
    assert response.json()["id"] == 1

# Testing concurrent operations
@pytest.mark.asyncio
async def test_concurrent_requests():
    async with AsyncClient(app=app, base_url="http://test") as client:
        # Run 10 requests concurrently
        responses = await asyncio.gather(
            *[client.get(f"/users/{i}") for i in range(1, 11)]
        )
        assert all(r.status_code == 200 for r in responses)

# Mock async dependencies
@pytest_asyncio.fixture
async def mock_db():
    # Setup mock database
    db = AsyncMock()
    yield db
    # Cleanup
```

### Quick Async Debugging Reference (Inlined for Standalone Use)

**Common Async Pitfalls:**

1. **Blocking the Event Loop**
   ```python
   # ❌ WRONG: Blocking call in async function
   async def bad_function():
       time.sleep(5)  # Blocks entire event loop!
       return "done"

   # ✅ CORRECT: Use asyncio.sleep
   async def good_function():
       await asyncio.sleep(5)  # Releases event loop
       return "done"
   ```

2. **Debugging Race Conditions**
   ```python
   # Add logging to track execution order
   import logging
   logging.basicConfig(level=logging.DEBUG)

   async def debug_task(name):
       logging.debug(f"{name}: Starting")
       await asyncio.sleep(1)
       logging.debug(f"{name}: Finished")
       return name

   # Run with detailed tracing
   asyncio.run(asyncio.gather(debug_task("A"), debug_task("B")), debug=True)
   ```

3. **Deadlock Detection**
   ```python
   # Use timeout to detect deadlocks
   try:
       result = await asyncio.wait_for(some_async_function(), timeout=5.0)
   except asyncio.TimeoutError:
       logging.error("Deadlock detected: operation timed out")
       # Investigate what's blocking
   ```

4. **Inspecting Running Tasks**
   ```python
   # Check all pending tasks
   tasks = asyncio.all_tasks()
   for task in tasks:
       print(f"Task: {task.get_name()}, Done: {task.done()}")
       if not task.done():
           print(f"  Current coro: {task.get_coro()}")
   ```

[Full FastAPI, pytest, and debugging patterns available in respective skills if deployed together]

---

**Python Version Compatibility:** This skill covers asyncio in Python 3.7+ and reflects current best practices for async programming in 2025.