Asyncio

by bobmatnyc718070a7d622MIT77 starsListed Oct 8, 2026Updated Oct 8, 2026Repository updated 2 months ago

Python asyncio - Modern concurrent programming with async/await, event loops, tasks, coroutines, primitives, aiohttp, and FastAPI async patterns

Instructions onlySoftware Development
AI-generated overview

Reference guide for Python asyncio concurrency, covering async/await, tasks, primitives, aiohttp, async databases and FastAPI.

What it does
This skill is an instructional reference on Python's asyncio library for concurrent programming. It explains async/await syntax, coroutines, tasks, event loop management, and synchronization primitives such as locks, semaphores, events, queues and conditions. It also documents async HTTP clients and servers with aiohttp, WebSocket handling, async database access with asyncpg and similar drivers, FastAPI async endpoints, and async testing with pytest-asyncio. It produces explanations and code examples rather than files or scripts.
When to use it
Use it when writing or reviewing concurrent Python code for I/O-bound work such as HTTP requests, database queries, WebSocket connections or background tasks. It is also useful when you need guidance on event loops, task coordination, or async patterns in FastAPI and aiohttp.
Requirements
No scripts are shipped; it is instructions only. Following the examples requires Python 3.7 or later, and optionally packages such as aiohttp, httpx, asyncpg, aiomysql, motor, fastapi, uvicorn and pytest-asyncio, plus network or database access for the HTTP and database examples.

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 clientpip install aiohttp
# Async HTTP requests (alternative)pip install httpx
# Async database driverspip install asyncpg aiomysql motor  # PostgreSQL, MySQL, MongoDB
# FastAPI with async supportpip install fastapi uvicorn[standard]
# Async testingpip 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 functionresult = 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 asyncioimport 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 loopasync 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 asynciofrom 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 asyncioimport 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 concurrentlyasync 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 resourcecounter = 0lock = 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 operationssemaphore = 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 asyncioimport 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 asyncioimport 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 asyncioimport aiohttpfrom 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 webimport 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 applicationapp = web.Application()app.router.add_get("/hello", handle_hello)app.router.add_post("/process", handle_post)app.router.add_get("/stream", handle_stream)
# Run serverif __name__ == "__main__":    web.run_app(app, host="0.0.0.0", port=8080)

4. WebSocket Client

python
import asyncioimport 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 asyncioimport 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 asynciofrom 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 asyncioimport asyncpgfrom 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)
# Usageasync 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, HTTPExceptionimport asyncioimport 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, BackgroundTasksimport 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, Dependsimport 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, WebSocketDisconnectfrom typing import Listimport asyncio
app = FastAPI()
# Active connectionsactive_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 updatesasync 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 asynciofrom 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 asynciofrom 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 asyncioimport logging
# Enable asyncio debug modeasyncio.run(main(), debug=True)
# Or set environment variable:# PYTHONASYNCIODEBUG=1 python script.py
# Configure logginglogging.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 asyncioimport 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 blockingasyncio.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 tasksasync 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.pyimport pytestimport asyncio
# Mark test as async@pytest.mark.asyncioasync def test_async_function():    """Test async function."""    result = await some_async_function()    assert result == "expected"
@pytest.mark.asyncioasync 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.fixtureasync def async_client():    """Async test fixture."""    client = await create_async_client()    yield client    await client.close()
@pytest.mark.asyncioasync 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 pytestfrom unittest.mock import AsyncMock, patch
@pytest.mark.asyncioasync 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 asyncioimport 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 asyncioimport aiohttp
# BAD: Create new session for each requestasync 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 poolasync 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 functionasync 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 executorasync 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/Oimport 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
# WRONGasync def bad():    result = async_function()  # Returns coroutine, doesn't execute!    print(result)  # Prints: <coroutine object>
# CORRECTasync def good():    result = await async_function()  # Actually executes    print(result)

❌ Anti-Pattern 2: Blocking the Event Loop

python
# WRONGimport time
async def bad():    time.sleep(5)  # Blocks entire event loop!
# CORRECTasync def good():    await asyncio.sleep(5)  # Non-blocking

❌ Anti-Pattern 3: Not Handling Cancellation

