graph-api-quality-auditor

Graph API Quality Auditor

Skill d'audit et d'amélioration du code Python appelant Microsoft Graph API, couvrant quatre axes : sécurité/auth, robustesse (erreurs & throttling), performance, et qualité du code.

Workflow obligatoire

Phase 1 — Audit : produire un rapport structuré AVANT toute correction.
Phase 2 — Amélioration : proposer du code corrigé, section par section.
Ne pas mélanger les deux phases. Ne jamais réécrire en bloc sans rapport préalable.

---

Phase 1 : Rapport d'audit

Analyser le code soumis selon les quatre axes ci-dessous. Pour chaque problème trouvé, assigner :

• 🔴 CRITIQUE — faille de sécurité, bug qui plante en production
• 🟠 MAJEUR — comportement incorrect sous charge, perte de données possible
• 🟡 MINEUR — sous-optimal, fragilité future
• 🔵 SUGGESTION — amélioration de lisibilité ou maintenabilité

Format du rapport

## Rapport d'audit Graph API

### Résumé
- X problèmes critiques, Y majeurs, Z mineurs, W suggestions
- Contexte détecté : [script ponctuel | backend web | déclenché par Power Automate]

### Axe 1 : Authentification & Sécurité
[liste des problèmes trouvés avec sévérité et ligne si possible]

### Axe 2 : Gestion des erreurs & Throttling
[...]

### Axe 3 : Performance & Utilisation de l'API
[...]

### Axe 4 : Qualité du code
[...]

