asyncio Tasks — Exécution concurrente avec gather, Task et Queue

En s'appuyant sur async def et await, cet article couvre comment exécuter plusieurs coroutines en concurrence. asyncio.gather pour « lance tout et attends tout », asyncio.create_task pour « lance maintenant, attends plus tard », et asyncio.Queue pour producteur / consommateur — ces trois couvrent presque tous les motifs async que tu écriras dans de vrais projets.

À propos de l'exécution du code ici

async / await tourne autour du timing, mais le runner de ce site bufferise la sortie de print et l'affiche d'un coup une fois le script terminé. La sortie en temps réel et le ressenti du temps écoulé ne correspondront pas à un vrai environnement Python. L'article utilise des schémas pour rendre le comportement interne clair, mais si tu veux voir print couler en direct ou ressentir le vrai timing, exécute le code dans ton environnement Python local avec asyncio.run.

asyncio.gather — exécuter plusieurs coroutines en concurrence

« Lancer plusieurs APIs en concurrence et n'avancer qu'une fois toutes les réponses reçues » — un cas d'usage async classique. Le code synchrone additionne les temps de réponse, mais gather finit en le temps du plus lent.

asyncio.gather(coro1, coro2, ...) démarre toutes les coroutines passées à la fois et retourne une liste de résultats une fois qu'elles sont toutes finies. La liste revient dans l'ordre où tu les as passées — pas l'ordre d'achèvement — donc l'entrée et la sortie restent alignées.

import asyncio

async def fetch(name):
    await asyncio.sleep(1)        # Simule un appel d'API avec une attente d'1 sec
    return f"{name} done"

# Lance 3 en parallèle → tous les résultats en 1 sec (vs 3 sec en séquentiel)
results = await asyncio.gather(
    fetch("A"),
    fetch("B"),
    fetch("C"),
)
print(results)                    # ['A done', 'B done', 'C done']  ← ordre d'entrée

Que se passe-t-il en cas d'exception

Par défaut, si une coroutine dans gather lève, tout l'appel lève et avorte. Pour collecter les exceptions à la place, passe asyncio.gather(..., return_exceptions=True) — les objets exception arrivent alors comme éléments de la liste pour que tu puisses vérifier les types après coup. Utile quand tu veux taper plusieurs APIs et tolérer des échecs partiels.

Exécute 3 coroutines en concurrence avec asyncio.gather et vérifie que l'ordre d'achèvement peut différer mais que la valeur de retour reste dans l'ordre d'entrée. Chaque tâche attend un temps différent pour que tu puisses voir l'ordre d'achèvement (B → A → C) vs l'ordre de retour ([A, B, C]) se décaler.

① Ajoute import asyncio.

② Définis async def task(name, secs): — await asyncio.sleep(secs) puis return f"{name} done".

③ Exécute avec await asyncio.gather(task("A", 0.3), task("B", 0.1), task("C", 0.5)) et stocke dans results.

④ Affiche print("results:", results).

⑤ Affiche print("count:", len(results)).

