aiskillstore/python-async-patterns
Python asyncio patterns for concurrent programming. Triggers on: asyncio, async, await, coroutine, gather, semaphore, TaskGroup, event loop, aiohttp, concurrent.
This is a copy. The original lives at comeonoliver/python-async-patterns.
npx skills add https://github.com/aiskillstore/marketplace --skill python-async-patternsAsyncio patterns for concurrent Python programming.
import asyncio
# 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())
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)
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])
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
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
# 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
| 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 |
from contextlib import asynccontextmanager
@asynccontextmanager
async def managed_connection():
conn = await create_connection()
try:
yield conn
finally:
await conn.close()
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/find-blocking-calls.sh - Scan code for blocking calls in async functions./assets/async-project-template.py - Production-ready async app skeletonPrerequisites:
python-typing-patterns - Type hints for async functionsRelated Skills:
python-fastapi-patterns - Async web APIspython-observability-patterns - Async logging and tracingpython-database-patterns - Async database accessTake aiskillstore/python-async-patterns from the repository into ~/.claude/skills for personal
use, or into .claude/skills inside a project.
The agent identifies a skill by the name field in its header. Two skills with the
same name cannot sit side by side — one of them will be ignored.