backend / python

Asyncio avancé : async context managers et générateurs

Leçon 221 exercice

Explication

Ce que vous allez apprendre

Implémenter __aenter__/__aexit__ pour un context manager dont l'ouverture nécessite d'attendre
Écrire un générateur asynchrone avec async def + yield et le consommer avec async for
Gérer proprement l'annulation d'une tâche asyncio avec CancelledError
Appliquer un pattern de retry avec backoff exponentiel pour un appel réseau instable
Protéger une opération critique d'une annulation externe avec asyncio.shield

Dans quel contexte ?

Un développeur maintient un service qui interroge une API tierce connue pour renvoyer occasionnellement des erreurs 503 temporaires. Sans stratégie de retry, chaque erreur transitoire fait échouer toute la requête utilisateur. Cette leçon montre comment implémenter appeler_avec_retry(), qui retente l'appel jusqu'à trois fois avec un délai croissant (backoff exponentiel), transformant une panne passagère en un simple ralentissement invisible pour l'utilisateur final.

Étendre les protocoles déjà connus au monde asynchrone

Cette leçon prolonge directement la précédente : les context managers et les générateurs, déjà vus dans leur version synchrone, existent aussi sous une forme asynchrone. __aenter__/__aexit__ remplacent __enter__/__exit__ quand l'ouverture ou la fermeture d'une ressource nécessite elle-même une opération d'attente, comme une vraie connexion réseau asynchrone.

Un flux de données qui arrive au fil du temps

Un générateur asynchrone (async def combiné à yield) modélise naturellement une source de données qui arrive progressivement, comme un flux réseau. On le consomme avec async for, qui laisse la possibilité à l'event loop de traiter d'autres tâches entre chaque valeur reçue, contrairement à un générateur classique qui bloquerait tout le programme le temps de produire chaque valeur.

Version synchrone	Équivalent asynchrone
`__enter__`/`__exit__`	`__aenter__`/`__aexit__`
`for x in y:`	`async for x in y:`
`[x for x in y]`	`[x async for x in y]`
`@contextmanager`	`@asynccontextmanager`

Annuler proprement, pas brutalement

Une tâche asynchrone peut être annulée à tout moment depuis l'extérieur. Cette annulation ne tue pas instantanément la coroutine : elle lève une exception spéciale, CancelledError, au prochain point d'attente rencontré. C'est l'occasion de nettoyer proprement des ressources dans un finally, mais la règle stricte est de toujours relaisser cette exception se propager après le nettoyage, plutôt que de l'avaler silencieusement.

Des patterns de robustesse pour la production

