Lance plusieurs quêtes en même temps sans attendre bêtement — un seul thread, zéro temps mort.
La programmation asynchrone permet à ton programme de faire autre chose pendant qu'il attend (réseau, fichier, timer) au lieu de rester bloqué. En Python, c'est le module asyncio qui gère ça : un seul thread, une event loop, et des coroutines qui se passent la main aux moments d'attente.
Pourquoi ? Parce que dans un programme classique, requests.get() ou time.sleep() gèlent TOUT. Si tu dois faire 100 requêtes API, tu les fais une par une. Avec async, tu les lances toutes et tu récupères les résultats au fur et à mesure.
Dans un MMO, tu ne restes pas planté devant le PNJ pendant que ta quête de craft de 10 minutes tourne. Tu lances le craft, tu pars faire un donjon, et tu reviens quand c'est prêt. Le code synchrone, c'est le joueur qui fixe la barre de progression sans bouger. async, c'est jouer pendant les temps de chargement.
Une fonction déclarée avec async def. L'appeler ne l'exécute PAS : ça crée un objet coroutine, une quête acceptée mais pas encore lancée. Il faut await pour l'exécuter.
ma_coro() seul ne fait rien et lève un warning "never awaited".Le chef d'orchestre. Une boucle qui exécute les coroutines à tour de rôle : dès qu'une coroutine attend (await), la loop donne la main à une autre. C'est asyncio.run() qui la démarre.
Le mot-clé qui dit : "exécute ça, et si ça doit attendre, rends la main à la loop pendant ce temps". Utilisable uniquement à l'intérieur d'une fonction async def.
Une coroutine planifiée dans la loop via asyncio.create_task(). Contrairement à un simple await, la Task démarre tout de suite en arrière-plan, sans attendre que tu l'await.
Concurrence : un cuisinier qui jongle entre plusieurs plats (asyncio). Parallélisme : plusieurs cuisiniers en même temps (multiprocessing). Async = un seul thread qui ne perd jamais de temps.
Un appel bloquant (time.sleep, requests.get) gèle toute la loop. Sa version non-bloquante (asyncio.sleep, aiohttp) rend la main pendant l'attente.
import asyncio # le module standard, rien à installer
async def charger_niveau(nom: str, duree: float) -> str:
# async def → cette fonction est une coroutine
print(f"Chargement de {nom}...")
await asyncio.sleep(duree) # attente NON bloquante : la loop fait autre chose
print(f"{nom} prêt !")
return nom # une coroutine peut retourner une valeur
async def main() -> None:
# await seul = séquentiel (1.0s + 0.5s = 1.5s au total)
await charger_niveau("Map", 1.0)
await charger_niveau("Sons", 0.5)
# gather = concurrent (max(1.0, 0.5) = 1.0s au total !)
resultats = await asyncio.gather(
charger_niveau("Map", 1.0),
charger_niveau("Sons", 0.5),
)
print(resultats) # ['Map', 'Sons'] — ordre des arguments conservé
asyncio.run(main()) # démarre la event loop — UN SEUL run par programme
async def pour déclarer, await pour exécuter en attendant poliment, asyncio.run() pour tout démarrer depuis le code synchrone.asyncio.run() s'appelle UNE fois, tout en bas du programme. Jamais à l'intérieur d'une coroutine — la loop tourne déjà.| FONCTION | RÔLE | EXEMPLE |
|---|---|---|
| asyncio.run(coro) | Démarre la event loop et exécute la coroutine principale | asyncio.run(main()) |
| await coro | Exécute la coroutine et attend son résultat (rend la main pendant l'attente) | res = await fetch() |
| asyncio.sleep(s) | Pause non bloquante — le remplaçant async de time.sleep | await asyncio.sleep(1) |
| asyncio.gather(*coros) | Lance plusieurs coroutines en concurrence, retourne la liste des résultats dans l'ordre | await gather(a(), b()) |
| asyncio.create_task(coro) | Planifie la coroutine immédiatement en arrière-plan, retourne une Task | t = create_task(coro()) |
| asyncio.wait_for(coro, timeout) | Comme await, mais lève TimeoutError si trop long | await wait_for(f(), 5) |
| asyncio.as_completed(coros) | Itère sur les résultats dans l'ordre où ils FINISSENT | for f in as_completed(...) |
| task.cancel() | Annule une Task en cours (lève CancelledError dedans) | t.cancel() |
gather retourne les résultats dans l'ordre des arguments, même si les coroutines finissent dans le désordre. Pour traiter au fil de l'eau, utilise as_completed.Un async def qui contient yield. Il produit des valeurs au fil de l'eau, avec la possibilité d'await entre chaque yield (attendre une API, un timer...).
La boucle qui consomme un générateur async : async for x in gen(). À chaque tour, elle await la prochaine valeur — la loop reste libre pendant l'attente.
La version compacte : [x async for x in gen()] collecte tout en une ligne. Fonctionne aussi en set, dict, et avec un filtre if.
Le projet type : un générateur qui yield 10 nombres aléatoires, une compréhension qui les collecte, et la mesure de runtime qui prouve la concurrence.
#!/usr/bin/env python3
"""Générateur async, compréhension async et mesure de runtime."""
import asyncio
import random
import time
from typing import AsyncGenerator, List
async def async_generator() -> AsyncGenerator[float, None]:
"""Yield 10 nombres aléatoires, en attendant 1s entre chaque."""
for _ in range(10):
await asyncio.sleep(1) # await possible ENTRE les yields
yield random.uniform(0, 10) # yield → générateur async
async def async_comprehension() -> List[float]:
"""Collecte les 10 nombres via une compréhension async."""
return [n async for n in async_generator()] # ≈ 10 secondes
async def measure_runtime() -> float:
"""Lance 4 compréhensions EN PARALLÈLE et mesure le temps total."""
debut = time.perf_counter()
await asyncio.gather(*(async_comprehension() for _ in range(4)))
return time.perf_counter() - debut # ≈ 10s, PAS 40s !
measure_runtime dure ~10s et pas 40s ? Parce que gather lance les 4 compréhensions en concurrence — chacune passe son temps à attendre asyncio.sleep(1), et ces attentes se superposent.AsyncGenerator[YieldType, SendType] (depuis typing). Oublier l'annotation = points en moins au checker Holberton.[n async for n in async_generator() if n > 5] — même logique qu'une compréhension classique.Deux compteurs qui tournent en même temps dans un seul thread. La preuve visuelle que la loop jongle entre les coroutines.
import asyncio
async def compteur(nom: str, intervalle: float) -> None:
# Chaque tour : affiche, puis rend la main pendant l'attente
for i in range(3):
print(f"[{nom}] tick {i}")
await asyncio.sleep(intervalle)
async def main() -> None:
await asyncio.gather(
compteur("rapide", 0.5),
compteur("lent", 1.0),
)
asyncio.run(main())
# Les ticks s'entremêlent : rapide, lent, rapide, rapide, lent...
asyncio.sleep par time.sleep et observe : les compteurs deviennent séquentiels. C'est LA démo de la différence bloquant / non-bloquant.Cas réel : lancer des téléchargements en arrière-plan avec create_task, mesurer le temps gagné, et couper ce qui traîne avec wait_for.
import asyncio
import time
import random
async def telecharger(fichier: str) -> str:
duree = random.uniform(0.5, 2.0) # simule une latence réseau aléatoire
await asyncio.sleep(duree)
return f"{fichier} ({duree:.2f}s)"
async def main() -> None:
debut = time.perf_counter()
# create_task : les 3 téléchargements démarrent ICI, tout de suite
tasks = [asyncio.create_task(telecharger(f)) for f in ("map.pak", "skins.pak", "sons.pak")]
# wait_for : on refuse d'attendre plus de 3 secondes au total
try:
resultats = await asyncio.wait_for(asyncio.gather(*tasks), timeout=3.0)
print(*resultats, sep="\n")
except asyncio.TimeoutError:
print("Serveur trop lent, abandon !")
print(f"Total : {time.perf_counter() - debut:.2f}s") # ≈ le PLUS LENT, pas la somme
asyncio.run(main())
Le classique du projet 0x01. Python - Async : lancer n coroutines à délai aléatoire et retourner les délais dans l'ordre d'arrivée, sans utiliser sort().
#!/usr/bin/env python3
"""Lance n wait_random en concurrence et collecte les délais."""
import asyncio
import random
from typing import List
async def wait_random(max_delay: int = 10) -> float:
"""Attend un délai aléatoire entre 0 et max_delay, puis le retourne."""
delay: float = random.uniform(0, max_delay)
await asyncio.sleep(delay)
return delay
async def wait_n(n: int, max_delay: int) -> List[float]:
"""Retourne les n délais dans l'ordre où ils se terminent."""
tasks = [asyncio.create_task(wait_random(max_delay)) for _ in range(n)]
# as_completed livre chaque task dès qu'elle finit → liste triée "gratuitement"
return [await task for task in asyncio.as_completed(tasks)]
if __name__ == "__main__":
print(asyncio.run(wait_n(5, 7)))
# [0.83, 2.15, 3.40, 5.62, 6.91] — croissant sans sort() !
#!/usr/bin/env python3, docstrings partout, annotations de type obligatoires (typing.List), et pycodestyle qui passe.as_completed rend les résultats par ordre de FIN, et qu'ici le résultat de chaque coroutine EST sa durée. Deux concepts pour le prix d'un.async def main():
charger_niveau("Map", 1.0) # oups, pas de await
# RuntimeWarning: coroutine 'charger_niveau'
# was never awaited — rien ne s'exécute !
Appeler une coroutine crée juste l'objet. Sans await (ou create_task), le code dedans ne tourne jamais.
async def main():
await charger_niveau("Map", 1.0)
# ou en arrière-plan :
task = asyncio.create_task(charger_niveau("Sons", 0.5))
await task
Le warning "never awaited" dans tes logs = un bug garanti à corriger immédiatement.
import time
async def tache():
time.sleep(2) # BLOQUANT : gèle TOUTE la loop
# toutes les autres coroutines sont figées 2s
time.sleep bloque le thread entier. Toute la concurrence est perdue — pire qu'un programme synchrone.
import asyncio
async def tache():
await asyncio.sleep(2) # rend la main 2s
# les autres coroutines continuent de tourner
Règle générale : dans une coroutine, chaque attente doit être une version async (asyncio.sleep, aiohttp, aiofiles...).
# 3 requêtes indépendantes... exécutées une par une
a = await fetch("/users") # 1s
b = await fetch("/items") # 1s
c = await fetch("/stats") # 1s → total 3s
Awaiter chaque coroutine l'une après l'autre = code async qui se comporte comme du synchrone. Aucun gain.
# les 3 partent ensemble → total ≈ 1s
a, b, c = await asyncio.gather(
fetch("/users"),
fetch("/items"),
fetch("/stats"),
)
Si les tâches ne dépendent pas l'une de l'autre, gather (ou create_task). Await en série uniquement quand B a besoin du résultat de A.
async def déclare, await exécute. Une coroutine appelée sans await ne tourne jamais — surveille les warnings "never awaited".
await uniquement dans async def. Dans une fonction normale, c'est SyntaxError. Le pont entre les deux mondes, c'est asyncio.run().
Jamais de bloquant dans la loop. time.sleep, requests, input() gèlent tout. Cherche l'équivalent async (asyncio.sleep, aiohttp).
gather pour l'indépendant, await pour le dépendant. La concurrence ne vaut que si les tâches n'ont pas besoin l'une de l'autre.
asyncio.run() une seule fois, au point d'entrée. À l'intérieur d'une coroutine, la loop existe déjà : create_task ou await directement.
Async = I/O, pas CPU. Pour du calcul lourd, asyncio n'aide pas — c'est multiprocessing qu'il te faut.
# ── DÉCLARER ────────────────────────────────────────────
async def quete() -> str: ... # coroutine (ne tourne pas toute seule)
# ── EXÉCUTER ────────────────────────────────────────────
asyncio.run(main()) # point d'entrée : démarre la event loop
await quete() # exécute et attend (rend la main pendant)
# ── CONCURRENCE ─────────────────────────────────────────
await asyncio.gather(a(), b()) # tout en même temps, résultats dans l'ordre
t = asyncio.create_task(c()) # démarre en fond MAINTENANT
await t # récupère le résultat plus tard
asyncio.as_completed(tasks) # itère par ordre de FIN
# ── GÉNÉRATEURS & COMPRÉHENSIONS ────────────────────────
async def gen(): yield x # générateur async (await possible entre yields)
async for x in gen(): ... # consomme sans bloquer la loop
[x async for x in gen()] # compréhension async (dans un async def)
# ── ATTENTES ────────────────────────────────────────────
await asyncio.sleep(1) # pause non bloquante (PAS time.sleep !)
await asyncio.wait_for(f(), 5) # timeout → asyncio.TimeoutError
# ── LA RÈGLE D'OR ───────────────────────────────────────
# I/O (réseau, disque, timer) → asyncio
# Calcul CPU → multiprocessing
# Code simple sans attente → reste synchrone !