Python : asyncio avancé (Task, TaskGroup, Queue)

Ce guide reprend asyncio là où l'introduction s'arrête : le cycle de vie précis des Task, la concurrence structurée avec TaskGroup, les timeouts, la file Queue, l'annulation propre et l'intégration de code bloquant. Il s'adresse aux développeurs qui écrivent déjà des coroutines et veulent des motifs robustes pour la production : lancer des centaines de tâches sans en perdre une, annuler proprement, borner le temps d'une opération, et éviter les erreurs qui gèlent la boucle. Chaque section montre un cas concret. Les exemples ont été exécutés avec Python 3.12.

Vous êtes à l'aise avec async/await et asyncio.run ? C'est le prérequis. Si ces bases sont floues, commencez par le guide asyncio en Python avant de revenir ici.

Ce que vous allez apprendre

• Gérer le cycle de vie complet d'une Task créée avec create_task.
• Choisir entre gather et TaskGroup selon la gestion d'erreur voulue.
• Borner la durée d'une opération avec asyncio.timeout.
• Construire un producteur/consommateur avec asyncio.Queue.
• Annuler une tâche et libérer ses ressources sans fuite.
• Déporter du code bloquant hors de la boucle avec run_in_executor.

La boucle d'événements et les coroutines en profondeur

Sous asyncio.run, un unique objet orchestre tout : la boucle d'événements (event loop). Elle maintient une file de coroutines prêtes à progresser et les fait avancer une par une. Chaque await sur une opération lente rend la main à la boucle, qui en profite pour faire avancer une autre coroutine. Rien ne s'exécute en parallèle au sens des threads : tout tourne sur un seul fil, la boucle passant d'une tâche à l'autre aux points d'await.

Une coroutine n'est qu'un objet décrivant un travail suspendable. L'appeler ne l'exécute pas : c'est la boucle qui la pilote. Ce point est central pour la suite : c'est la boucle qui décide quand reprendre une coroutine, et c'est pourquoi un appel bloquant au mauvais endroit paralyse tout le programme.

import asyncio

async def travail(nom, duree):
    print(f"  {nom} demarre")
    await asyncio.sleep(duree)
    print(f"  {nom} termine apres {duree}s")
    return nom.upper()

async def main():
    resultat = await travail("telechargement", 0.2)
    print(f"  recu: {resultat}")

asyncio.run(main())

L'exécution montre la coroutine qui démarre, cède la main pendant l'attente, puis renvoie sa valeur :

  telechargement demarre
  telechargement termine apres 0.2s
  recu: TELECHARGEMENT

create_task et le cycle de vie d'une Task

await sur une coroutine l'exécute immédiatement et attend son résultat : rien ne se chevauche. Pour lancer un travail en arrière-plan et continuer, on l'emballe dans une Task avec asyncio.create_task. La tâche est aussitôt planifiée sur la boucle et commence à progresser dès le prochain await, sans qu'on l'attende explicitement.

Une Task passe par des états observables : done() indique si elle est terminée, result() renvoie sa valeur (ou relance son exception), cancel() demande son annulation. Tant que la boucle ne reprend pas la main via un await, la tâche ne progresse pas, même créée.

async def main():
    tache = asyncio.create_task(travail("tache-A", 0.2))
    print(f"  cree, done={tache.done()}")
    await asyncio.sleep(0.3)
    print(f"  apres attente, done={tache.done()}")
    print(f"  resultat={tache.result()}")

asyncio.run(main())

La sortie confirme que la tâche démarre pendant le sleep du main, puis livre son résultat :

  cree, done=False
  tache-A demarre
  tache-A termine apres 0.2s
  apres attente, done=True
  resultat=TACHE-A

Gardez une référence à vos Task

La boucle ne conserve qu'une référence faible vers les tâches créées. Si vous ne stockez pas l'objet renvoyé par create_task, le ramasse-miettes peut le supprimer en plein vol et la tâche disparaît sans terminer. Conservez toujours la référence, dans une variable ou une liste, jusqu'à la fin de la tâche.

gather : lancer en masse et gérer les erreurs

asyncio.gather prend plusieurs coroutines, les lance en concurrence et renvoie leurs résultats dans l'ordre des arguments, pas dans l'ordre de fin. C'est l'outil du quotidien pour paralléliser des appels d'API ou des requêtes indépendantes. Par défaut, la première exception annule l'attente et remonte immédiatement, alors que les autres tâches, elles, continuent en arrière-plan.

Le paramètre return_exceptions=True change ce comportement : au lieu de propager, gather capture chaque exception et la place dans la liste de résultats, à la position de la tâche fautive. On récupère alors un mélange de valeurs et d'objets exception, et aucune tâche n'est interrompue par l'échec d'une autre.

