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 loop

async 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, failures

async 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 cancellation