Async Python Patterns
Unverified●28/40Claude Code◐PartialHas SKILL.md but declares no allowed-tools — Claude Code will ask for permission each time
Cursor◐PartialPlain prose you can paste in — but no Cursor rules file
Codex◐PartialPlain prose you can paste in — but no AGENTS.md
Gemini CLI◐PartialPlain prose you can paste in
Copilot◐PartialPlain prose you can paste in — but no Copilot instructions file
npx agentalley add async-python-patternsWho is stuck, and on what
Master Python asyncio, concurrent programming, and async/await patterns for high-performance applications. Use when building async APIs, concurrent systems, or I/O-bound applications requiring non-blocking operations.
The whole source
Frontmatter — 2 properties
| name | async-python-patterns |
|---|---|
| description | Master Python asyncio, concurrent programming, and async/await patterns for high-performance applications. Use when building async APIs, concurrent systems, or I/O-bound applications requiring non-blocking operations. |
| 1 | --- |
| 2 | name: async-python-patterns |
| 3 | description: Master Python asyncio, concurrent programming, and async/await patterns for high-performance applications. Use when building async APIs, concurrent systems, or I/O-bound applications requiring non-blocking operations. |
| 4 | ---A5 — No allowed-tools declared — no way to tell what this skill may touch |
| 5 | |
| 6 | # Async Python Patterns |
| 7 | |
| 8 | Comprehensive guidance for implementing asynchronous Python applications using asyncio, concurrent programming patterns, and async/await for building high-performance, non-blocking systems. |
| 9 | |
| 10 | ## When to Use This Skill |
| 11 | |
| 12 | - Building async web APIs (FastAPI, aiohttp, Sanic) |
| 13 | - Implementing concurrent I/O operations (database, file, network) |
| 14 | - Creating web scrapers with concurrent requests |
| 15 | - Developing real-time applications (WebSocket servers, chat systems) |
| 16 | - Processing multiple independent tasks simultaneously |
| 17 | - Building microservices with async communication |
| 18 | - Optimizing I/O-bound workloads |
| 19 | - Implementing async background tasks and queues |
| 20 | |
| 21 | ## Sync vs Async Decision Guide |
| 22 | |
| 23 | Before adopting async, consider whether it's the right choice for your use case. |
| 24 | |
| 25 | | Use Case | Recommended Approach | |
| 26 | |----------|---------------------| |
| 27 | | Many concurrent network/DB calls | `asyncio` | |
| 28 | | CPU-bound computation | `multiprocessing` or thread pool | |
| 29 | | Mixed I/O + CPU | Offload CPU work with `asyncio.to_thread()` | |
| 30 | | Simple scripts, few connections | Sync (simpler, easier to debug) | |
| 31 | | Web APIs with high concurrency | Async frameworks (FastAPI, aiohttp) | |
| 32 | |
| 33 | **Key Rule:** Stay fully sync or fully async within a call path. Mixing creates hidden blocking and complexity. |
| 34 | |
| 35 | ## Core Concepts |
| 36 | |
| 37 | ### 1. Event Loop |
| 38 | |
| 39 | The event loop is the heart of asyncio, managing and scheduling asynchronous tasks. |
| 40 | |
| 41 | **Key characteristics:** |
| 42 | |
| 43 | - Single-threaded cooperative multitasking |
| 44 | - Schedules coroutines for execution |
| 45 | - Handles I/O operations without blocking |
| 46 | - Manages callbacks and futures |
| 47 | |
| 48 | ### 2. Coroutines |
| 49 | |
| 50 | Functions defined with `async def` that can be paused and resumed. |
| 51 | |
| 52 | **Syntax:** |
| 53 | |
| 54 | ```python |
| 55 | async def my_coroutine(): |
| 56 | result = await some_async_operation() |
| 57 | return result |
| 58 | ``` |
| 59 | |
| 60 | ### 3. Tasks |
| 61 | |
| 62 | Scheduled coroutines that run concurrently on the event loop. |
| 63 | |
| 64 | ### 4. Futures |
| 65 | |
| 66 | Low-level objects representing eventual results of async operations. |
| 67 | |
| 68 | ### 5. Async Context Managers |
| 69 | |
| 70 | Resources that support `async with` for proper cleanup. |
| 71 | |
| 72 | ### 6. Async Iterators |
| 73 | |
| 74 | Objects that support `async for` for iterating over async data sources. |
| 75 | |
| 76 | ## Quick Start |
| 77 | |
| 78 | ```python |
| 79 | import asyncio |
| 80 | |
| 81 | async def main(): |
| 82 | print("Hello") |
| 83 | await asyncio.sleep(1) |
| 84 | print("World") |
| 85 | |
| 86 | # Python 3.7+ |
| 87 | asyncio.run(main()) |
| 88 | ``` |
| 89 | |
| 90 | ## Fundamental Patterns |
| 91 | |
| 92 | ### Pattern 1: Basic Async/Await |
| 93 | |
| 94 | ```python |
| 95 | import asyncio |
| 96 | |
| 97 | async def fetch_data(url: str) -> dict: |
| 98 | """Fetch data from URL asynchronously."""A4 — This skill pulls in web or user content but never says to treat that content as data. A signal, not proof. |
| 99 | await asyncio.sleep(1) # Simulate I/O |
| 100 | return {"url": url, "data": "result"} |
| 101 | |
| 102 | async def main(): |
| 103 | result = await fetch_data("https://api.example.com") |
| 104 | print(result) |
| 105 | |
| 106 | asyncio.run(main()) |
| 107 | ``` |
| 108 | |
| 109 | ### Pattern 2: Concurrent Execution with gather() |
| 110 | |
| 111 | ```python |
| 112 | import asyncio |
| 113 | from typing import List |
| 114 | |
| 115 | async def fetch_user(user_id: int) -> dict: |
| 116 | """Fetch user data.""" |
| 117 | await asyncio.sleep(0.5) |
| 118 | return {"id": user_id, "name": f"User {user_id}"} |
| 119 | |
| 120 | async def fetch_all_users(user_ids: List[int]) -> List[dict]: |
| 121 | """Fetch multiple users concurrently.""" |
| 122 | tasks = [fetch_user(uid) for uid in user_ids] |
| 123 | results = await asyncio.gather(*tasks) |
| 124 | return results |
| 125 | |
| 126 | async def main(): |
| 127 | user_ids = [1, 2, 3, 4, 5] |
| 128 | users = await fetch_all_users(user_ids) |
| 129 | print(f"Fetched {len(users)} users") |
| 130 | |
| 131 | asyncio.run(main()) |
| 132 | ``` |
| 133 | |
| 134 | ### Pattern 3: Task Creation and Management |
| 135 | |
| 136 | ```python |
| 137 | import asyncio |
| 138 | |
| 139 | async def background_task(name: str, delay: int): |
| 140 | """Long-running background task.""" |
| 141 | print(f"{name} started") |
| 142 | await asyncio.sleep(delay) |
| 143 | print(f"{name} completed") |
| 144 | return f"Result from {name}" |
| 145 | |
| 146 | async def main(): |
| 147 | # Create tasks |
| 148 | task1 = asyncio.create_task(background_task("Task 1", 2)) |
| 149 | task2 = asyncio.create_task(background_task("Task 2", 1)) |
| 150 | |
| 151 | # Do other work |
| 152 | print("Main: doing other work") |
| 153 | await asyncio.sleep(0.5) |
| 154 | |
| 155 | # Wait for tasks |
| 156 | result1 = await task1 |
| 157 | result2 = await task2 |
| 158 | |
| 159 | print(f"Results: {result1}, {result2}") |
| 160 | |
| 161 | asyncio.run(main()) |
| 162 | ``` |
| 163 | |
| 164 | ### Pattern 4: Error Handling in Async Code |
| 165 | |
| 166 | ```python |
| 167 | import asyncio |
| 168 | from typing import List, Optional |
| 169 | |
| 170 | async def risky_operation(item_id: int) -> dict: |
| 171 | """Operation that might fail.""" |
| 172 | await asyncio.sleep(0.1) |
| 173 | if item_id % 3 == 0: |
| 174 | raise ValueError(f"Item {item_id} failed") |
| 175 | return {"id": item_id, "status": "success"} |
| 176 | |
| 177 | async def safe_operation(item_id: int) -> Optional[dict]: |
| 178 | """Wrapper with error handling.""" |
| 179 | try: |
| 180 | return await risky_operation(item_id) |
| 181 | except ValueError as e: |
| 182 | print(f"Error: {e}") |
| 183 | return None |
| 184 | |
| 185 | async def process_items(item_ids: List[int]): |
| 186 | """Process multiple items with error handling.""" |
| 187 | tasks = [safe_operation(iid) for iid in item_ids] |
| 188 | results = await asyncio.gather(*tasks, return_exceptions=True) |
| 189 | |
| 190 | # Filter out failures |
| 191 | successful = [r for r in results if r is not None and not isinstance(r, Exception)] |
| 192 | failed = [r for r in results if isinstance(r, Exception)] |
| 193 | |
| 194 | print(f"Success: {len(successful)}, Failed: {len(failed)}") |
| 195 | return successful |
| 196 | |
| 197 | asyncio.run(process_items([1, 2, 3, 4, 5, 6])) |
| 198 | ``` |
| 199 | |
| 200 | ### Pattern 5: Timeout Handling |
| 201 | |
| 202 | ```python |
| 203 | import asyncio |
| 204 | |
| 205 | async def slow_operation(delay: int) -> str: |
| 206 | """Operation that takes time.""" |
| 207 | await asyncio.sleep(delay) |
| 208 | return f"Completed after {delay}s" |
| 209 | |
| 210 | async def with_timeout(): |
| 211 | """Execute operation with timeout.""" |
| 212 | try: |
| 213 | result = await asyncio.wait_for(slow_operation(5), timeout=2.0) |
| 214 | print(result) |
| 215 | except asyncio.TimeoutError: |
| 216 | print("Operation timed out") |
| 217 | |
| 218 | asyncio.run(with_timeout()) |
| 219 | ``` |
| 220 | |
| 221 | ## Detailed worked examples and patterns |
| 222 | |
| 223 | Detailed sections (starting with `## Advanced Patterns`) live in `references/details.md`. Read that file when the navigation summary above is insufficient. |
| 224 | |
| 225 | ## Common Pitfalls |
| 226 | |
| 227 | ### 1. Forgetting await |
| 228 | |
| 229 | ```python |
| 230 | # Wrong - returns coroutine object, doesn't execute |
| 231 | result = async_function() |
| 232 | |
| 233 | # Correct |
| 234 | result = await async_function() |
| 235 | ``` |
| 236 | |
| 237 | ### 2. Blocking the Event Loop |
| 238 | |
| 239 | ```python |
| 240 | # Wrong - blocks event loop |
| 241 | import time |
| 242 | async def bad(): |
| 243 | time.sleep(1) # Blocks! |
| 244 | |
| 245 | # Correct |
| 246 | async def good(): |
| 247 | await asyncio.sleep(1) # Non-blocking |
| 248 | ``` |
| 249 | |
| 250 | ### 3. Not Handling Cancellation |
| 251 | |
| 252 | ```python |
| 253 | async def cancelable_task(): |
| 254 | """Task that handles cancellation.""" |
| 255 | try: |
| 256 | while True: |
| 257 | await asyncio.sleep(1) |
| 258 | print("Working...") |
| 259 | except asyncio.CancelledError: |
| 260 | print("Task cancelled, cleaning up...") |
| 261 | # Perform cleanup |
| 262 | raise # Re-raise to propagate cancellation |
| 263 | ``` |
| 264 | |
| 265 | ### 4. Mixing Sync and Async Code |
| 266 | |
| 267 | ```python |
| 268 | # Wrong - can't call async from sync directly |
| 269 | def sync_function(): |
| 270 | result = await async_function() # SyntaxError! |
| 271 | |
| 272 | # Correct |
| 273 | def sync_function(): |
| 274 | result = asyncio.run(async_function()) |
| 275 | ``` |
| 276 | |
| 277 | ## Testing Async Code |
| 278 | |
| 279 | ```python |
| 280 | import asyncio |
| 281 | import pytest |
| 282 | |
| 283 | # Using pytest-asyncio |
| 284 | @pytest.mark.asyncio |
| 285 | async def test_async_function(): |
| 286 | """Test async function.""" |
| 287 | result = await fetch_data("https://api.example.com") |
| 288 | assert result is not None |
| 289 | |
| 290 | @pytest.mark.asyncio |
| 291 | async def test_with_timeout(): |
| 292 | """Test with timeout.""" |
| 293 | with pytest.raises(asyncio.TimeoutError): |
| 294 | await asyncio.wait_for(slow_operation(5), timeout=1.0) |
| 295 | ``` |
| 296 |
Reviews
Installed this one?Write the first review and take the Trailblazer badge.
Alternatives
Subagent Driven DevelopmentUse when executing implementation plans with independent tasks in the current session◐◐◐◐◐●36/40Python Code Style & DocumentationPython code style, linting, formatting, naming conventions, and documentation standards. Use when writing new code, reviewing style, configuring linters, writing docstrings, or establishing project standards.◐····●35/40Competitor Price Analysis 💲Competitor pricing strategy analysis and market positioning. Price mapping, pricing gaps identification, elasticity signals evaluation, and strategic pricing optimization. Use when the user asks about competitor pricing, price analysis, pricing strategy, or co◐····●34/40Competitor Price Tracker 📊Set up competitor price tracking and monitoring workflows. Track price changes, detect promotions, analyze pricing patterns, and get alerts for competitive price movements.◐····●34/40