Asyncio Advanced

Asyncio Advanced Patterns ()

Modern Python asyncio patterns using structured concurrency, TaskGroup, and Python 3.11+ features.

Overview

• Implementing concurrent HTTP requests or database queries
• Building async services with proper cancellation handling
• Managing multiple concurrent tasks with error propagation
• Rate limiting async operations with semaphores
• Bridging sync code to async contexts

Quick Reference

TaskGroup (Replaces gather)

import asyncio

async def fetch_user_data(user_id: str) -> dict:
    """Fetch user data concurrently - all tasks complete or all cancelled."""
    async with asyncio.TaskGroup() as tg:
        user_task = tg.create_task(fetch_user(user_id))
        orders_task = tg.create_task(fetch_orders(user_id))
        preferences_task = tg.create_task(fetch_preferences(user_id))

    # All tasks guaranteed complete here
    return {
        "user": user_task.result(),
        "orders": orders_task.result(),
        "preferences": preferences_task.result(),
    }

TaskGroup with Timeout

async def fetch_with_timeout(urls: list[str], timeout_sec: float = 30) -> list[dict]:
    """Fetch all URLs with overall timeout - structured concurrency."""
    results = []

    async with asyncio.timeout(timeout_sec):
        async with asyncio.TaskGroup() as tg:
            tasks = [tg.create_task(fetch_url(url)) for url in urls]

    return [t.result() for t in tasks]

Semaphore for Concurrency Limiting

class RateLimitedClient:
    """HTTP client with concurrency limiting."""

    def __init__(self, max_concurrent: int = 10):
        self._semaphore = asyncio.Semaphore(max_concurrent)
        self._session: aiohttp.ClientSession | None = None

    async def fetch(self, url: str) -> dict:
        async with self._semaphore:  # Limit concurrent requests
            async with self._session.get(url) as response:
                return await response.json()

    async def fetch_many(self, urls: list[str]) -> list[dict]:
        async with asyncio.TaskGroup() as tg:
            tasks = [tg.create_task(self.fetch(url)) for url in urls]
        return [t.result() for t in tasks]

Exception Group Handling

async def process_batch(items: list[dict]) -> tuple[list[dict], list[Exception]]:
    """Process batch, collecting both successes and failures."""
    results = []
    errors = []

    try:
        async with asyncio.TaskGroup() as tg:
            tasks = [tg.create_task(process_item(item)) for item in items]
    except* ValueError as eg:
        # Handle specific exception types from ExceptionGroup
        errors.extend(eg.exceptions)
    except* Exception as eg:
        errors.extend(eg.exceptions)
    else:
        results = [t.result() for t in tasks]

    return results, errors

Sync-to-Async Bridge

import asyncio
from concurrent.futures import ThreadPoolExecutor

# For CPU-bound or blocking sync code
async def run_blocking_operation(data: bytes) -> dict:
    """Run blocking sync code in thread pool."""
    return await asyncio.to_thread(cpu_intensive_parse, data)

# For sync code that needs async context
def sync_caller():
    """Call async code from sync context (not in existing loop)."""
    return asyncio.run(async_main())

# For sync code within existing async context
async def wrapper_for_sync_lib():
    """Bridge sync library to async - use with care."""
    loop = asyncio.get_running_loop()
    with ThreadPoolExecutor() as pool:
        result = await loop.run_in_executor(pool, sync_blocking_call)
    return result

Cancellation Handling

async def cancellable_operation(resource_id: str) -> dict:
    """Properly handle cancellation - NEVER swallow CancelledError."""
    resource = await acquire_resource(resource_id)
    try:
        return await process_resource(resource)
    except asyncio.CancelledError:
        # Clean up but RE-RAISE - this is critical!
        await cleanup_resource(resource)
        raise  # ALWAYS re-raise CancelledError
    finally:
        await release_resource(resource)

Key Decisions

Anti-Patterns (FORBIDDEN)

# NEVER use gather() for new code - no structured concurrency
results = await asyncio.gather(task1(), task2())  # LEGACY

