Modern Python Engineering
Chapitre 3
03-Python-Async
03-Python-Async
Python Async — Cours Complet
1. Concepts Fondamentaux
asyncio — Programmation Asynchrone
L'asyncio (Python 3.4+) permet la concurrence via une boucle d'événements. Contrairement au threading, il n'y a pas de concurrence système : tout se passe dans un seul thread, avec du cooperative multitasking.
import asyncio
import time
# Coroutine — fonction async
async def say_hello(name: str) -> str:
await asyncio.sleep(1) # point de suspension non-bloquant
return f"Hello, {name}!"
# Point d'entrée (3.10+)
async def main() -> None:
result = await say_hello("Alice")
print(result)
asyncio.run(main())
Boucle d'Événements
Diagramme en cours de génération...
# Inspecter la boucle
loop = asyncio.get_event_loop()
print(loop.is_running())
print(loop.get_debug())
2. Coroutines et await
async def
async def fetch_data(url: str) -> dict:
"""Coroutine qui simule un fetch HTTP."""
await asyncio.sleep(0.5) # I/O simulée
return {"url": url, "data": "..."}
# Une coroutine ne fait rien tant qu'elle n'est pas awaitée
coro = fetch_data("https://example.com") # <coroutine object>
# result = await coro # c'est ici que ça s'exécute
async for — Async Iterators
class AsyncRange:
"""Itérateur asynchrone."""
def __init__(self, start: int, end: int) -> None:
self.current = start
self.end = end
def __aiter__(self):
return self
async def __anext__(self) -> int:
if self.current >= self.end:
raise StopAsyncIteration
await asyncio.sleep(0.1)
value = self.current
self.current += 1
return value
async def main() -> None:
async for i in AsyncRange(0, 5):
print(i)
async with — Async Context Managers
class AsyncResource:
async def __aenter__(self):
print("Acquiring resource...")
await asyncio.sleep(0.1)
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
print("Releasing resource...")
await asyncio.sleep(0.1)
async def use(self) -> str:
return "Resource used"
async def main() -> None:
async with AsyncResource() as resource:
result = await resource.use()
print(result)
3. Tasks et Futures
Task — Exécution Concurrente
async def say_after(delay: float, message: str) -> str:
await asyncio.sleep(delay)
return message
async def main() -> None:
# Créer des tasks (exécutées en parallèle)
task1 = asyncio.create_task(say_after(1, "Hello"))
task2 = asyncio.create_task(say_after(2, "World"))
print("Tasks created, waiting...")
result1 = await task1
result2 = await task2
print(f"{result1} {result2}") # "Hello World" (après ~2s)
asyncio.run(main())
Future — Résultat d'une opération asynchrone
async def set_future(fut: asyncio.Future, value: str) -> None:
await asyncio.sleep(1)
fut.set_result(value)
async def main() -> None:
fut = asyncio.Future()
asyncio.create_task(set_future(fut, "done"))
result = await fut
print(result) # "done" après 1s
4. Concurrence
asyncio.gather
async def fetch(url: str) -> dict:
await asyncio.sleep(1)
return {"url": url, "status": 200}
async def main() -> None:
urls = [
"https://api.example.com/1",
"https://api.example.com/2",
"https://api.example.com/3",
]
# Toutes en parallèle
results = await asyncio.gather(*[fetch(url) for url in urls])
for result in results:
print(result)
# Avec gestion d'erreurs
async def safe_fetch(url: str) -> dict | None:
try:
return await fetch(url)
except Exception:
return None
results = await asyncio.gather(
*[safe_fetch(url) for url in urls],
return_exceptions=True, # ne propage pas les exceptions
)
asyncio.as_completed
async def process_as_completed() -> None:
tasks = [fetch(f"https://api.example.com/{i}") for i in range(5)]
for coro in asyncio.as_completed(tasks):
result = await coro
print(f"Completed: {result}")
# Traite chaque résultat dès qu'il est prêt
asyncio.wait
async def main() -> None:
task1 = asyncio.create_task(fetch("url1"))
task2 = asyncio.create_task(fetch("url2"))
task3 = asyncio.create_task(fetch("url3"))
done, pending = await asyncio.wait(
[task1, task2, task3],
timeout=5.0, # timeout optionnel
return_when=asyncio.FIRST_COMPLETED, # ou ALL_COMPLETED, FIRST_EXCEPTION
)
for task in done:
print(f"Done: {task.result()}")
for task in pending:
task.cancel() # annule les tasks en attente
asyncio.shield
async def cancellable() -> None:
"""Protège une opération de l'annulation."""
await asyncio.sleep(10)
async def main() -> None:
task = asyncio.create_task(cancellable())
await asyncio.sleep(0.1)
# Annuler, mais protège le sleep
task.cancel()
try:
await asyncio.shield(task)
except asyncio.CancelledError:
print("Task was cancelled but shielded part completed")
5. Async I/O
aiohttp — HTTP Client/Serveur Async
import aiohttp
import asyncio
async def fetch_url(session: aiohttp.ClientSession, url: str) -> dict:
async with session.get(url) as response:
return await response.json()
async def main() -> None:
async with aiohttp.ClientSession() as session:
urls = [
"https://jsonplaceholder.typicode.com/posts/1",
"https://jsonplaceholder.typicode.com/posts/2",
]
tasks = [fetch_url(session, url) for url in urls]
results = await asyncio.gather(*tasks)
print(results)
asyncio.run(main())
aiofiles — Async File I/O
import aiofiles
import asyncio
async def read_large_file(filepath: str) -> str:
async with aiofiles.open(filepath, mode="r") as f:
content = await f.read()
return content
async def write_large_file(filepath: str, data: str) -> None:
async with aiofiles.open(filepath, mode="w") as f:
await f.write(data)
async def process_file() -> None:
# Lecture asynchrone (ne bloque pas l'event loop)
async with aiofiles.open("large_file.txt") as f:
async for line in f:
print(line.strip())
6. Async Generators
from typing import AsyncGenerator
async def fibonacci_async(limit: int) -> AsyncGenerator[int, None]:
"""Générateur asynchrone de Fibonacci."""
a, b = 0, 1
while a < limit:
await asyncio.sleep(0.1) # simule un calcul lent
yield a
a, b = b, a + b
async def main() -> None:
async for num in fibonacci_async(100):
print(num)
# Async generator avec send
async def accumulator() -> AsyncGenerator[int, int]:
total = 0
while True:
value = await yield total # await yield !
if value is None:
break
total += value
async def use_accumulator() -> None:
gen = accumulator()
await gen.asend(None) # premier démarrage
print(await gen.asend(10)) # 10
print(await gen.asend(20)) # 30
7. Async Queues
import random
async def producer(queue: asyncio.Queue[int], name: str) -> None:
"""Producteur qui ajoute des items à la queue."""
for i in range(5):
item = random.randint(1, 100)
await queue.put(item)
print(f"{name} produced {item}")
await asyncio.sleep(random.random())
async def consumer(queue: asyncio.Queue[int], name: str) -> None:
"""Consommateur qui traite les items."""
while True:
item = await queue.get()
print(f"{name} consumed {item} (queue size: {queue.qsize()})")
queue.task_done()
await asyncio.sleep(random.random() * 0.5)
async def main() -> None:
queue: asyncio.Queue[int] = asyncio.Queue(maxsize=10)
# Créer producteurs et consommateurs
producers = [asyncio.create_task(producer(queue, f"P{i}")) for i in range(3)]
consumers = [asyncio.create_task(consumer(queue, f"C{i}")) for i in range(2)]
await asyncio.gather(*producers)
await queue.join() # attend que tous les items soient traités
for c in consumers:
c.cancel()
asyncio.run(main())
8. Synchronisation
Lock — Exclusion Mutuelle
async def critical_section(lock: asyncio.Lock, name: str) -> None:
async with lock:
print(f"{name} entered critical section")
await asyncio.sleep(1)
print(f"{name} leaving critical section")
async def main() -> None:
lock = asyncio.Lock()
await asyncio.gather(
critical_section(lock, "A"),
critical_section(lock, "B"),
critical_section(lock, "C"),
)
Semaphore — Limiter la Concurrence
async def limited_request(sem: asyncio.Semaphore, url: str) -> dict:
async with sem:
print(f"Fetching {url}")
await asyncio.sleep(1)
return {"url": url, "status": 200}
async def main() -> None:
sem = asyncio.Semaphore(3) # max 3 requêtes simultanées
urls = [f"https://api.example.com/{i}" for i in range(10)]
tasks = [limited_request(sem, url) for url in urls]
results = await asyncio.gather(*tasks)
Event — Signal entre Coroutines
async def waiter(event: asyncio.Event, name: str) -> None:
print(f"{name} waiting for event...")
await event.wait()
print(f"{name} received event!")
async def setter(event: asyncio.Event) -> None:
await asyncio.sleep(2)
print("Setting event!")
event.set()
async def main() -> None:
event = asyncio.Event()
await asyncio.gather(
waiter(event, "A"),
waiter(event, "B"),
setter(event),
)
9. FastAPI Async
from fastapi import FastAPI, BackgroundTasks
import asyncio
app = FastAPI()
# Route asynchrone
@app.get("/")
async def root() -> dict:
await asyncio.sleep(0.1)
return {"message": "Hello Async World!"}
# Background tasks
def write_log(message: str) -> None:
with open("log.txt", "a") as f:
f.write(f"{message}\n")
@app.post("/process")
async def process(item: dict, background_tasks: BackgroundTasks) -> dict:
background_tasks.add_task(write_log, f"Processed: {item}")
# Traitement lourd...
await asyncio.sleep(1)
return {"status": "done", "data": item}
# WebSocket
from fastapi import WebSocket
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket) -> None:
await websocket.accept()
while True:
data = await websocket.receive_text()
await websocket.send_text(f"Echo: {data}")
10. Trio vs asyncio
Trio — Structure de Concurrence Alternative
import trio
async def child() -> str:
await trio.sleep(1)
return "Trio says hello!"
async def main() -> None:
async with trio.open_nursery() as nursery:
nursery.start_soon(child)
nursery.start_soon(child)
trio.run(main)
Comparaison
| Feature | asyncio | Trio |
|---|---|---|
| Paradigme | Tasks/Callbacks | Structured concurrency |
| Syntaxe | asyncio.gather | nursery.start_soon |
| Annulation | task.cancel() | nursery.cancel_scope |
| Timeouts | asyncio.wait_for | trio.fail_after |
| Écosystème | Très large | Plus petit |
| Performance | Comparable | Comparable |
# Trio — structured concurrency
async def fetch_all(urls: list[str]) -> list[dict]:
async with trio.open_nursery() as nursery:
results = []
for url in urls:
nursery.start_soon(fetch_one, url, results)
return results
# asyncio équivalent
async def fetch_all_asyncio(urls: list[str]) -> list[dict]:
return await asyncio.gather(*[fetch_one(url) for url in urls])
11. Bonnes Pratiques
# 1. Toujours utiliser asyncio.run() comme point d'entrée
async def main() -> None: ...
if __name__ == "__main__":
asyncio.run(main())
# 2. Éviter de mélanger sync et async
# Mauvais :
def sync_func():
asyncio.run(async_func()) # anti-pattern !
# Bon :
async def async_func():
await another_async_func()
# 3. Timeouts pour éviter les blocages
try:
result = await asyncio.wait_for(long_op(), timeout=5.0)
except asyncio.TimeoutError:
print("Operation timed out")
# 4. asyncio.gather avec return_exceptions
results = await asyncio.gather(
*tasks,
return_exceptions=True,
)
# 5. Grouper les connexions (aiohttp)
async with aiohttp.ClientSession() as session:
async with session.get(url) as resp:
...