(Exécute avec succès et l'explication apparaîtra.)

create_task et wait_for — lance maintenant, attends plus tard, avec timeout

asyncio.create_task(coro) enveloppe une coroutine dans un objet « Task » et la démarre immédiatement. Contrairement au « lance tout et attends tout » de gather, c'est le motif async classique de « lance maintenant, fais autre chose, puis await task pour récupérer le résultat plus tard ».

asyncio.wait_for(awaitable, timeout=N) est un filet de sécurité : « lève TimeoutError si ça ne finit pas en N secondes ». C'est standard de le combiner avec des Tasks comme garde-fou quand une Web API ne répond pas.

import asyncio

async def slow_api():
    await asyncio.sleep(2)
    return "response"

# Lance avec create_task (la Task démarre tout de suite)
task = asyncio.create_task(slow_api())

# D'autres tâches peuvent se faire pendant que la Task tourne en fond
print("task fired, doing other work...")

# Récupère le résultat avec await quand tu en as besoin
result = await task
print(result)                       # response

# wait_for ajoute un timeout (abandonne après 1 sec)
try:
    result = await asyncio.wait_for(slow_api(), timeout=1.0)
except asyncio.TimeoutError:
    print("timeout!")               # 2 sec de réponse vs 1 sec de budget → ici

Trois chemins vers « done » — return / exception / cancel

Il y a 3 chemins vers done : (1) achèvement normal — return a produit une valeur, (2) exception — quelque chose a levé à l'intérieur, (3) cancel — task.cancel() l'a interrompue. task.done() retourne True pour les trois chemins, et task.exception() extrait l'exception s'il y en a une.

Méthodes utiles de l'objet Task

L'objet Task retourné par create_task supporte des opérations utiles : task.cancel() pour interrompre, task.done() pour vérifier l'achèvement, task.result() pour récupérer le résultat d'une Task finie (lève si pas finie), et task.exception() pour récupérer une exception levée. Pratique pour contrôler du travail en arrière-plan de longue durée.

Lance deux Tasks avec create_task, fais autre chose entre-temps, puis récupère leurs résultats.

① Ajoute import asyncio.

② Définis async def task(name): — await asyncio.sleep(0) pour basculer, puis return f"{name} done".

③ Utilise asyncio.create_task(task("A")) et asyncio.create_task(task("B")) pour lancer 2 Tasks, en les stockant dans t1 et t2.

④ Pendant que les Tasks tournent, affiche tasks fired.

⑤ Utilise await t1 et await t2 pour récupérer chaque résultat et affiche comme A: ◯ / B: ◯.

Utilise asyncio.wait_for pour définir un timeout et essaie les cas dans le budget et hors-budget.

① Ajoute import asyncio.

② Définis async def slow_task(): — await asyncio.sleep(0.5) pour attendre 0,5 sec, puis return "response done".

③ await asyncio.wait_for(slow_task(), timeout=1.0) — finit en moins d'1 sec, donc ça réussit. Affiche comme success: ◯.

④ try: / except asyncio.TimeoutError: autour de await asyncio.wait_for(slow_task(), timeout=0.1) — 0,1 sec ne suffit pas, donc TimeoutError se déclenche. Attrape-la et affiche timeout!.

asyncio.Queue — producteur / consommateur

« Je veux qu'une coroutine alimente des valeurs et qu'une autre les consomme » — un motif fréquent en scraping, traitement de jobs, gestion de flux et autres situations où deux boucles à vitesses différentes doivent se synchroniser.

asyncio.Queue est une queue async pour passer des valeurs entre coroutines (un FIFO = First In First Out — les valeurs sortent dans l'ordre où tu les as mises). Utilise await queue.put(value) pour insérer et await queue.get() pour retirer — et quand la queue est vide / pleine, ça bascule automatiquement vers une autre tâche et attend, gardant les choses simples.

import asyncio

queue = asyncio.Queue()

# Mettre des valeurs
await queue.put("item-1")
await queue.put("item-2")

# Les retirer (FIFO = premier entré, premier sorti)
print(await queue.get())            # item-1
print(await queue.get())            # item-2

# get() sur une queue vide bascule vers une autre tâche et attend une valeur
# print(await queue.get())          # ← se met en pause ici jusqu'à un put

Arrêter proprement avec une sentinelle

Côté consommateur, while True: item = await queue.get() attend indéfiniment qu'une valeur arrive. Le motif est de faire pousser au producteur un marqueur de fin à la fin (typiquement None, ou une sentinelle custom = un objet de garde dédié que tu peux distinguer des vraies données). Le consommateur sort de la boucle dès qu'il voit le marqueur.

Essaie put et get à l'intérieur d'une seule coroutine pour confirmer les bases de Queue et FIFO (premier entré, premier sorti).

① Ajoute import asyncio et crée une asyncio.Queue vide.

② await queue.put("a"), "b", "c" — insère 3 valeurs dans l'ordre.

③ Appelle await queue.get() 3 fois et rassemble les valeurs dans une liste, puis affiche comme pulled: ◯.

Lancer producteur / consommateur en concurrence avec gather

Queue paie vraiment quand plusieurs coroutines se passent des valeurs. Sépare « alimenter » et « consommer » en coroutines distinctes et lance-les en concurrence avec gather — await put / await get agissent comme points de bascule, et les deux moitiés s'imbriquent naturellement.

import asyncio

async def producer(queue):
    for i in range(3):
        await queue.put(f"item-{i}")
    await queue.put(None)           # marqueur de fin

async def consumer(queue):
    while True:
        item = await queue.get()
        if item is None:            # sortir sur le marqueur de fin
            break
        print(f"processing: {item}")

queue = asyncio.Queue()
await asyncio.gather(producer(queue), consumer(queue))
# Sortie :
# processing: item-0
# processing: item-1
# processing: item-2

Construis un setup où le producteur dépose 3 items et le consommateur les rassemble dans une liste. Utilise None comme marqueur de fin.

① Ajoute import asyncio, crée une asyncio.Queue vide, et une liste vide results.

② Définis async def producer(queue): — fais 3 fois f"item-{i}" puis met None comme marqueur de fin.

③ Définis async def consumer(queue, results): — while True: appelle get ; si None alors break, sinon append à results.

④ Lance avec asyncio.gather(producer(queue), consumer(queue, results)) et affiche comme count: ◯ / first: ◯ / last: ◯.