### Verdict global
[Phrase de synthèse + priorités d'action]

---

Axe 1 : Authentification & Sécurité

Checks à effectuer

Flux d'authentification

• Utilise-t-on msal (recommandé) ou requests brut avec un token hardcodé ?
• Delegated flow (utilisateur) vs Application flow (service) — est-ce cohérent avec le contexte ?
• PublicClientApplication vs ConfidentialClientApplication — bon choix pour le contexte ?
• Pour Power Automate / scripts serveur : utiliser acquire_token_for_client (client credentials)
• Pour apps au nom d'un utilisateur : PKCE ou device flow, jamais password grant

Secrets & configuration

• Client secret / certificat dans le code source → 🔴 CRITIQUE
• Secret dans un fichier .env commité → 🔴 CRITIQUE
• Absence de chargement via os.environ ou Azure Key Vault → 🟠 MAJEUR
• tenant_id, client_id hardcodés (pas secrets, mais non configurables) → 🟡 MINEUR

Scopes

• Scopes trop larges (Mail.ReadWrite au lieu de Mail.Read) → 🔴 CRITIQUE
• Utilisation de .default sans commentaire expliquant pourquoi → 🟡 MINEUR
• Scopes application vs delegated — cohérence avec le flux → 🟠 MAJEUR

Cache de tokens

• Absence de cache → token renouvelé à chaque appel → 🟠 MAJEUR (throttling + latence)
• SerializableTokenCache non persisté entre les runs (scripts longs) → 🟡 MINEUR
• En backend web : cache partagé entre threads sans verrou → 🔴 CRITIQUE

Patterns recommandés

# ✅ Auth avec cache — script ponctuel ou Azure Function
import msal, os, json

def get_token(cache_file: str | None = None) -> str:
    cache = msal.SerializableTokenCache()
    if cache_file and os.path.exists(cache_file):
        cache.deserialize(open(cache_file).read())

    app = msal.ConfidentialClientApplication(
        client_id=os.environ["GRAPH_CLIENT_ID"],
        client_credential=os.environ["GRAPH_CLIENT_SECRET"],
        authority=f"https://login.microsoftonline.com/{os.environ['GRAPH_TENANT_ID']}",
        token_cache=cache,
    )
    result = app.acquire_token_silent(
        scopes=["https://graph.microsoft.com/.default"], account=None
    )
    if not result:
        result = app.acquire_token_for_client(
            scopes=["https://graph.microsoft.com/.default"]
        )
    if "access_token" not in result:
        raise RuntimeError(f"Échec auth MSAL : {result.get('error_description')}")

    if cache_file and cache.has_state_changed:
        open(cache_file, "w").write(cache.serialize())

    return result["access_token"]

---

Axe 2 : Gestion des erreurs & Throttling

Checks à effectuer

Throttling (erreurs 429 / 503)

• Absence de retry sur 429 → crash en production sous charge → 🔴 CRITIQUE
• Retry sans respecter l'en-tête Retry-After → aggrave le throttling → 🔴 CRITIQUE
• Retry illimité ou sans backoff exponentiel → 🟠 MAJEUR

Gestion des codes d'erreur Graph

• raise_for_status() seul — perd le corps JSON de l'erreur Graph → 🟠 MAJEUR
• Aucune distinction entre erreurs retryables (429, 503, 504) et définitives (403, 404) → 🟠 MAJEUR
• Erreurs de token expiration (401) non gérées avec re-tentative d'auth → 🟠 MAJEUR

Logging

• Aucun logging des erreurs → impossible à déboguer en production → 🟠 MAJEUR
• print() au lieu de logging → 🟡 MINEUR

Pattern recommandé : client Graph avec retry

import time, logging, requests
from typing import Any

logger = logging.getLogger(__name__)

GRAPH_BASE = "https://graph.microsoft.com/v1.0"

RETRYABLE_STATUS = {429, 500, 502, 503, 504}

def graph_request(
    method: str,
    endpoint: str,
    token: str,
    max_retries: int = 5,
    **kwargs,
) -> Any:
    """Appel Graph API avec retry automatique sur throttling."""
    url = f"{GRAPH_BASE}{endpoint}" if not endpoint.startswith("http") else endpoint
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json",
        **kwargs.pop("headers", {}),
    }
    for attempt in range(max_retries):
        resp = requests.request(method, url, headers=headers, **kwargs)

        if resp.status_code in RETRYABLE_STATUS:
            wait = int(resp.headers.get("Retry-After", 2 ** attempt))
            logger.warning(
                "Graph API %s %s → %s, retry dans %ss (tentative %s/%s)",
                method, endpoint, resp.status_code, wait, attempt + 1, max_retries,
            )
            time.sleep(wait)
            continue

        if not resp.ok:
            err = resp.json().get("error", {})
            logger.error(
                "Graph API erreur %s : %s — %s",
                resp.status_code, err.get("code"), err.get("message"),
            )
            resp.raise_for_status()

        return resp.json() if resp.content else None

    raise RuntimeError(f"Graph API : max retries atteint pour {method} {endpoint}")

---

Axe 3 : Performance & Utilisation de l'API

Checks à effectuer

Sélection des champs ($select)

• Requêtes sans $select → récupère 40+ champs inutiles → 🟡 MINEUR à 🟠 MAJEUR (volume)
• $select manquant sur les listes d'utilisateurs ou messages → 🟠 MAJEUR

Filtrage côté serveur ($filter)

• Filtre appliqué en Python après récupération de toutes les pages → 🔴 sur grands volumes
• Utilisation de $filter sur des propriétés non indexées sans ConsistencyLevel: eventual → erreur 400

Pagination