# NEVER swallow CancelledError - breaks structured concurrency
except asyncio.CancelledError:
    return None  # BREAKS TaskGroup and timeout!

# NEVER use create_task() without TaskGroup - tasks leak
asyncio.create_task(background_work())  # Fire and forget = leaked task

# NEVER yield inside async context managers (PEP 789)
async with asyncio.timeout(10):
    yield item  # DANGEROUS - cancellation bugs!

# NEVER use asyncio.run() inside existing event loop
async def handler():
    asyncio.run(other_async())  # CRASHES - loop already running

# NEVER block the event loop with sync calls
async def bad_handler():
    time.sleep(1)  # BLOCKS ALL TASKS
    requests.get(url)  # BLOCKS ALL TASKS

Related Skills

• sqlalchemy-2-async - Async database sessions with SQLAlchemy 2.0
• fastapi-advanced - Async FastAPI patterns
• background-jobs - Celery/ARQ for heavy async work
• streaming-api-patterns - SSE/WebSocket async patterns

Capability Details

taskgroup-patterns

Keywords: taskgroup, structured concurrency, concurrent tasks, parallel execution Solves:

• How do I run multiple async tasks concurrently?
• Replace asyncio.gather with TaskGroup
• Handle exceptions from multiple tasks

timeout-patterns

Keywords: timeout, asyncio.timeout, cancel, deadline Solves:

• How do I add timeouts to async operations?
• Timeout multiple concurrent operations
• Cancel tasks after deadline

semaphore-limiting

Keywords: semaphore, rate limit, concurrency limit, throttle Solves:

• How do I limit concurrent async operations?
• Rate limit HTTP requests
• Prevent connection pool exhaustion

exception-groups

Keywords: ExceptionGroup, except, multiple exceptions, error handling Solves:*

• How do I handle multiple task failures?
• Collect errors from concurrent operations
• Python 3.11+ exception group patterns

sync-async-bridge

Keywords: to_thread, run_in_executor, sync to async, blocking code Solves:

• How do I call sync code from async?
• Run CPU-bound code without blocking
• Bridge sync libraries to async context

Async Implementation Checklist

Before Starting

• [ ] Python version >= 3.11 (for TaskGroup, ExceptionGroup)
• [ ] Using async-compatible libraries (aiohttp, asyncpg, aiofiles)
• [ ] No blocking sync calls in async code paths

TaskGroup Usage

• [ ] Using async with asyncio.TaskGroup() instead of asyncio.gather()
• [ ] All tasks created with tg.create_task()
• [ ] Handling ExceptionGroup with except* syntax
• [ ] No fire-and-forget create_task() outside TaskGroup

Timeout Handling

• [ ] Using async with asyncio.timeout(seconds) for deadlines
• [ ] Timeout values are reasonable for the operation
• [ ] Timeout exceptions are caught and handled appropriately

Cancellation Safety

• [ ] Never swallowing asyncio.CancelledError
• [ ] Always re-raising CancelledError after cleanup
• [ ] Resources cleaned up in finally blocks
• [ ] No yield inside async with timeout/taskgroup contexts

Concurrency Limiting

• [ ] Using asyncio.Semaphore for rate limiting
• [ ] Semaphore created once, not per-call
• [ ] Semaphore combined with timeout to prevent deadlock
• [ ] Max concurrency matches resource limits (connections, API rate)

Sync-to-Async Bridge

• [ ] Using asyncio.to_thread() for blocking sync code
• [ ] Not calling asyncio.run() inside async context
• [ ] Thread pool sized appropriately for workload
• [ ] CPU-bound work offloaded to process pool or worker

Testing

• [ ] Using pytest-asyncio for async tests
• [ ] Tests use @pytest.mark.asyncio decorator
• [ ] Mock async functions return coroutines or use AsyncMock
• [ ] Timeouts added to prevent hanging tests

Performance

• [ ] Connection pools are shared (not created per-request)
• [ ] HTTP sessions reused across requests
• [ ] Batch operations where possible
• [ ] Monitoring for event loop blocking (> 100ms)

