async-check

Vérification Patterns Async Python

Tu es un expert Python asyncio. Tu dois vérifier l'utilisation correcte des patterns async/await, identifier les anti-patterns et les problèmes de concurrence.

Arguments

$ARGUMENTS

Arguments :

• (Optionnel) Chemin vers un module spécifique

Exemple : /python:async-check app/services/

MISSION

Étape 1 : Analyser le Code Async

# Rechercher les fonctions async
grep -rn "async def" app/ --include="*.py"

# Rechercher les await
grep -rn "await " app/ --include="*.py"

# Rechercher asyncio.run (potentiel problème)
grep -rn "asyncio.run" app/ --include="*.py"

# Rechercher les .result() sur futures (blocking)
grep -rn "\.result()" app/ --include="*.py"

Étape 2 : Anti-Patterns Courants

1. Blocking dans Async

# ❌ MAUVAIS - Bloque l'event loop
async def get_data():
    response = requests.get("https://api.example.com")  # BLOCKING!
    return response.json()

# ✅ BON - Non-bloquant
async def get_data():
    async with httpx.AsyncClient() as client:
        response = await client.get("https://api.example.com")
        return response.json()

2. Sync Sleep dans Async

# ❌ MAUVAIS - Bloque l'event loop
async def slow_task():
    time.sleep(5)  # BLOCKING!
    return "done"

# ✅ BON - Non-bloquant
async def slow_task():
    await asyncio.sleep(5)
    return "done"

3. Await dans une Boucle (N+1)

# ❌ MAUVAIS - Séquentiel, lent
async def fetch_all_users(user_ids: list[int]) -> list[User]:
    users = []
    for user_id in user_ids:
        user = await fetch_user(user_id)  # N requêtes séquentielles
        users.append(user)
    return users

# ✅ BON - Concurrent
async def fetch_all_users(user_ids: list[int]) -> list[User]:
    tasks = [fetch_user(user_id) for user_id in user_ids]
    users = await asyncio.gather(*tasks)
    return list(users)

# ✅ ENCORE MIEUX - Avec limite de concurrence
async def fetch_all_users(user_ids: list[int], max_concurrent: int = 10) -> list[User]:
    semaphore = asyncio.Semaphore(max_concurrent)

    async def fetch_with_limit(user_id: int) -> User:
        async with semaphore:
            return await fetch_user(user_id)

    tasks = [fetch_with_limit(user_id) for user_id in user_ids]
    users = await asyncio.gather(*tasks)
    return list(users)