python-async-patterns

name	python-async-patterns
description	5 async patterns with full implementations for Python concurrent programming
version	1.0.0
category	toolchain
author	Claude MPM Team
license	MIT
progressive_disclosure	{"entry_point":{"summary":"5 production-ready async patterns: gather, worker pools, retry with backoff, TaskGroup, AsyncWorkerPool","when_to_use":"When implementing async/concurrent operations in Python","quick_start":"Choose pattern based on use case: gather for parallel ops, worker pool for rate limiting, retry for unreliable services"}}
context_limit	700
tags	["python","async","asyncio","concurrency","worker-pool","retry","backoff","gather","task-group"]
requires_tools	[]

Async Programming Patterns

Concurrent Task Execution

# Pattern 1: Gather with timeout and error handling
async def process_concurrent_tasks(
    tasks: list[Coroutine[Any, Any, T]],
    timeout: float = 10.0
) -> list[T | Exception]:
    """Process tasks concurrently with timeout and exception handling."""
    try:
        async with asyncio.timeout(timeout):  # Python 3.11+
            # return_exceptions=True prevents one failure from cancelling others
            return await asyncio.gather(*tasks, return_exceptions=True)
    except asyncio.TimeoutError:
        logger.warning("Tasks timed out after %s seconds", timeout)
        raise

Worker Pool with Concurrency Control

# Pattern 2: Semaphore-based worker pool
async def worker_pool(
    tasks: list[Callable[[], Coroutine[Any, Any, T]]],
    max_workers: int = 10
) -> list[T]:
    """Execute tasks with bounded concurrency using semaphore."""
    semaphore = asyncio.Semaphore(max_workers)

    async def bounded_task(task: Callable) -> T:
          semaphore:
              task()

      asyncio.gather(*[bounded_task(t)  t  tasks])