Asyncio Advanced Examples

Example 1: Concurrent API Fetching with TaskGroup

import asyncio
import aiohttp
from dataclasses import dataclass

@dataclass
class UserData:
    profile: dict
    orders: list
    preferences: dict

async def fetch_user_dashboard(user_id: str) -> UserData:
    """Fetch all user data concurrently with proper error handling."""
    async with aiohttp.ClientSession() as session:
        async with asyncio.TaskGroup() as tg:
            profile_task = tg.create_task(
                fetch_json(session, f"/api/users/{user_id}")
            )
            orders_task = tg.create_task(
                fetch_json(session, f"/api/users/{user_id}/orders")
            )
            prefs_task = tg.create_task(
                fetch_json(session, f"/api/users/{user_id}/preferences")
            )

        return UserData(
            profile=profile_task.result(),
            orders=orders_task.result(),
            preferences=prefs_task.result(),
        )

async def fetch_json(session: aiohttp.ClientSession, path: str) -> dict:
    async with session.get(f"https://api.example.com{path}") as resp:
        return await resp.json()

Example 2: Rate-Limited Bulk Processing

import asyncio
from typing import TypeVar, Callable, Awaitable

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

class BulkProcessor:
    """Process items with concurrency and rate limiting."""

    def __init__(self, max_concurrent: int = 10, timeout: float = 30.0):
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.timeout = timeout

    async def process_all(
        self,
        items: list[T],
        processor: Callable[[T], Awaitable[R]],
    ) -> tuple[list[R], list[tuple[T, Exception]]]:
        """Process all items, returning successes and failures."""
        successes: list[R] = []
        failures: list[tuple[T, Exception]] = []

        async def process_one(item: T) -> tuple[T, R | Exception]:
            async with self.semaphore:
                try:
                    async with asyncio.timeout(self.timeout):
                        result = await processor(item)
                        return (item, result)
                except Exception as e:
                    return (item, e)

        async with asyncio.TaskGroup() as tg:
            tasks = [tg.create_task(process_one(item)) for item in items]

        for task in tasks:
            item, result = task.result()
            if isinstance(result, Exception):
                failures.append((item, result))
            else:
                successes.append(result)

        return successes, failures

# Usage
processor = BulkProcessor(max_concurrent=5, timeout=10.0)
successes, failures = await processor.process_all(
    items=urls,
    processor=fetch_and_parse,
)
print(f"Processed {len(successes)}, failed {len(failures)}")

Example 3: Graceful Shutdown with Cleanup

import asyncio
import signal
from contextlib import asynccontextmanager

class AsyncService:
    def __init__(self):
        self._shutdown_event = asyncio.Event()
        self._tasks: set[asyncio.Task] = set()

    async def start(self):
        """Start service with graceful shutdown handling."""
        loop = asyncio.get_running_loop()

        # Handle signals
        for sig in (signal.SIGTERM, signal.SIGINT):
            loop.add_signal_handler(
                sig,
                lambda: asyncio.create_task(self.shutdown())
            )

        try:
            async with asyncio.TaskGroup() as tg:
                tg.create_task(self.worker_loop())
                tg.create_task(self.health_check_loop())
                tg.create_task(self._wait_for_shutdown())
        except* asyncio.CancelledError:
            pass  # Expected on shutdown

    async def _wait_for_shutdown(self):
        await self._shutdown_event.wait()
        raise asyncio.CancelledError()

    async def shutdown(self):
        """Graceful shutdown - complete current work."""
        print("Shutting down...")
        self._shutdown_event.set()

    async def worker_loop(self):
        while not self._shutdown_event.is_set():
            try:
                async with asyncio.timeout(5.0):
                    await self.process_next_job()
            except asyncio.TimeoutError:
                continue
            except asyncio.CancelledError:
                print("Worker cancelled, finishing current job...")
                raise

    async def health_check_loop(self):
        while not self._shutdown_event.is_set():
            await asyncio.sleep(30)
            await self.health_check()

