Gérer les échecs et les délais

Une consultation distante peut prendre trop de temps ou échouer alors que les autres répondent normalement. Sans stratégie explicite, un appel bloqué retarde votre outil de dépenses et une exception peut faire échouer toute la collecte. Vous allez borner l’attente de chaque appel et représenter les échecs sans perdre les résultats obtenus.

Vous avez déjà utilisé asyncio.gather pour lancer les consultations et un asyncio.Semaphore pour limiter leur nombre simultané. Ici, le sémaphore continue de protéger le service ; le délai maximal s’applique à la consultation elle-même, une fois le créneau obtenu.

Définir une réponse exploitable en cas d’échec

Une valeur absente ne suffit pas toujours à expliquer ce qui s’est passé. Le résultat ci-dessous conserve la devise, le taux lorsqu’il est disponible et un message d’erreur sinon. La fonction consulter_taux simule un service : elle réussit pour EUR, dépasse le délai pour USD et signale une erreur de service pour GBP.

import asyncio
from dataclasses import dataclass


@dataclass
class ResultatTaux:
    devise: str
    taux: float | None
    erreur: str | None


async def consulter_taux(devise: str) -> float:
    """Simule une consultation de service distant."""
    if devise == "USD":
        await asyncio.sleep(0.2)
        return 1.08
    if devise == "GBP":
        await asyncio.sleep(0.01)
        raise OSError("Service de taux indisponible")
    await asyncio.sleep(0.01)
    return 1.0


async def recuperer_taux(
    devises: list[str],
    limite: int = 2,
    delai: float = 0.05,
) -> list[ResultatTaux]:
    semaphore = asyncio.Semaphore(limite)

    async def recuperer_une(devise: str) -> ResultatTaux:
        try:
            async with semaphore:
                taux = await asyncio.wait_for(
                    consulter_taux(devise),
                    timeout=delai,
                )
        except TimeoutError:
            return ResultatTaux(devise, None, f"Délai dépassé ({delai} s)")
        except OSError as erreur:
            return ResultatTaux(devise, None, str(erreur))
        return ResultatTaux(devise, taux, None)

    return await asyncio.gather(*(recuperer_une(devise) for devise in devises))


async def main() -> None:
    resultats = await recuperer_taux(["EUR", "USD", "GBP"])
    for resultat in resultats:
        if resultat.erreur is None:
            print(f"{resultat.devise}: taux {resultat.taux}")
        else:
            print(f"{resultat.devise}: échec — {resultat.erreur}")


if __name__ == "__main__":
    asyncio.run(main())

EUR: taux 1.0
USD: échec — Délai dépassé (0.05 s)
GBP: échec — Service de taux indisponible

asyncio.wait_for attend au plus le nombre de secondes donné. Si ce délai est dépassé, il annule la coroutine surveillée et déclenche TimeoutError, que le code transforme ici en résultat explicite. L’erreur OSError est également convertie en résultat, tandis que les autres erreurs inattendues ne sont pas masquées.

Comprendre l’annulation et les exceptions

Une annulation n’est pas une erreur métier à convertir en résultat ordinaire. Lorsqu’un appelant annule la collecte, asyncio interrompt les coroutines concernées ; en ne capturant ni asyncio.CancelledError ni toutes les exceptions avec un except Exception, vous permettez à cette interruption de se propager. C’est important pour arrêter proprement un traitement devenu inutile.

Chaque tâche de collecte attrape uniquement les deux situations prévues : le délai dépassé et l’erreur de service simulée par OSError. Ainsi, ces échecs sont isolés et les autres devises peuvent aboutir. Toute exception non prévue remonte au niveau de gather au lieu d’être dissimulée comme une simple absence de taux.

Les résultats de gather restent dans l’ordre des devises transmises, même si les appels finissent dans un ordre différent. Vous pouvez donc relier chaque réponse à sa devise sans réorganiser les résultats. Dans une intégration réelle, remplacez la simulation par votre client réseau et adaptez les erreurs capturées aux exceptions documentées par ce client.

À retenir

• Utilisez asyncio.wait_for pour fixer un délai maximal à une consultation.
• Convertissez les erreurs attendues en résultats explicites pour préserver les réponses disponibles.
• Ne masquez pas les exceptions inconnues et laissez les annulations se propager.
• Précisez si le délai couvre uniquement l’appel distant ou aussi l’attente d’un créneau de concurrence.

Vous pourrez ensuite mesurer les performances du traitement avec une méthode reproductible avant toute optimisation.