async def peut_echouer(n):
    await asyncio.sleep(0.1)
    if n == 2:
        raise ValueError(f"echec sur {n}")
    return n * 10

async def main():
    res = await asyncio.gather(
        peut_echouer(1), peut_echouer(2), peut_echouer(3),
        return_exceptions=True,
    )
    print(f"  resultats: {res}")

asyncio.run(main())

L'exception est rangée à sa place au lieu de faire tout échouer :

  resultats: [10, ValueError('echec sur 2'), 30]

Ce mode est pratique pour un traitement par lot où l'échec d'un élément ne doit pas condamner les autres. Attention toutefois : une exception rangée dans la liste n'est pas relancée automatiquement. À vous d'inspecter les résultats avec isinstance(r, Exception) pour ne pas avaler une erreur silencieusement.

TaskGroup : la concurrence structurée (Python 3.11+)

asyncio.TaskGroup, arrivé en Python 3.11, applique le principe de concurrence structurée : toutes les tâches lancées dans le bloc async with doivent se terminer avant que le bloc ne se ferme. On crée les tâches avec tg.create_task, et la sortie du async with attend automatiquement la fin de toutes. Plus besoin de collecter les tâches à la main ni d'oublier un await.

Sa vraie force est la gestion d'erreur. Si une tâche du groupe échoue, TaskGroup annule toutes les autres puis propage l'erreur, regroupée dans un ExceptionGroup que l'on filtre avec la syntaxe except*. Fini le cas où une tâche plante pendant que les autres continuent à consommer des ressources dans le vide.

async def main():
    async with asyncio.TaskGroup() as tg:
        tg.create_task(travail("import-clients", 0.3))
        tg.create_task(travail("import-commandes", 0.1))
        tg.create_task(travail("import-stocks", 0.2))
    print("  les trois imports sont termines")

asyncio.run(main())

Les tâches démarrent dans l'ordre de création, se terminent dans l'ordre de leur durée, et le bloc n'avance qu'une fois toutes finies :

  import-clients demarre
  import-commandes demarre
  import-stocks demarre
  import-commandes termine apres 0.1s
  import-stocks termine apres 0.2s
  import-clients termine apres 0.3s
  les trois imports sont termines

Quand une tâche lève une exception, le groupe annule les autres et regroupe le tout :

async def main():
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(travail("ok", 0.3))
            tg.create_task(peut_echouer(2))
    except* ValueError as eg:
        print(f"  capture ExceptionGroup: {eg.exceptions}")

asyncio.run(main())

La tâche ok démarre puis se fait annuler dès que l'autre échoue, et l'erreur arrive empaquetée :

  ok demarre
  capture ExceptionGroup: (ValueError('echec sur 2'),)

TaskGroup par défaut, gather pour la tolérance

Sur du code moderne (3.11+), TaskGroup est le choix par défaut : il garantit qu'aucune tâche ne survit au bloc et propage les erreurs proprement. Réservez gather(..., return_exceptions=True) aux cas où vous voulez explicitement tolérer les échecs partiels et inspecter chaque résultat un par un.

Borner le temps avec asyncio.timeout (3.11+)

Une opération réseau qui traîne peut bloquer tout un traitement. Le gestionnaire de contexte asyncio.timeout (Python 3.11) impose une durée maximale à un bloc async with. Passé le délai, il annule ce qui s'exécute à l'intérieur et lève un TimeoutError. C'est plus lisible que l'ancien asyncio.wait_for, et ça borne un groupe d'opérations d'un coup, pas une seule coroutine.

L'annulation déclenchée par le timeout est une vraie annulation : la coroutine interrompue reçoit un CancelledError en interne, ce qui lui laisse la possibilité de nettoyer ses ressources avant de disparaître. Le TimeoutError remonte, lui, au niveau du async with.

async def main():
    try:
        async with asyncio.timeout(0.2):
            await travail("lente", 1.0)
    except TimeoutError:
        print("  operation interrompue par timeout")

asyncio.run(main())

La tâche démarre mais n'a pas le temps de finir : le délai de 0,2 s la coupe net.

  lente demarre
  operation interrompue par timeout

asyncio.Queue : le motif producteur/consommateur

Quand des tâches doivent se transmettre du travail, la file asyncio.Queue est l'outil idéal. Un ou plusieurs producteurs y déposent des éléments avec await file.put(x), un ou plusieurs consommateurs les récupèrent avec await file.get(). La file régule le débit : avec maxsize, un producteur trop rapide se met en attente dès que la file est pleine, ce qui évite de saturer la mémoire (mécanisme de contre-pression).

Le motif classique consiste à envoyer une valeur sentinelle (None ici) pour signaler au consommateur qu'il n'y a plus rien à traiter. En production, on préfère souvent file.join() couplé à task_done() pour attendre que tout soit consommé, sans sentinelle.