python
# WRONGasync def bad():    await asyncio.sleep(10)    # No cleanup if cancelled
# CORRECTasync 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
# WRONGasync def bad():    session = aiohttp.ClientSession()    response = await session.get(url)    # Session never closed - resource leak!
# CORRECTasync 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 scriptpython script.py
# Run with debug modePYTHONASYNCIODEBUG=1 python script.py
# Run testspytest -v --asyncio-mode=auto
# Install async dependenciespip install aiohttp asyncpg motor pytest-asyncio

Essential Imports

python
import asyncioimport aiohttpimport asyncpgfrom typing import List, Dict, Any

Resources

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 patternfrom fastapi import FastAPI, Dependsfrom sqlalchemy.ext.asyncio import AsyncSession, create_async_enginefrom sqlalchemy.orm import sessionmaker
app = FastAPI()
# Async database setupengine = 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 pytestimport pytestimport pytest_asynciofrom httpx import AsyncClient
# Async test fixture@pytest_asyncio.fixtureasync def async_client():    async with AsyncClient(app=app, base_url="http://test") as client:        yield client
# Async test function@pytest.mark.asyncioasync 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.asyncioasync 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.fixtureasync 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 functionasync def bad_function():    time.sleep(5)  # Blocks entire event loop!    return "done"
# ✅ CORRECT: Use asyncio.sleepasync def good_function():    await asyncio.sleep(5)  # Releases event loop    return "done"

  2. Debugging Race Conditions

    python
    # Add logging to track execution orderimport logginglogging.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 tracingasyncio.run(asyncio.gather(debug_task("A"), debug_task("B")), debug=True)

  3. Deadlock Detection

    python
    # Use timeout to detect deadlockstry:    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 taskstasks = 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.

Source and attribution

Source:bobmatnyc/claude-mpm-skillsintoolchains/python/async/asyncioat commit718070a

License: MIT

Content belongs to its original authors. SourceWeft indexes it from a public repository.

Report or request removal

More from bobmatnyc/claude-mpm-skills

Wordpress Security Validation

bobmatnyc

Teaches security-first WordPress development patterns for nonces, sanitization, validation and output escaping.

Security77updated 2 months ago

Bug Fix

bobmatnyc

Guides a structured bug-fix workflow: reproduce, find root cause, fix, verify, add regression tests, and document the PR.

Software Development77updated 2 months ago

Threat Modeling

bobmatnyc

Runs a lightweight STRIDE threat modeling workshop and turns identified risks into mitigations, tickets, and tests.

Security77updated 2 months ago

Webapp Testing

bobmatnyc

Reference guide of Playwright patterns for writing web application tests, covering selectors, waits, assertions and debugging.

Software Development77updated 2 months ago

Opentelemetry

bobmatnyc

Reference guidance for instrumenting services with OpenTelemetry and exporting OTLP telemetry through a Collector.

DevOps & Cloud77updated 2 months ago

Skill Creator

bobmatnyc

Guides the creation, structuring, and packaging of modular agent skills using progressive disclosure.

AI & Agents77updated 2 months ago

More in Software Development

Playground

anthropics

Featured

Builds self-contained interactive HTML playgrounds with controls, live preview, and copyable prompt output.

Software Development37Kupdated today

M5 Onboard

anthropics

Featured

Provisions M5Stack ESP32 boards by detecting them on USB, flashing UIFlow 2.0 firmware, and installing a MicroPython app bundle.

Software Development37Kupdated today

Cardputer Buddy

anthropics

Featured

Guides iterating on a Cardputer-Adv MicroPython app bundle: adding apps, pushing files over USB-serial, tailing logs, and running REPL commands.

Software Development37Kupdated today

Web Design Guidelines

vercel-labs

Featured

Reviews UI code against Web Interface Guidelines fetched from a remote source and reports findings as file:line entries.

Software Development32Kupdated 5 weeks ago

Vercel React Native Skills

vercel-labs

Featured

React Native and Expo best-practice rules covering list performance, animation, navigation, UI patterns, state, and monorepo setup.

Software Development32Kupdated 5 weeks ago

Vercel React Best Practices

vercel-labs

Featured

Vercel's React and Next.js performance guidelines, organized as 70 rules for writing, reviewing, and refactoring code.

Software Development32Kupdated 5 weeks ago