Async Python Patterns
Apply proven asyncio patterns when your solo project needs concurrent HTTP, DB access, or streaming without blocking.
Overview
async-python-patterns is an agent skill for the Build phase that supplies detailed asyncio worked examples for backend concurrent I/O.
Install
npx skills add https://github.com/wshobson/agents --skill async-python-patternsWhat is this skill?
- Worked examples for async context managers and cleanup-safe resource handling
- Async iterators and generators for paginated or delayed streams
- Patterns map directly to I/O-bound APIs and agent tool loops
- Copy-paste snippets for asyncio.run and structured concurrency habits
- Advanced sections beyond basic async/await for production-style code
Adoption & trust: 11.4k installs on skills.sh; 36.5k GitHub stars; 2/3 security scanners passed (skills.sh audits).
What problem does it solve?
You need non-blocking Python I/O but keep producing blocking calls, leaky connections, or fragile async control flow.
Who is it for?
Indie builders implementing Python backends, scrapers, or agent tools that must handle many concurrent I/O operations correctly.
Skip if: Greenfield learners who have not used async/await yet, or teams standardizing on threaded-only sync stacks with no asyncio adoption.
When should I use this skill?
Implementing or refactoring asyncio code for I/O-bound Python services, workers, or integrations.
What do I get? / Deliverables
Your agent implements asyncio patterns with proper context managers, async iteration, and structured concurrency suitable for APIs and workers.
- Runnable asyncio pattern modules or snippets aligned to your backend
- Connection and pagination flows using async context managers and generators
Recommended Skills
Journey fit
Async Python is implemented while building server-side logic, workers, and integration code—not during idea or launch marketing. Patterns target backend concurrency (context managers, async generators, pagination fetchers), which belong on the backend shelf.
How it compares
Use instead of asking the model for generic async snippets without lifecycle or pagination patterns baked in.
Common Questions / FAQ
Who is async-python-patterns for?
Solo Python builders and small teams shipping APIs, CLIs, or agent backends who want copy-paste asyncio patterns rather than reinventing concurrency每个 time.
When should I use async-python-patterns?
During Build backend work when you add async database access, HTTP clients, streaming consumers, or paginated fetchers; also when refactoring sync I/O that blocks your event loop.
Is async-python-patterns safe to install?
It is instructional code patterns without mandatory shell or secret access; review the Security Audits panel on this page before trusting any third-party skill package in your agent.
SKILL.md
READMESKILL.md - Async Python Patterns
# async-python-patterns — detailed worked examples ## Advanced Patterns ### Pattern 6: Async Context Managers ```python import asyncio from typing import Optional class AsyncDatabaseConnection: """Async database connection context manager.""" def __init__(self, dsn: str): self.dsn = dsn self.connection: Optional[object] = None async def __aenter__(self): print("Opening connection") await asyncio.sleep(0.1) # Simulate connection self.connection = {"dsn": self.dsn, "connected": True} return self.connection async def __aexit__(self, exc_type, exc_val, exc_tb): print("Closing connection") await asyncio.sleep(0.1) # Simulate cleanup self.connection = None async def query_database(): """Use async context manager.""" async with AsyncDatabaseConnection("postgresql://localhost") as conn: print(f"Using connection: {conn}") await asyncio.sleep(0.2) # Simulate query return {"rows": 10} asyncio.run(query_database()) ``` ### Pattern 7: Async Iterators and Generators ```python import asyncio from typing import AsyncIterator async def async_range(start: int, end: int, delay: float = 0.1) -> AsyncIterator[int]: """Async generator that yields numbers with delay.""" for i in range(start, end): await asyncio.sleep(delay) yield i async def fetch_pages(url: str, max_pages: int) -> AsyncIterator[dict]: """Fetch paginated data asynchronously.""" for page in range(1, max_pages + 1): await asyncio.sleep(0.2) # Simulate API call yield { "page": page, "url": f"{url}?page={page}", "data": [f"item_{page}_{i}" for i in range(5)] } async def consume_async_iterator(): """Consume async iterator.""" async for number in async_range(1, 5): print(f"Number: {number}") print("\nFetching pages:") async for page_data in fetch_pages("https://api.example.com/items", 3): print(f"Page {page_data['page']}: {len(page_data['data'])} items") asyncio.run(consume_async_iterator()) ``` ### Pattern 8: Producer-Consumer Pattern ```python import asyncio from asyncio import Queue from typing import Optional async def producer(queue: Queue, producer_id: int, num_items: int): """Produce items and put them in queue.""" for i in range(num_items): item = f"Item-{producer_id}-{i}" await queue.put(item) print(f"Producer {producer_id} produced: {item}") await asyncio.sleep(0.1) await queue.put(None) # Signal completion async def consumer(queue: Queue, consumer_id: int): """Consume items from queue.""" while True: item = await queue.get() if item is None: queue.task_done() break print(f"Consumer {consumer_id} processing: {item}") await asyncio.sleep(0.2) # Simulate work queue.task_done() async def producer_consumer_example(): """Run producer-consumer pattern.""" queue = Queue(maxsize=10) # Create tasks producers = [ asyncio.create_task(producer(queue, i, 5)) for i in range(2) ] consumers = [ asyncio.create_task(consumer(queue, i)) for i in range(3) ] # Wait for producers await asyncio.gather(*producers) # Wait for queue to be empty await queue.join() # Cancel consumers for c in consumers: c.cancel() asyncio.run(producer_consumer_example()) ``` ### Pattern 9: Semaphore for Rate Limiting ```python import asyncio from typing import List async def api_call(url: str, semaphore: asyncio.Semaphore) -> dict: """Make API call with rate limiting.""" async with semaphore: print(f"Calling {url}") await asyncio.sleep(0.5) # Simulate API call return {"url": url, "status": 200} async def rate_limited_requests(urls: List[str], max_concurrent: int = 5): """Make multiple requests with rate li