async def producteur(file, n):
    for i in range(n):
        await asyncio.sleep(0.05)
        await file.put(i)
        print(f"  produit {i}")
    await file.put(None)

async def consommateur(file):
    while True:
        item = await file.get()
        if item is None:
            file.task_done()
            break
        print(f"  consomme {item}")
        file.task_done()

async def main():
    file = asyncio.Queue(maxsize=2)
    await asyncio.gather(producteur(file, 3), consommateur(file))

asyncio.run(main())

Production et consommation s'entrelacent, la file jouant le tampon entre les deux :

  produit 0
  consomme 0
  produit 1
  consomme 1
  produit 2
  consomme 2

Annulation et nettoyage : CancelledError

Annuler une tâche n'est pas la tuer brutalement. tache.cancel() demande une annulation : au prochain await, la coroutine reçoit un asyncio.CancelledError injecté à l'endroit où elle était suspendue. Ce mécanisme laisse une fenêtre pour libérer proprement les ressources, fermer une connexion, écrire un état, avant de disparaître.

La règle d'or : on intercepte CancelledError pour nettoyer, puis on le relance avec raise. Avaler cette exception sans la relancer empêche la boucle de considérer la tâche comme réellement annulée et casse la logique d'annulation en cascade de TaskGroup et timeout.

async def tache_longue():
    try:
        print("  travail en cours...")
        await asyncio.sleep(10)
    except asyncio.CancelledError:
        print("  nettoyage: fermeture des ressources")
        raise

async def main():
    t = asyncio.create_task(tache_longue())
    await asyncio.sleep(0.1)
    t.cancel()
    try:
        await t
    except asyncio.CancelledError:
        print("  tache bien annulee")

asyncio.run(main())

Le bloc de nettoyage s'exécute bien avant que la tâche ne soit déclarée annulée :

  travail en cours...
  nettoyage: fermeture des ressources
  tache bien annulee

Ne bâillonnez jamais CancelledError

Attraper CancelledError pour l'ignorer (sans raise) transforme une tâche annulée en tâche immortelle. Elle continue de tourner alors que tout le reste s'arrête, et un TaskGroup ou un timeout reste bloqué à l'attendre indéfiniment. Nettoyez, puis relancez systématiquement.

run_in_executor : intégrer du code bloquant

Certaines opérations n'ont pas de variante asynchrone : lecture disque lourde, calcul intensif, bibliothèque synchrone historique. Les appeler directement dans une coroutine gèle la boucle. La parade consiste à les exécuter dans un pool de threads via loop.run_in_executor, ou plus simplement asyncio.to_thread (Python 3.9+), qui rend la main à la boucle pendant que le thread travaille.

Ces deux fonctions renvoient un objet attendable : on peut donc les combiner avec gather ou TaskGroup comme n'importe quelle coroutine. Le code bloquant tourne hors de la boucle, sur un thread séparé, et la boucle reste libre de faire progresser le reste.

import time

def calcul_bloquant(n):
    time.sleep(0.2)
    return sum(range(n))

async def main():
    loop = asyncio.get_running_loop()
    r1, r2 = await asyncio.gather(
        loop.run_in_executor(None, calcul_bloquant, 1000),
        asyncio.to_thread(calcul_bloquant, 2000),
    )
    print(f"  resultats: {r1}, {r2}")

asyncio.run(main())

Les deux calculs bloquants tournent en parallèle sur des threads, sans figer la boucle :

  resultats: 499500, 1999000

Threads pour l'I/O bloquante, pas pour le CPU

run_in_executor avec le pool de threads par défaut convient au code bloquant sur des I/O (fichiers, bibliothèques réseau synchrones). Pour du calcul pur limité par le processeur, le verrou global (GIL) empêche le gain : passez un ProcessPoolExecutor en premier argument pour déporter le travail sur des processus séparés.

httpx et aiohttp : l'I/O réseau asynchrone

asyncio.sleep simule l'attente, mais en vrai le gain vient des appels réseau. La bibliothèque standard n'offre pas de client HTTP asynchrone ; on se tourne vers httpx ou aiohttp. httpx propose une API très proche de requests, avec un client asynchrone via async with httpx.AsyncClient(), ce qui rend la transition douce. aiohttp est le vétéran, à la fois client et serveur, très répandu dans les projets asynchrones existants.

L'exemple ci-dessous (non exécuté ici, il dépend du réseau et d'une installation pip install httpx) montre le motif : un client partagé réutilise les connexions, et gather lance toutes les requêtes en concurrence. C'est là que l'asynchrone paie vraiment : cent requêtes qui prennent chacune 200 ms se chevauchent au lieu de s'additionner.