Example 4: Exception Group Handling

import asyncio

async def fetch_from_multiple_sources(query: str) -> list[dict]:
    """Try multiple sources, collect partial results on failures."""
    sources = ["source_a", "source_b", "source_c"]
    results = []
    errors = []

    try:
        async with asyncio.TaskGroup() as tg:
            tasks = {
                source: tg.create_task(fetch_from_source(source, query))
                for source in sources
            }
    except* ConnectionError as eg:
        # Some sources failed with connection errors
        errors.extend(eg.exceptions)
        # Collect successful results
        for source, task in tasks.items():
            if task.done() and not task.exception():
                results.append(task.result())
    except* TimeoutError as eg:
        errors.extend(eg.exceptions)
        for source, task in tasks.items():
            if task.done() and not task.exception():
                results.append(task.result())
    else:
        # All succeeded
        results = [t.result() for t in tasks.values()]

    if errors:
        print(f"Partial results: {len(results)} succeeded, {len(errors)} failed")

    return results

Example 5: Async Context Manager with Resource Pool

import asyncio
from contextlib import asynccontextmanager
from typing import AsyncIterator

class ConnectionPool:
    """Async connection pool with proper lifecycle management."""

    def __init__(self, max_size: int = 10):
        self._semaphore = asyncio.Semaphore(max_size)
        self._connections: asyncio.Queue = asyncio.Queue(maxsize=max_size)
        self._initialized = False

    async def initialize(self, dsn: str):
        """Pre-create connections."""
        for _ in range(self._semaphore._value):
            conn = await create_connection(dsn)
            await self._connections.put(conn)
        self._initialized = True

    @asynccontextmanager
    async def acquire(self) -> AsyncIterator[Connection]:
        """Get a connection from the pool."""
        async with self._semaphore:
            conn = await self._connections.get()
            try:
                yield conn
            finally:
                # Return connection to pool
                if conn.is_healthy():
                    await self._connections.put(conn)
                else:
                    # Replace unhealthy connection
                    new_conn = await create_connection(self._dsn)
                    await self._connections.put(new_conn)

    async def close(self):
        """Close all connections."""
        while not self._connections.empty():
            conn = await self._connections.get()
            await conn.close()

# Usage
pool = ConnectionPool(max_size=20)
await pool.initialize("postgres://...")

async with pool.acquire() as conn:
    result = await conn.execute("SELECT * FROM users")

Semaphore Patterns for Concurrency Limiting

Basic Rate Limiting

import asyncio
import aiohttp

class RateLimitedClient:
    """HTTP client with concurrency and rate limiting."""

    def __init__(
        self,
        max_concurrent: int = 10,
        requests_per_second: float = 100,
    ):
        self._semaphore = asyncio.Semaphore(max_concurrent)
        self._rate_limiter = AsyncRateLimiter(requests_per_second)
        self._session: aiohttp.ClientSession | None = None

    async def __aenter__(self):
        self._session = aiohttp.ClientSession()
        return self

    async def __aexit__(self, *args):
        if self._session:
            await self._session.close()

    async def get(self, url: str) -> dict:
        async with self._semaphore:
            await self._rate_limiter.acquire()
            async with self._session.get(url) as resp:
                return await resp.json()


class AsyncRateLimiter:
    """Token bucket rate limiter."""

    def __init__(self, rate: float):
        self._rate = rate
        self._tokens = rate
        self._last_update = asyncio.get_event_loop().time()
        self._lock = asyncio.Lock()

    async def acquire(self):
        async with self._lock:
            now = asyncio.get_event_loop().time()
            self._tokens = min(
                self._rate,
                self._tokens + (now - self._last_update) * self._rate
            )
            self._last_update = now

            if self._tokens < 1:
                wait_time = (1 - self._tokens) / self._rate
                await asyncio.sleep(wait_time)
                self._tokens = 0
            else:
                self._tokens -= 1

Database Connection Limiting

