Python Asyncio

Quick Reference

When to Use This Skill

Use for async/concurrent programming:

• Network requests (HTTP, WebSockets)
• Database queries
• File I/O operations
• Multiple concurrent I/O operations
• FastAPI/Starlette async endpoints

Related skills:

• For FastAPI: see python-fastapi
• For type hints: see python-type-hints
• For gotchas: see python-gotchas

---

Python Asyncio Complete Guide

Overview

Asyncio is Python's built-in framework for writing concurrent code using async/await syntax. It's ideal for I/O-bound operations like network requests, file I/O, and database queries.

When to Use Asyncio

Good Use Cases

• Network requests (HTTP clients, WebSockets)
• Database queries
• File I/O operations
• Web servers (FastAPI, Starlette)
• Message queues
• Multiple concurrent I/O operations

When NOT to Use

• CPU-bound tasks (use multiprocessing instead)
• Simple sequential scripts
• Legacy codebases without async support

Core Concepts

Basic Async/Await

import asyncio

# Async function (coroutine)
async def fetch_data(url: str) -> dict:
    # Simulated async I/O
    await asyncio.sleep(1)
    return {"url": url, "data": "..."}

# Running a coroutine
async def main():
    result = await fetch_data("https://api.example.com")
    print(result)

# Entry point
asyncio.run(main())

Concurrent Execution with gather()

import asyncio
import aiohttp

async def fetch_url(session: aiohttp.ClientSession, url: str) -> dict:
    async with session.get(url) as response:
        return {"url": url, "status": response.status}

