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

# 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:
        ...