Coroutine (must be awaited)

async def fetch(url: str) -> str: async with aiohttp.ClientSession() as session: async with session.get(url) as response: return await response.text()

Entry point

async def main(): result = await fetch("https://example.com") return result

asyncio.run(main())

Pattern 1: Concurrent with gather async def fetch_all(urls: list[str]) -> list[str]: """Fetch multiple URLs concurrently.""" async with aiohttp.ClientSession() as session: tasks = [fetch_one(session, url) for url in urls] return await asyncio.gather(*tasks, return_exceptions=True)

Pattern 2: Bounded Concurrency async def fetch_with_limit(urls: list[str], limit: int = 10): """Limit concurrent requests.""" semaphore = asyncio.Semaphore(limit)

async def bounded_fetch(url):
    async with semaphore:
        return await fetch_one(url)

return await asyncio.gather(*[bounded_fetch(url) for url in urls])

Pattern 3: TaskGroup (Python 3.11+) async def process_items(items): """Structured concurrency with automatic cleanup.""" async with asyncio.TaskGroup() as tg: for item in items: tg.create_task(process_one(item)) # All tasks complete here, or exception raised

Pattern 4: Timeout async def with_timeout(): try: async with asyncio.timeout(5.0): # Python 3.11+ result = await slow_operation() except asyncio.TimeoutError: result = None return result

Critical Warnings

WRONG - blocks event loop

async def bad(): time.sleep(5) # Never use time.sleep! requests.get(url) # Blocking I/O!

CORRECT

async def good(): await asyncio.sleep(5) async with aiohttp.ClientSession() as s: await s.get(url)

WRONG - orphaned task

async def bad(): asyncio.create_task(work()) # May be garbage collected!

CORRECT - keep reference

async def good(): task = asyncio.create_task(work()) await task

Quick Reference Pattern Use Case gather(*tasks) Multiple independent operations Semaphore(n) Rate limiting, resource constraints TaskGroup() Structured concurrency (3.11+) Queue() Producer-consumer timeout(s) Timeout wrapper (3.11+) Lock() Shared mutable state Async Context Manager from contextlib import asynccontextmanager

@asynccontextmanager async def managed_connection(): conn = await create_connection() try: yield conn finally: await conn.close()

Additional Resources

For detailed patterns, load:

./references/concurrency-patterns.md - Queue, Lock, producer-consumer ./references/aiohttp-patterns.md - HTTP client/server patterns ./references/mixing-sync-async.md - run_in_executor, thread pools ./references/debugging-async.md - Debug mode, profiling, finding issues ./references/production-patterns.md - Graceful shutdown, health checks, signal handling ./references/error-handling.md - Retry with backoff, circuit breakers, partial failures ./references/performance.md - uvloop, connection pooling, buffer sizing Scripts ./scripts/find-blocking-calls.sh - Scan code for blocking calls in async functions Assets ./assets/async-project-template.py - Production-ready async app skeleton See Also

Prerequisites:

python-typing-patterns - Type hints for async functions

Related Skills:

python-fastapi-patterns - Async web APIs python-observability-patterns - Async logging and tracing python-database-patterns - Async database access

python-async-patterns

安装