import asyncio
import httpx

async def recuperer(client, url):
    reponse = await client.get(url)
    return url, reponse.status_code

async def main():
    urls = [
        "https://jsonplaceholder.typicode.com/users",
        "https://jsonplaceholder.typicode.com/posts",
        "https://jsonplaceholder.typicode.com/todos",
    ]
    async with httpx.AsyncClient(timeout=10.0) as client:
        resultats = await asyncio.gather(
            *(recuperer(client, u) for u in urls)
        )
    for url, code in resultats:
        print(f"{code} <- {url}")

asyncio.run(main())

Un seul client, réutilisé

Créez le client une fois et partagez-le entre toutes les requêtes via le async with. Ouvrir un AsyncClient par requête relance à chaque fois la négociation de connexion et annule le bénéfice du pool de connexions. Le même conseil vaut pour aiohttp.ClientSession.

Les pièges à éviter

L'asynchrone a des angles morts qui piègent même les développeurs aguerris. Le premier, le plus fréquent : bloquer la boucle. Un appel synchrone comme time.sleep, une requête requests.get ou un gros calcul dans une coroutine gèle tout le programme, car la boucle ne peut pas passer à une autre tâche pendant ce temps. Le code suivant le prouve : un time.sleep(0.2) empêche les autres coroutines d'avancer.

import time

async def main():
    debut = time.perf_counter()
    async def tic():
        for _ in range(3):
            await asyncio.sleep(0.05)
            print(f"    tic a {time.perf_counter()-debut:.2f}s")
    async def mauvais():
        time.sleep(0.2)  # bloque toute la boucle
        print(f"    bloquant fini a {time.perf_counter()-debut:.2f}s")
    await asyncio.gather(tic(), mauvais())

asyncio.run(main())

La sortie est sans appel : aucun tic ne s'affiche pendant les 0,2 s de blocage, tout est figé jusqu'à la fin du sleep synchrone.

    bloquant fini a 0.20s
    tic a 0.20s
    tic a 0.25s
    tic a 0.30s

Le deuxième piège est d'oublier le await. Appeler une coroutine sans l'attendre crée un objet coroutine qui n'est jamais exécuté ; Python émet un avertissement coroutine was never awaited et le travail attendu ne se produit pas. Le troisième est de mélanger synchrone et asynchrone sans réfléchir : appeler du code async depuis du code sync exige un asyncio.run, et l'inverse (du sync bloquant dans de l'async) réclame run_in_executor. Confondre les deux mondes est la source la plus courante de bugs subtils.

Contrôle de connaissances

Vérifiez que l'essentiel de ce guide est acquis. Les questions portent uniquement sur ce qui vient d'être expliqué ici.

Contrôle de connaissances

Validez vos connaissances avec ce quiz interactif

6 questions

6 min.

70% requis

Informations

• Le chronomètre démarre au clic sur Démarrer
• Questions à choix multiples, vrai/faux et réponses courtes
• Vous pouvez naviguer entre les questions
• Les résultats détaillés sont affichés à la fin

Démarrer le quiz

Lance le quiz et démarre le chronomètre

À retenir

• La boucle d'événements fait progresser les coroutines sur un seul thread, en profitant de chaque await pour changer de tâche.
• create_task planifie un travail en arrière-plan ; gardez une référence à la Task sinon elle peut disparaître.
• gather renvoie les résultats dans l'ordre des arguments ; return_exceptions=True tolère les échecs partiels.
• TaskGroup (3.11) est le choix par défaut : concurrence structurée et annulation des autres tâches en cas d'échec, via ExceptionGroup.
• asyncio.timeout (3.11) borne la durée d'un bloc et annule proprement ce qui dépasse.
• asyncio.Queue relie producteurs et consommateurs avec contre-pression grâce à maxsize.
• On intercepte CancelledError pour nettoyer, puis on le relance : jamais l'avaler.
• run_in_executor ou asyncio.to_thread déportent le code bloquant hors de la boucle.
• Pour le réseau, on passe par httpx ou aiohttp ; jamais requests ou time.sleep dans une coroutine.

Pour aller plus loin

• LiteLLM Async : un cas concret d'appels d'API en parallèle avec gather sur des modèles de langage.
• FastAPI : un framework web entièrement asynchrone qui repose sur les coroutines et async def.
• Les gestionnaires de contexte : comprendre le async with qui structure TaskGroup et timeout.
• concurrent.futures : l'API de plus haut niveau pour lancer un lot de tâches, complémentaire des coroutines pour du code bloquant.
• Comprendre le GIL : pourquoi un seul thread suffit à asyncio pour saturer l'I/O sans jamais heurter le verrou global.

×

📖 Voir la documentation →