Le retry avec backoff exponentiel (attendre de plus en plus longtemps entre chaque tentative) et asyncio.shield (protéger une opération critique d'une annulation externe) sont deux techniques qu'on retrouve constamment dans du code réseau réel, où les échecs transitoires et les annulations intempestives sont la norme plutôt que l'exception.

Bonne pratique

Dans un except asyncio.CancelledError:, effectuez uniquement le nettoyage strictement nécessaire (fermer un fichier, libérer un verrou) puis relancez toujours l'exception avec raise. L'avaler silencieusement empêche l'appelant de savoir que la tâche a réellement été annulée, ce qui peut bloquer l'arrêt propre de tout le programme.

Commandes & code

Asyncio avancé

Context managers et générateurs asynchrones, patterns de production.

import asyncio

# --- Async context managers : __aenter__ / __aexit__ ---
class ConnexionAsync:
    def __init__(self, dsn):
        self.dsn = dsn

    async def __aenter__(self):
        print(f"Ouverture asynchrone de la connexion vers {self.dsn}")
        await asyncio.sleep(0.1)      # simule une connexion reseau
        return self

    async def __aexit__(self, exc_type, exc_value, traceback):
        print("Fermeture asynchrone de la connexion")
        await asyncio.sleep(0.05)
        return False

async def utiliser_connexion():
    async with ConnexionAsync("postgres://async-host") as conn:
        print(f"Utilisation de {conn.dsn}")

asyncio.run(utiliser_connexion())

# contextlib.asynccontextmanager : version simplifiee avec un generateur
from contextlib import asynccontextmanager

@asynccontextmanager
async def session_http(url):
    print(f"Ouverture de session vers {url}")
    session = {"url": url, "ouverte": True}
    try:
        yield session
    finally:
        session["ouverte"] = False
        print("Session fermee")

async def main_session():
    async with session_http("https://api.example.com") as session:
        print(session)

asyncio.run(main_session())

# --- Generateurs asynchrones : async def + yield ---
async def flux_de_donnees(n):
    for i in range(n):
        await asyncio.sleep(0.1)      # simule une source de donnees en streaming
        yield i

async def consommer_flux():
    async for valeur in flux_de_donnees(5):
        print(f"Recu : {valeur}")

asyncio.run(consommer_flux())

# Async comprehension : consomme un generateur asynchrone
async def main_comprehension():
    resultats = [v async for v in flux_de_donnees(3) if v % 2 == 0]
    print(resultats)

asyncio.run(main_comprehension())

# --- Pattern production : client HTTP asynchrone avec retry et backoff ---
async def appeler_avec_retry(operation, tentatives=3, delai_initial=0.5):
    derniere_erreur = None
    for tentative in range(tentatives):
        try:
            return await operation()
        except Exception as e:
            derniere_erreur = e
            attente = delai_initial * (2 ** tentative)     # backoff exponentiel
            print(f"Tentative {tentative + 1} echouee, retry dans {attente}s")
            await asyncio.sleep(attente)
    raise derniere_erreur

async def operation_instable():
    import random
    if random.random() < 0.7:
        raise ConnectionError("Echec simule")
    return "succes"

# --- Annulation cooperative de taches ---
async def tache_longue():
    try:
        await asyncio.sleep(10)
    except asyncio.CancelledError:
        print("Tache annulee proprement, nettoyage en cours")
        raise         # toujours re-lever CancelledError apres nettoyage

async def main_annulation():
    tache = asyncio.create_task(tache_longue())
    await asyncio.sleep(0.5)
    tache.cancel()
    try:
        await tache
    except asyncio.CancelledError:
        print("Confirmation : la tache a bien ete annulee")

asyncio.run(main_annulation())

# shield : proteger une tache critique d'une annulation externe
async def operation_critique():
    await asyncio.sleep(2)
    return "donnees sauvegardees"

async def main_shield():
    try:
        resultat = await asyncio.wait_for(
            asyncio.shield(operation_critique()), timeout=1
        )
    except asyncio.TimeoutError:
        print("Timeout cote appelant, mais l'operation critique continue en arriere-plan")

Résumé

__aenter__/__aexit__ (ou @asynccontextmanager) définissent des context managers asynchrones.
async def + yield crée un générateur asynchrone, consommé avec async for.
Une tâche annulée reçoit CancelledError au point de await : il faut toujours la re-lever après nettoyage.
asyncio.shield protège une opération critique d'une annulation externe.

Exercices pratiques

1 disponible

Mission : sauver une sauvegarde annulée en plein milieu

Objectif : Corriger une annulation de tâche qui avale silencieusement CancelledError, et concevoir un retry avec backoff exponentiel pour un appel réseau instable.

Contexte

Un service critique de sauvegarde intercepte asyncio.CancelledError dans sa coroutine tache_longue() mais ne relève jamais l'exception après son nettoyage, se contentant d'un print. Depuis ce changement, appeler tache.cancel() puis await tache ne lève plus jamais d'erreur côté appelant, qui croit à tort que tout s'est bien passé alors que la tâche a été interrompue en cours de route.

Tu dois corriger cette annulation silencieuse, puis concevoir un pattern de retry avec délai croissant pour un appel réseau qui échoue une fois sur deux.

Résoudre l’exercice →