Apprendre Python de zéro jusqu'au niveau expert : syntaxe, structures de données, POO, générateurs, décorateurs, typage, tests, async et internals du langage.
Votre script télécharge 500 pages web une par une et prend 4 minutes. Votre calcul d'images sature un seul cœur sur huit. Ces deux problèmes n'ont pas la même solution, et Python propose trois outils distincts. Le choix dépend d'une seule question : le programme attend-il (I/O) ou calcule-t-il (CPU) ?
Le Global Interpreter Lock empêche deux threads d'exécuter du bytecode Python en même temps dans un même processus. Conséquences :
| Type de tâche | Threads | Processus | asyncio |
|---|---|---|---|
| I/O (réseau, disque, BDD) | efficace | efficace mais lourd | le plus efficace |
| CPU (calcul pur Python) | inutile | efficace | inutile |
Pendant une attente réseau, le GIL est relâché : les autres threads avancent. Pendant un calcul, il ne l'est pas.
Depuis Python 3.13, une version expérimentale free-threaded (python3.13t) supprime le GIL. Elle reste marginale en production mais change la donne à moyen terme.
from concurrent.futures import ThreadPoolExecutor, as_completed
import httpx
urls = [f"https://httpbin.org/delay/1?i={i}" for i in range(20)]
def telecharger(url: str) -> int:
return httpx.get(url, timeout=10).status_code
with ThreadPoolExecutor(max_workers=10) as pool:
futures = {pool.submit(telecharger, u): u for u in urls}
for f in as_completed(futures):
print(futures[f], f.result())20 requêtes d'une seconde en environ 2 secondes au lieu de 20.
Les threads partagent la mémoire. Une opération comme compteur += 1 n'est pas atomique :
import threading
verrou = threading.Lock()
compteur = 0
def incrementer():
global compteur
for _ in range(100_000):
with verrou:
compteur += 1Pour communiquer entre threads, queue.Queue est thread-safe et évite la plupart des verrous manuels.
from concurrent.futures import ProcessPoolExecutor
def est_premier(n: int) -> bool:
if n < 2:
return False
return all(n % d for d in range(2, int(n ** 0.5) + 1))
if __name__ == "__main__":
nombres = range(10_000_000, 10_050_000)
with ProcessPoolExecutor() as pool:
resultats = list(pool.map(est_premier, nombres, chunksize=1000))
print(sum(resultats))Chaque processus a son propre interpréteur et son propre GIL. Contreparties : démarrage plus lent, arguments et résultats sérialisés (pickle). Le garde if __name__ == "__main__" est obligatoire sur Windows et macOS.
asyncio exécute des milliers de tâches I/O dans un seul thread, grâce à une boucle d'événements. Une coroutine rend la main à chaque await, la boucle passe à une autre tâche prête.
import asyncio
async def attendre(nom: str, secondes: float) -> str:
await asyncio.sleep(secondes)
return f"{nom} terminé"
async def main():
resultats = await asyncio.gather(
attendre("A", 1),
attendre("B", 2),
attendre("C", 1),
)
print(resultats) # après ~2s, pas 4
asyncio.run(main())import asyncio
import httpx
async def recuperer(client: httpx.AsyncClient, url: str, sem: asyncio.Semaphore) -> int:
async with sem:
r = await client.get(url, timeout=10)
return r.status_code
async def main(urls: list[str]):
sem = asyncio.Semaphore(20) # max 20 requêtes simultanées
async with httpx.AsyncClient() as client:
async with asyncio.TaskGroup() as tg:
taches = [tg.create_task(recuperer(client, u, sem)) for u in urls]
return [t.result() for t in taches]TaskGroup (Python 3.11+) annule proprement toutes les tâches si l'une échoue. Je l'utilise à la place de gather dans le code neuf.
Le sémaphore n'est pas optionnel : lancer 10 000 requêtes d'un coup sur une API revient à s'auto-infliger un déni de service et à se faire bannir.
async def main():
try:
async with asyncio.timeout(5):
await operation_lente()
except TimeoutError:
print("Abandon après 5s")Un appel synchrone dans une coroutine bloque toute la boucle :
async def mauvais():
time.sleep(2) # gèle toutes les tâches
requests.get(url) # idem
async def correct():
await asyncio.sleep(2)
await asyncio.to_thread(requests.get, url) # délègue à un threadRègle absolue : dans du code async, utilisez des bibliothèques async (httpx, asyncpg, aiofiles) ou asyncio.to_thread.
async def producteur(q: asyncio.Queue):
for i in range(10):
await q.put(i)
await q.put(None)
async def consommateur(q: asyncio.Queue):
while (item := await q.get()) is not None:
print("traite", item)
await asyncio.sleep(0.1)
async def main():
q: asyncio.Queue[int | None] = asyncio.Queue(maxsize=5)
await asyncio.gather(producteur(q), consommateur(q))async def lignes_paginees(client, url):
while url:
r = await client.get(url)
data = r.json()
for item in data["items"]:
yield item
url = data.get("next")
async for item in lignes_paginees(client, "/api/users"):
...Tâche bloquée par le réseau/disque ?
├── Oui → peu de tâches, code existant synchrone → ThreadPoolExecutor
│ → beaucoup de tâches, code neuf → asyncio
└── Non, calcul CPU
├── calcul vectorisable → NumPy / Polars
└── sinon → ProcessPoolExecutorÉcrire un crawler qui part d'une URL, suit les liens du même domaine jusqu'à une profondeur de 2, avec 10 requêtes simultanées maximum, et affiche les pages en erreur. Squelette :
import asyncio, re
from urllib.parse import urljoin, urlparse
import httpx
LIEN = re.compile(r'href="([^"#]+)"')
async def crawler(depart: str, profondeur_max: int = 2):
domaine = urlparse(depart).netloc
vus: set[str] = set()
sem = asyncio.Semaphore(10)
async def visiter(client, url, profondeur):
if url in vus or profondeur > profondeur_max:
return
vus.add(url)
async with sem:
try:
r = await client.get(url, timeout=10, follow_redirects=True)
except httpx.HTTPError as e:
print("ERREUR", url, e)
return
if r.status_code >= 400:
print(r.status_code, url)
return
async with asyncio.TaskGroup() as tg:
for href in LIEN.findall(r.text):
suivant = urljoin(url, href)
if urlparse(suivant).netloc == domaine:
tg.create_task(visiter(client, suivant, profondeur + 1))
async with httpx.AsyncClient() as client:
await visiter(client, depart, 0)
print(f"{len(vus)} pages visitées")
asyncio.run(crawler("https://example.com"))La concurrence en Python n'est pas compliquée une fois la question I/O ou CPU tranchée. La plupart des applications web modernes (FastAPI, Starlette) sont construites sur asyncio : c'est l'outil à maîtriser en priorité.