Loading...
Loading...
Expert in asynchronous programming patterns across languages (Python asyncio, JavaScript/TypeScript promises, C# async/await, Rust futures). Use for concurrent programming, event loops, async patterns, error handling, backpressure, cancellation, and performance optimization in async systems.
npx skill4agent add martinholovsky/claude-skills-generator async-expert# tests/test_data_fetcher.py
import pytest
import asyncio
from unittest.mock import AsyncMock, patch
@pytest.mark.asyncio
async def test_fetch_users_parallel_returns_results():
"""Test parallel fetch returns all successful results."""
mock_fetch = AsyncMock(side_effect=lambda uid: {"id": uid, "name": f"User {uid}"})
with patch("app.fetcher.fetch_user", mock_fetch):
from app.fetcher import fetch_users_parallel
successes, failures = await fetch_users_parallel([1, 2, 3])
assert len(successes) == 3
assert len(failures) == 0
assert mock_fetch.call_count == 3
@pytest.mark.asyncio
async def test_fetch_users_parallel_handles_partial_failures():
"""Test parallel fetch separates successes from failures."""
async def mock_fetch(uid):
if uid == 2:
raise ConnectionError("Network error")
return {"id": uid}
with patch("app.fetcher.fetch_user", mock_fetch):
from app.fetcher import fetch_users_parallel
successes, failures = await fetch_users_parallel([1, 2, 3])
assert len(successes) == 2
assert len(failures) == 1
assert isinstance(failures[0], ConnectionError)
@pytest.mark.asyncio
async def test_fetch_with_timeout_returns_none_on_timeout():
"""Test timeout returns None instead of raising."""
async def slow_fetch():
await asyncio.sleep(10)
return "data"
with patch("app.fetcher.fetch_data", slow_fetch):
from app.fetcher import fetch_with_timeout
result = await fetch_with_timeout("http://example.com", timeout=0.1)
assert result is None# app/fetcher.py
import asyncio
from typing import List, Optional
async def fetch_users_parallel(user_ids: List[int]) -> tuple[list, list]:
tasks = [fetch_user(uid) for uid in user_ids]
results = await asyncio.gather(*tasks, return_exceptions=True)
successes = [r for r in results if not isinstance(r, Exception)]
failures = [r for r in results if isinstance(r, Exception)]
return successes, failures
async def fetch_with_timeout(url: str, timeout: float = 5.0) -> Optional[str]:
try:
async with asyncio.timeout(timeout):
return await fetch_data(url)
except asyncio.TimeoutError:
return None# Run async tests
pytest tests/ -v --asyncio-mode=auto
# Check for blocking calls
grep -r "time\.sleep\|requests\.\|urllib\." src/
# Run with coverage
pytest --cov=app --cov-report=term-missing# BAD: Sequential - 3 seconds total
async def fetch_all_sequential():
user = await fetch_user() # 1 sec
posts = await fetch_posts() # 1 sec
comments = await fetch_comments() # 1 sec
return user, posts, comments
# GOOD: Parallel - 1 second total
async def fetch_all_parallel():
return await asyncio.gather(
fetch_user(),
fetch_posts(),
fetch_comments()
)# BAD: Unbounded concurrency overwhelms server
async def process_all_bad(items):
return await asyncio.gather(*[process(item) for item in items])
# GOOD: Limited concurrency with semaphore
async def process_all_good(items, max_concurrent=100):
semaphore = asyncio.Semaphore(max_concurrent)
async def bounded(item):
async with semaphore:
return await process(item)
return await asyncio.gather(*[bounded(item) for item in items])# BAD: Manual task management
async def fetch_all_manual():
tasks = [asyncio.create_task(fetch(url)) for url in urls]
try:
return await asyncio.gather(*tasks)
except Exception:
for task in tasks:
task.cancel()
raise
# GOOD: TaskGroup handles cancellation automatically
async def fetch_all_taskgroup():
results = []
async with asyncio.TaskGroup() as tg:
for url in urls:
task = tg.create_task(fetch(url))
results.append(task)
return [task.result() for task in results]# BAD: Blocking call freezes event loop
async def process_data_bad(data):
result = heavy_cpu_computation(data) # Blocks!
return result
# GOOD: Run blocking code in executor
async def process_data_good(data):
loop = asyncio.get_event_loop()
result = await loop.run_in_executor(None, heavy_cpu_computation, data)
return result# BAD: Using blocking libraries
import requests
async def fetch_bad(url):
return requests.get(url).json() # Blocks event loop!
# GOOD: Use async libraries
import aiohttp
async def fetch_good(url):
async with aiohttp.ClientSession() as session:
async with session.get(url) as response:
return await response.json()
# BAD: Blocking sleep
import time
async def delay_bad():
time.sleep(1) # Blocks!
# GOOD: Async sleep
async def delay_good():
await asyncio.sleep(1) # Yields to event loopasync def fetch_users_parallel(user_ids: List[int]) -> tuple[List[dict], List[Exception]]:
tasks = [fetch_user(uid) for uid in user_ids]
# gather with return_exceptions=True prevents one failure from canceling others
results = await asyncio.gather(*tasks, return_exceptions=True)
successes = [r for r in results if not isinstance(r, Exception)]
failures = [r for r in results if isinstance(r, Exception)]
return successes, failuresasync function fetchUsersParallel(userIds) {
const results = await Promise.allSettled(userIds.map(id => fetchUser(id)));
const successes = results.filter(r => r.status === 'fulfilled').map(r => r.value);
const failures = results.filter(r => r.status === 'rejected').map(r => r.reason);
return { successes, failures };
}async def fetch_with_timeout(url: str, timeout: float = 5.0) -> Optional[str]:
try:
async with asyncio.timeout(timeout): # Python 3.11+
return await fetch_data(url)
except asyncio.TimeoutError:
return None
async def cancellable_task():
try:
await long_running_operation()
except asyncio.CancelledError:
await cleanup()
raise # Re-raise to signal cancellationasync function fetchWithTimeout(url, timeoutMs = 5000) {
const controller = new AbortController();
const timeoutId = setTimeout(() => controller.abort(), timeoutMs);
try {
const response = await fetch(url, { signal: controller.signal });
clearTimeout(timeoutId);
return await response.json();
} catch (error) {
if (error.name === 'AbortError') return null;
throw error;
}
}async def retry_with_backoff(
func: Callable,
max_retries: int = 3,
base_delay: float = 1.0,
exponential_base: float = 2.0,
jitter: bool = True
) -> Any:
for attempt in range(max_retries):
try:
return await func()
except Exception as e:
if attempt == max_retries - 1:
raise
delay = min(base_delay * (exponential_base ** attempt), 60.0)
if jitter:
delay *= (0.5 + random.random())
await asyncio.sleep(delay)async function retryWithBackoff(fn, { maxRetries = 3, baseDelay = 1000 } = {}) {
for (let attempt = 0; attempt < maxRetries; attempt++) {
try {
return await fn();
} catch (error) {
if (attempt === maxRetries - 1) throw error;
const delay = Math.min(baseDelay * Math.pow(2, attempt), 60000);
await new Promise(r => setTimeout(r, delay));
}
}
}from contextlib import asynccontextmanager
@asynccontextmanager
async def get_db_connection(dsn: str):
conn = DatabaseConnection(dsn)
try:
await conn.connect()
yield conn
finally:
if conn.connected:
await conn.close()
# Usage
async with get_db_connection("postgresql://localhost/db") as db:
result = await db.execute("SELECT * FROM users")async function withConnection(dsn, callback) {
const conn = new DatabaseConnection(dsn);
try {
await conn.connect();
return await callback(conn);
} finally {
if (conn.connected) {
await conn.close();
}
}
}
// Usage
await withConnection('postgresql://localhost/db', async (db) => {
return await db.execute('SELECT * FROM users');
});# ❌ BAD: Returns coroutine object, not data
async def get_data():
result = fetch_data() # Missing await!
return result
# ✅ GOOD
async def get_data():
return await fetch_data()# ❌ BAD: Sequential execution - 3 seconds total
async def fetch_all():
user = await fetch_user()
posts = await fetch_posts()
comments = await fetch_comments()
# ✅ GOOD: Parallel execution - 1 second total
async def fetch_all():
return await asyncio.gather(
fetch_user(),
fetch_posts(),
fetch_comments()
)# ❌ BAD: Unbounded concurrency (10,000 simultaneous connections!)
async def process_all(items):
return await asyncio.gather(*[process_item(item) for item in items])
# ✅ GOOD: Limit concurrency with semaphore
async def process_all(items, max_concurrent=100):
semaphore = asyncio.Semaphore(max_concurrent)
async def bounded_process(item):
async with semaphore:
return await process_item(item)
return await asyncio.gather(*[bounded_process(item) for item in items])time.sleep()asyncio.sleep()pytest --asyncio-mode=autogrep -r "time\.sleep\|requests\." src/pytest --cov=app