async def fetch_all(urls: list[str]) -> list[dict]:
    async with aiohttp.ClientSession() as session:
        # Run all requests concurrently
        tasks = [fetch_url(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        return results

# Usage
urls = [
    "https://api.example.com/1",
    "https://api.example.com/2",
    "https://api.example.com/3",
]
results = asyncio.run(fetch_all(urls))

TaskGroup (Python 3.11+)

import asyncio

async def process_item(item: str) -> str:
    await asyncio.sleep(1)
    return f"Processed: {item}"

async def process_all(items: list[str]) -> list[str]:
    results = []

    # TaskGroup provides structured concurrency
    async with asyncio.TaskGroup() as tg:
        tasks = [tg.create_task(process_item(item)) for item in items]

    # All tasks complete when exiting the context
    return [task.result() for task in tasks]

# Exception handling with TaskGroup
async def process_with_errors(items: list[str]):
    try:
        async with asyncio.TaskGroup() as tg:
            for item in items:
                tg.create_task(process_item(item))
    except* ValueError as eg:
        # Handle ValueError exceptions
        for exc in eg.exceptions:
            print(f"ValueError: {exc}")
    except* TypeError as eg:
        # Handle TypeError exceptions
        for exc in eg.exceptions:
            print(f"TypeError: {exc}")

create_task() vs await

import asyncio

async def task_a():
    await asyncio.sleep(2)
    return "A done"

async def task_b():
    await asyncio.sleep(1)
    return "B done"

# Sequential (slower - 3 seconds total)
async def sequential():
    result_a = await task_a()  # Wait 2 seconds
    result_b = await task_b()  # Wait 1 second
    return result_a, result_b

# Concurrent (faster - 2 seconds total)
async def concurrent():
    task1 = asyncio.create_task(task_a())  # Start immediately
    task2 = asyncio.create_task(task_b())  # Start immediately
    result_a = await task1
    result_b = await task2
    return result_a, result_b

# Using gather (recommended for multiple tasks)
async def concurrent_gather():
    result_a, result_b = await asyncio.gather(task_a(), task_b())
    return result_a, result_b

Advanced Patterns

Semaphores for Rate Limiting

import asyncio
import aiohttp

async def fetch_with_limit(
    session: aiohttp.ClientSession,
    url: str,
    semaphore: asyncio.Semaphore
) -> dict:
    async with semaphore:  # Limits concurrent requests
        async with session.get(url) as response:
            return await response.json()

async def fetch_many(urls: list[str], max_concurrent: int = 10) -> list[dict]:
    semaphore = asyncio.Semaphore(max_concurrent)

    async with aiohttp.ClientSession() as session:
        tasks = [fetch_with_limit(session, url, semaphore) for url in urls]
        return await asyncio.gather(*tasks)

Timeouts

import asyncio

async def slow_operation():
    await asyncio.sleep(10)
    return "done"

async def with_timeout():
    try:
        # Wait at most 5 seconds
        result = await asyncio.wait_for(slow_operation(), timeout=5.0)
        return result
    except asyncio.TimeoutError:
        print("Operation timed out")
        return None

# Using timeout context manager (Python 3.11+)
async def with_timeout_context():
    async with asyncio.timeout(5.0):
        result = await slow_operation()
        return result

Queues for Producer-Consumer

import asyncio
from typing import Any

async def producer(queue: asyncio.Queue, items: list[Any]):
    for item in items:
        await queue.put(item)
        print(f"Produced: {item}")
    # Signal completion
    await queue.put(None)

async def consumer(queue: asyncio.Queue, consumer_id: int):
    while True:
        item = await queue.get()
        if item is None:
            # Put sentinel back for other consumers
            await queue.put(None)
            break
        print(f"Consumer {consumer_id} processing: {item}")
        await asyncio.sleep(0.5)  # Simulate work
        queue.task_done()

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

    items = list(range(20))

    # Start producer and consumers
    async with asyncio.TaskGroup() as tg:
        tg.create_task(producer(queue, items))
        for i in range(3):
            tg.create_task(consumer(queue, i))

asyncio.run(main())

Async Generators

import asyncio
from typing import AsyncIterator

async def fetch_pages(url: str, max_pages: int = 10) -> AsyncIterator[dict]:
    """Async generator for paginated API."""
    page = 1
    while page <= max_pages:
        # Simulate API call
        await asyncio.sleep(0.1)
        data = {"page": page, "items": [f"item_{i}" for i in range(10)]}

        if not data["items"]:
            break

        yield data
        page += 1

async def process_pages():
    async for page_data in fetch_pages("https://api.example.com"):
        print(f"Processing page {page_data['page']}")
        for item in page_data["items"]:
            process(item)

Async Context Managers

from contextlib import asynccontextmanager
from typing import AsyncIterator

class AsyncDatabaseConnection:
    async def __aenter__(self):
        await self.connect()
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await self.disconnect()
        return False

    async def connect(self):
        print("Connecting...")
        await asyncio.sleep(0.1)

    async def disconnect(self):
        print("Disconnecting...")
        await asyncio.sleep(0.1)

# Using decorator
@asynccontextmanager
async def async_session() -> AsyncIterator[dict]:
    session = {"connected": True}
    try:
        yield session
    finally:
        session["connected"] = False
        await asyncio.sleep(0.1)  # Cleanup

# Usage
async def main():
    async with AsyncDatabaseConnection() as db:
        await db.query("SELECT * FROM users")

    async with async_session() as session:
        print(session)

Performance Optimization

Eager Task Factory (Python 3.12+)

import asyncio

async def cached_operation(key: str) -> str:
    cache = {"a": "value_a", "b": "value_b"}
    if key in cache:
        return cache[key]  # Returns synchronously
    await asyncio.sleep(1)  # Only if cache miss
    return f"fetched_{key}"

async def main():
    loop = asyncio.get_event_loop()
    # Enable eager task execution
    loop.set_task_factory(asyncio.eager_task_factory)

    # Cached operations complete synchronously without event loop overhead
    result = await cached_operation("a")

Using uvloop

# Install: pip install uvloop

import asyncio

try:
    import uvloop
    # 2-4x performance improvement
    asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
except ImportError:
    pass  # Fall back to default event loop

async def main():
    # Your async code here
    pass

asyncio.run(main())

Free-Threaded asyncio (Python 3.14+)

# Python 3.14 improvements for free-threaded builds:
# - Thread-safe asyncio with lock-free data structures
# - Linear scaling with number of threads
# - 10-20% single-threaded performance improvement
# - Reduced memory usage

import asyncio
import threading

async def per_thread_loop():
    """Each thread can run its own event loop."""
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    try:
        await asyncio.sleep(1)
    finally:
        loop.close()

# Multiple event loops in parallel (free-threaded build)
threads = [
    threading.Thread(target=lambda: asyncio.run(per_thread_loop()))
    for _ in range(4)
]
for t in threads:
    t.start()
for t in threads:
    t.join()

Common Gotchas

Blocking the Event Loop

import asyncio
import time

# BAD: Blocks the event loop
async def bad_example():
    time.sleep(5)  # Blocks everything!
    return "done"

# GOOD: Use async sleep
async def good_example():
    await asyncio.sleep(5)
    return "done"

# GOOD: Run blocking code in executor
async def blocking_in_executor():
    loop = asyncio.get_event_loop()
    result = await loop.run_in_executor(None, time.sleep, 5)
    return result

# For CPU-bound work, use ProcessPoolExecutor
from concurrent.futures import ProcessPoolExecutor

async def cpu_bound_work(data: list) -> list:
    loop = asyncio.get_event_loop()
    with ProcessPoolExecutor() as pool:
        result = await loop.run_in_executor(pool, heavy_computation, data)
    return result

Creating Tasks in Wrong Context

import asyncio

# BAD: Task created outside async context
# task = asyncio.create_task(some_coroutine())  # RuntimeError!

# GOOD: Create tasks inside async function
async def main():
    task = asyncio.create_task(some_coroutine())
    await task

asyncio.run(main())

Forgetting to Await

import asyncio

async def fetch():
    await asyncio.sleep(1)
    return "data"

# BAD: Coroutine never executed
async def bad():
    result = fetch()  # Just creates coroutine object!
    print(result)     # Prints coroutine object, not "data"

# GOOD: Always await coroutines
async def good():
    result = await fetch()
    print(result)  # Prints "data"

Exception Handling in Tasks

import asyncio

async def failing_task():
    await asyncio.sleep(1)
    raise ValueError("Task failed!")

async def main():
    # BAD: Exception silently lost
    task = asyncio.create_task(failing_task())
    await asyncio.sleep(2)  # Task exception ignored

    # GOOD: Always await tasks or use TaskGroup
    task = asyncio.create_task(failing_task())
    try:
        await task
    except ValueError as e:
        print(f"Caught: {e}")

    # BEST: Use TaskGroup (Python 3.11+)
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(failing_task())
    except* ValueError as eg:
        for exc in eg.exceptions:
            print(f"Caught: {exc}")

Async Libraries

HTTP Clients

# aiohttp - Most popular
import aiohttp

async def fetch_aiohttp(url: str) -> dict:
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.json()

# httpx - Supports both sync and async
import httpx

async def fetch_httpx(url: str) -> dict:
    async with httpx.AsyncClient() as client:
        response = await client.get(url)
        return response.json()

Database

# asyncpg - PostgreSQL
import asyncpg

async def query_postgres():
    conn = await asyncpg.connect("postgresql://user:pass@localhost/db")
    rows = await conn.fetch("SELECT * FROM users")
    await conn.close()
    return rows

# aiosqlite - SQLite
import aiosqlite

async def query_sqlite():
    async with aiosqlite.connect("database.db") as db:
        async with db.execute("SELECT * FROM users") as cursor:
            return await cursor.fetchall()

Web Frameworks

# FastAPI (built on Starlette)
from fastapi import FastAPI

app = FastAPI()

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    return {"item_id": item_id}

# Starlette
from starlette.applications import Starlette
from starlette.responses import JSONResponse
from starlette.routing import Route

async def homepage(request):
    return JSONResponse({"hello": "world"})

app = Starlette(routes=[Route("/", homepage)])

Additional References

For production-ready patterns beyond this guide, see:

• [Async Patterns Library](references/async-patterns-library.md) - Token bucket rate limiter, retry with exponential backoff, connection pools, batch processors, event bus, transaction context managers, async cache with TTL, graceful shutdown handlers

Asyncio Patterns Library

Production-ready async patterns for Python applications.

Rate Limiter with Token Bucket

import asyncio
from dataclasses import dataclass
from time import monotonic

@dataclass
class TokenBucket:
    """Token bucket rate limiter."""
    rate: float  # tokens per second
    capacity: float
    tokens: float = 0.0
    last_update: float = 0.0

    def __post_init__(self):
        self.tokens = self.capacity
        self.last_update = monotonic()
        self._lock = asyncio.Lock()

    async def acquire(self, tokens: float = 1.0) -> None:
        """Acquire tokens, waiting if necessary."""
        async with self._lock:
            while True:
                now = monotonic()
                elapsed = now - self.last_update
                self.tokens = min(self.capacity, self.tokens + elapsed * self.rate)
                self.last_update = now

                if self.tokens >= tokens:
                    self.tokens -= tokens
                    return

                wait_time = (tokens - self.tokens) / self.rate
                await asyncio.sleep(wait_time)


# Usage
limiter = TokenBucket(rate=10, capacity=100)  # 10 req/s, burst of 100

async def rate_limited_request(url: str):
    await limiter.acquire()
    async with httpx.AsyncClient() as client:
        return await client.get(url)

Retry with Exponential Backoff

import asyncio
from functools import wraps
from typing import TypeVar, Callable, Awaitable
import random

T = TypeVar("T")

def retry_async(
    max_retries: int = 3,
    base_delay: float = 1.0,
    max_delay: float = 60.0,
    exponential_base: float = 2.0,
    jitter: bool = True,
    exceptions: tuple[type[Exception], ...] = (Exception,),
):
    """Retry decorator with exponential backoff."""
    def decorator(func: Callable[..., Awaitable[T]]) -> Callable[..., Awaitable[T]]:
        @wraps(func)
        async def wrapper(*args, **kwargs) -> T:
            last_exception = None
            for attempt in range(max_retries + 1):
                try:
                    return await func(*args, **kwargs)
                except exceptions as e:
                    last_exception = e
                    if attempt == max_retries:
                        raise

                    delay = min(base_delay * (exponential_base ** attempt), max_delay)
                    if jitter:
                        delay *= (0.5 + random.random())

                    await asyncio.sleep(delay)

            raise last_exception  # Should never reach here

        return wrapper
    return decorator


# Usage
@retry_async(max_retries=3, exceptions=(httpx.HTTPError, asyncio.TimeoutError))
async def fetch_with_retry(url: str) -> dict:
    async with httpx.AsyncClient() as client:
        response = await client.get(url, timeout=10.0)
        response.raise_for_status()
        return response.json()

Connection Pool

import asyncio
from contextlib import asynccontextmanager
from typing import AsyncIterator, TypeVar, Generic

T = TypeVar("T")

class AsyncPool(Generic[T]):
    """Generic async connection pool."""

    def __init__(
        self,
        factory: Callable[[], Awaitable[T]],
        max_size: int = 10,
        min_size: int = 1,
    ):
        self._factory = factory
        self._max_size = max_size
        self._min_size = min_size
        self._pool: asyncio.Queue[T] = asyncio.Queue(maxsize=max_size)
        self._size = 0
        self._lock = asyncio.Lock()

    async def _create_connection(self) -> T:
        conn = await self._factory()
        self._size += 1
        return conn

    @asynccontextmanager
    async def acquire(self) -> AsyncIterator[T]:
        """Acquire a connection from the pool."""
        conn = None

        # Try to get from pool
        try:
            conn = self._pool.get_nowait()
        except asyncio.QueueEmpty:
            # Create new if under limit
            async with self._lock:
                if self._size < self._max_size:
                    conn = await self._create_connection()

        # Wait for available connection
        if conn is None:
            conn = await self._pool.get()

        try:
            yield conn
        finally:
            # Return to pool
            await self._pool.put(conn)


# Usage
async def create_db_connection():
    return await asyncpg.connect("postgresql://...")

pool = AsyncPool(create_db_connection, max_size=20)

async def query_database():
    async with pool.acquire() as conn:
        return await conn.fetch("SELECT * FROM users")

Batch Processor

import asyncio
from typing import TypeVar, Callable, Awaitable
from dataclasses import dataclass, field

T = TypeVar("T")
R = TypeVar("R")

@dataclass
class BatchProcessor(Generic[T, R]):
    """Process items in batches with concurrency control."""
    processor: Callable[[list[T]], Awaitable[list[R]]]
    batch_size: int = 100
    max_concurrent: int = 5
    timeout: float | None = None

    async def process(self, items: list[T]) -> list[R]:
        """Process all items in batches."""
        semaphore = asyncio.Semaphore(self.max_concurrent)
        results: list[R] = []

        async def process_batch(batch: list[T]) -> list[R]:
            async with semaphore:
                return await self.processor(batch)

        # Create batches
        batches = [
            items[i:i + self.batch_size]
            for i in range(0, len(items), self.batch_size)
        ]

        # Process all batches
        if self.timeout:
            async with asyncio.timeout(self.timeout):
                batch_results = await asyncio.gather(
                    *[process_batch(batch) for batch in batches]
                )
        else:
            batch_results = await asyncio.gather(
                *[process_batch(batch) for batch in batches]
            )

        # Flatten results
        for batch_result in batch_results:
            results.extend(batch_result)

        return results


# Usage
async def process_urls(urls: list[str]) -> list[dict]:
    async with httpx.AsyncClient() as client:
        responses = await asyncio.gather(*[client.get(url) for url in urls])
        return [r.json() for r in responses]

processor = BatchProcessor(process_urls, batch_size=10, max_concurrent=3)
results = await processor.process(all_urls)

Event Bus

import asyncio
from collections import defaultdict
from typing import Callable, Awaitable, Any
from dataclasses import dataclass, field

@dataclass
class AsyncEventBus:
    """Async pub/sub event bus."""
    _handlers: dict[str, list[Callable[..., Awaitable[None]]]] = field(
        default_factory=lambda: defaultdict(list)
    )

    def subscribe(self, event: str, handler: Callable[..., Awaitable[None]]) -> None:
        """Subscribe to an event."""
        self._handlers[event].append(handler)

    def unsubscribe(self, event: str, handler: Callable[..., Awaitable[None]]) -> None:
        """Unsubscribe from an event."""
        self._handlers[event].remove(handler)

    async def publish(self, event: str, **data: Any) -> None:
        """Publish an event to all subscribers."""
        handlers = self._handlers.get(event, [])
        if handlers:
            await asyncio.gather(*[h(**data) for h in handlers])

    async def publish_background(self, event: str, **data: Any) -> None:
        """Publish event without waiting for handlers."""
        handlers = self._handlers.get(event, [])
        for handler in handlers:
            asyncio.create_task(handler(**data))


# Usage
bus = AsyncEventBus()

async def on_user_created(user_id: int, email: str):
    await send_welcome_email(email)

async def on_user_created_analytics(user_id: int, email: str):
    await track_signup(user_id)

bus.subscribe("user.created", on_user_created)
bus.subscribe("user.created", on_user_created_analytics)

await bus.publish("user.created", user_id=123, email="user@example.com")

Async Context Manager for Transactions

from contextlib import asynccontextmanager
from typing import AsyncIterator
from sqlalchemy.ext.asyncio import AsyncSession

@asynccontextmanager
async def transaction(session: AsyncSession) -> AsyncIterator[AsyncSession]:
    """Transaction context manager with automatic rollback on error."""
    try:
        yield session
        await session.commit()
    except Exception:
        await session.rollback()
        raise


# Usage
async def create_user_with_profile(session: AsyncSession, user_data: dict):
    async with transaction(session):
        user = User(**user_data)
        session.add(user)
        await session.flush()  # Get user.id

        profile = Profile(user_id=user.id, **profile_data)
        session.add(profile)
        # Commit happens automatically if no exception

Async Cache with TTL

import asyncio
from dataclasses import dataclass, field
from time import monotonic
from typing import TypeVar, Generic, Callable, Awaitable

T = TypeVar("T")

@dataclass
class CacheEntry(Generic[T]):
    value: T
    expires_at: float

@dataclass
class AsyncCache(Generic[T]):
    """Async cache with TTL support."""
    default_ttl: float = 300.0  # 5 minutes
    _cache: dict[str, CacheEntry[T]] = field(default_factory=dict)
    _lock: asyncio.Lock = field(default_factory=asyncio.Lock)

    async def get(self, key: str) -> T | None:
        """Get value from cache."""
        async with self._lock:
            entry = self._cache.get(key)
            if entry is None:
                return None
            if monotonic() > entry.expires_at:
                del self._cache[key]
                return None
            return entry.value

    async def set(self, key: str, value: T, ttl: float | None = None) -> None:
        """Set value in cache."""
        async with self._lock:
            expires_at = monotonic() + (ttl or self.default_ttl)
            self._cache[key] = CacheEntry(value=value, expires_at=expires_at)

    async def get_or_set(
        self,
        key: str,
        factory: Callable[[], Awaitable[T]],
        ttl: float | None = None,
    ) -> T:
        """Get from cache or compute and store."""
        value = await self.get(key)
        if value is not None:
            return value

        value = await factory()
        await self.set(key, value, ttl)
        return value


# Usage
cache = AsyncCache[dict](default_ttl=60)

async def get_user(user_id: int) -> dict:
    return await cache.get_or_set(
        f"user:{user_id}",
        lambda: fetch_user_from_db(user_id),
        ttl=300,
    )

Graceful Shutdown Handler

import asyncio
import signal
from typing import Callable, Awaitable

class GracefulShutdown:
    """Handle graceful shutdown of async applications."""

    def __init__(self):
        self._shutdown_event = asyncio.Event()
        self._cleanup_handlers: list[Callable[[], Awaitable[None]]] = []

    def register_cleanup(self, handler: Callable[[], Awaitable[None]]) -> None:
        """Register a cleanup handler."""
        self._cleanup_handlers.append(handler)

    def is_shutting_down(self) -> bool:
        """Check if shutdown is in progress."""
        return self._shutdown_event.is_set()

    async def wait_for_shutdown(self) -> None:
        """Wait for shutdown signal."""
        await self._shutdown_event.wait()

    async def shutdown(self) -> None:
        """Perform graceful shutdown."""
        self._shutdown_event.set()
        for handler in reversed(self._cleanup_handlers):
            try:
                await handler()
            except Exception as e:
                print(f"Cleanup error: {e}")

    def setup_signals(self) -> None:
        """Set up signal handlers."""
        loop = asyncio.get_event_loop()
        for sig in (signal.SIGTERM, signal.SIGINT):
            loop.add_signal_handler(
                sig,
                lambda: asyncio.create_task(self.shutdown())
            )


# Usage
shutdown = GracefulShutdown()
shutdown.setup_signals()

shutdown.register_cleanup(close_database)
shutdown.register_cleanup(close_redis)

async def main():
    # Start services
    server = await start_server()

    # Wait for shutdown
    await shutdown.wait_for_shutdown()

    # Server cleanup
    await server.close()