• Traitement de la première page seulement, @odata.nextLink ignoré → 🔴 CRITIQUE (données manquantes)
• Boucle de pagination incorrecte (condition d'arrêt manquante) → 🔴 CRITIQUE

Batching

• Multiples appels séquentiels indépendants (>4) qui pourraient être groupés en /$batch → 🟠 MAJEUR

Patterns recommandés

# ✅ Pagination complète
def get_all_pages(token: str, endpoint: str, params: dict | None = None) -> list:
    results = []
    url = endpoint
    while url:
        data = graph_request("GET", url, token, params=params)
        results.extend(data.get("value", []))
        url = data.get("@odata.nextLink")  # None stoppe la boucle
        params = None  # params déjà encodés dans nextLink
    return results

# ✅ Exemple d'appel optimisé
users = get_all_pages(
    token,
    "/users",
    params={"$select": "id,displayName,mail,userPrincipalName", "$top": 999},
)

# ✅ Filtre avancé (requiert ConsistencyLevel)
members = graph_request(
    "GET",
    "/users",
    token,
    headers={"ConsistencyLevel": "eventual"},
    params={
        "$filter": "department eq 'Engineering'",
        "$select": "id,displayName",
        "$count": "true",
    },
)

Batching ($batch) — pour 2 à 20 requêtes indépendantes

def graph_batch(token: str, requests_list: list[dict]) -> list:
    """
    requests_list : liste de dicts {"id": "1", "method": "GET", "url": "/users/xxx"}
    Limite : 20 requêtes par batch.
    """
    payload = {"requests": requests_list}
    result = graph_request("POST", "/$batch", token, json=payload)
    return sorted(result["responses"], key=lambda r: r["id"])

---

Axe 4 : Qualité du code

Checks à effectuer

Structure

• Logique d'auth mélangée avec logique métier → 🟡 MINEUR
• Client HTTP instancié dans chaque fonction (pas de session partagée) → 🟡 MINEUR
• Pas de séparation config / client / logique métier → 🟡 MINEUR

Type hints & contrats

• Absence de type hints sur les fonctions publiques → 🔵 SUGGESTION
• Retour dict non typé quand la structure est connue (TypedDict ou dataclass) → 🔵 SUGGESTION

Constantes & configuration

• URLs Graph hardcodées ("https://graph.microsoft.com/v1.0/...") répétées → 🟡 MINEUR
• Valeurs magiques (timeouts, tailles de page) sans constante nommée → 🟡 MINEUR

Testabilité

• Appels HTTP non mockables (pas d'injection de client) → 🟡 MINEUR
• Absence de tests unitaires sur la logique de retry/pagination → 🔵 SUGGESTION

Structure recommandée pour un module Graph

graph_client/
├── __init__.py
├── auth.py          # get_token(), cache management
├── client.py        # graph_request(), graph_batch(), pagination
├── config.py        # GraphConfig dataclass, chargement depuis env
└── services/
    ├── users.py     # get_user(), list_users(), etc.
    └── calendar.py  # list_events(), create_event(), etc.

---

Contextes spécifiques

Script ponctuel / automatisation CNESST-style

• Prioriser : cache de token sur disque, retry 429, logs structurés
• Acceptable : pas de batching si <5 appels, structure simple
• Éviter : secrets en dur, absence de retry, pagination ignorée

Backend web / API Python (FastAPI, Flask)

• Prioriser : cache de token thread-safe (threading.Lock), session requests partagée, async si FastAPI
• Acceptable : httpx.AsyncClient à la place de requests
• Éviter : instanciation MSAL à chaque requête, blocage du thread principal

Déclenché par Power Automate (HTTP action → Azure Function Python)

• Prioriser : validation du token entrant si delegated, gestion des timeouts courts (PA timeout = 2 min)
• Acceptable : client credentials si le flow PA appelle une fonction de service
• Éviter : attentes longues dans la fonction (utiliser Durable Functions pour les longues opérations)

---

Phase 2 : Présentation des corrections

Après le rapport, proposer les corrections dans cet ordre de priorité :

1. 🔴 CRITIQUE — corrections immédiates, toujours proposées
2. 🟠 MAJEUR — corrections recommandées, proposées par défaut
3. 🟡 MINEUR / 🔵 SUGGESTION — proposer en bloc, laisser le choix à l'utilisateur

Format de chaque correction :

### Correction : [titre]
**Problème** : [description courte]
**Avant** :
```python
# code problématique

Après :

# code corrigé

Pourquoi : [explication concise]

Terminer avec une synthèse : « Ce qui fonctionne bien » + « Prochaine étape recommandée ».