class DatabasePool:
    """Async database pool with connection limiting."""

    def __init__(self, dsn: str, max_connections: int = 20):
        self._dsn = dsn
        self._semaphore = asyncio.Semaphore(max_connections)
        self._pool = None

    async def execute(self, query: str, *args) -> list:
        async with self._semaphore:
            async with self._pool.acquire() as conn:
                return await conn.fetch(query, *args)

    async def execute_many(self, queries: list[tuple[str, tuple]]) -> list:
        """Execute multiple queries with connection limiting."""
        async with asyncio.TaskGroup() as tg:
            tasks = [
                tg.create_task(self.execute(q, *args))
                for q, args in queries
            ]
        return [t.result() for t in tasks]

Bounded Work Queue

class BoundedWorkQueue:
    """Process items with bounded concurrency."""

    def __init__(self, max_workers: int = 10):
        self._semaphore = asyncio.Semaphore(max_workers)
        self._results: list = []

    async def process_all(
        self,
        items: list,
        processor: Callable[[Any], Awaitable[Any]],
    ) -> list:
        async def bounded_process(item):
            async with self._semaphore:
                return await processor(item)

        async with asyncio.TaskGroup() as tg:
            tasks = [tg.create_task(bounded_process(item)) for item in items]

        return [t.result() for t in tasks]

Common Pitfalls

# WRONG: Creating semaphore inside coroutine
async def bad_fetch(url: str):
    sem = asyncio.Semaphore(10)  # New semaphore each call!
    async with sem:
        return await fetch(url)

# CORRECT: Share semaphore across calls
SEM = asyncio.Semaphore(10)

async def good_fetch(url: str):
    async with SEM:
        return await fetch(url)

# WRONG: Semaphore without timeout
async with sem:
    await potentially_slow_operation()  # Can block other tasks indefinitely

# CORRECT: Semaphore with timeout
async with asyncio.timeout(30):
    async with sem:
        await potentially_slow_operation()

TaskGroup Patterns

Basic TaskGroup Usage

import asyncio
from typing import TypeVar

T = TypeVar("T")

async def fetch_all_concurrent(tasks: list[Coroutine[Any, Any, T]]) -> list[T]:
    """Run all tasks concurrently, fail-fast on any exception."""
    async with asyncio.TaskGroup() as tg:
        created = [tg.create_task(task) for task in tasks]
    return [t.result() for t in created]

TaskGroup with Partial Failure Handling

async def fetch_with_partial_failures(urls: list[str]) -> tuple[list[dict], list[str]]:
    """Collect successes and failures separately."""
    successes = []
    failures = []

    try:
        async with asyncio.TaskGroup() as tg:
            tasks = [(url, tg.create_task(fetch(url))) for url in urls]
    except* Exception as eg:
        # TaskGroup failed - collect individual results
        for url, task in tasks:
            if task.done():
                try:
                    successes.append(task.result())
                except Exception:
                    failures.append(url)
            else:
                failures.append(url)
    else:
        successes = [t.result() for _, t in tasks]

    return successes, failures

TaskGroup with Timeout per Task

async def fetch_with_individual_timeouts(
    items: list[dict],
    timeout_per_item: float = 5.0,
) -> list[dict | None]:
    """Each task has its own timeout."""
    async def fetch_with_timeout(item: dict) -> dict | None:
        try:
            async with asyncio.timeout(timeout_per_item):
                return await process_item(item)
        except asyncio.TimeoutError:
            return None

    async with asyncio.TaskGroup() as tg:
        tasks = [tg.create_task(fetch_with_timeout(item)) for item in items]

    return [t.result() for t in tasks]

TaskGroup vs gather Comparison

When to Still Use gather

# Only for Python 3.10 compatibility or return_exceptions pattern
results = await asyncio.gather(
    *tasks,
    return_exceptions=True  # Collect all, don't fail fast
)

# Filter successes and failures
successes = [r for r in results if not isinstance(r, Exception)]
failures = [r for r in results if isinstance(r, Exception)]