backend / python
ASGI : internals et frameworks async
Explication
Ce que vous allez apprendre
- Comprendre ce qu'ASGI standardise entre un serveur comme Uvicorn et une application Python
- Expliquer pourquoi ASGI existe en plus de WSGI (WebSocket, streaming, concurrence)
- Lire les trois ingrédients d'une application ASGI :
scope,receive,send - Écrire un middleware ASGI qui enveloppe une application sans la modifier
- Situer FastAPI/Starlette comme une couche ergonomique construite sur ce protocole brut
Dans quel contexte ?
Un développeur observe qu'un en-tête de réponse personnalisé (X-Response-Time) ajouté par un middleware FastAPI n'apparaît pas dans certaines réponses, notamment celles en streaming. Pour comprendre pourquoi, il faut descendre sous FastAPI et regarder comment send est réellement appelé plusieurs fois pour une réponse en streaming — ce que cette leçon détaille en manipulant directement le protocole ASGI, sans aucun framework par-dessus.
Le protocole invisible derrière FastAPI
Quand vous écrivez une route FastAPI avec @app.get("/ping"), un mécanisme bas niveau traduit la connexion réseau brute en cet appel de fonction confortable. Ce mécanisme s'appelle ASGI (Asynchronous Server Gateway Interface) : un contrat standardisé entre un serveur (comme Uvicorn) et une application Python, qui permet à n'importe quel framework respectant ce contrat de tourner sur n'importe quel serveur compatible.
| Protocole | Modèle | WebSocket | Streaming natif |
|---|---|---|---|
| WSGI (Flask, Django classique) | Synchrone, un thread par requête | Non | Non |
| ASGI (FastAPI, Starlette) | Asynchrone, basé sur asyncio | Oui | Oui |
Pourquoi pas simplement WSGI ?
WSGI, le standard plus ancien utilisé par Flask ou Django classique, a été conçu pour un monde synchrone : une requête, une réponse, un thread bloqué en attendant. Il ne sait gérer ni les WebSocket (connexions bidirectionnelles persistantes) ni le vrai streaming. ASGI a été pensé dès le départ pour asyncio : une application peut gérer des milliers de connexions concurrentes sans bloquer un thread par connexion.
Les trois ingrédients : scope, receive, send
Une application ASGI est une simple fonction asynchrone à trois paramètres. Le scope est un dictionnaire qui décrit la connexion (type, méthode, chemin, en-têtes) — l'équivalent de l'objet requête, mais en données brutes. receive est une fonction qu'on appelle pour obtenir les événements entrants (corps de la requête, déconnexion). send est une fonction qu'on appelle pour émettre la réponse, en général en au moins deux étapes : d'abord les en-têtes et le code de statut, puis le corps.
Les middlewares : une simple composition de fonctions
Un middleware ASGI n'est rien de plus qu'une fonction qui reçoit une application et en renvoie une autre qui l'enveloppe : elle peut inspecter ou modifier scope, intercepter ce que send transmet (par exemple pour ajouter un en-tête de timing), puis déléguer à l'application d'origine. C'est exactement ce mécanisme, empilé plusieurs fois, qui construit un pipeline de traitement de requête.
Astuce
Pour déboguer un comportement ASGI inattendu (en-tête manquant, réponse tronquée), ajoutez temporairement un print(scope) ou print(message) directement dans un middleware minimal. Vous verrez exactement les dictionnaires bruts échangés, sans l'abstraction confortable mais parfois opaque de FastAPI.
Comprendre ce protocole brut aide à déboguer un comportement inattendu de FastAPI/Starlette, puisque ces frameworks ne sont finalement qu'une couche ergonomique par-dessus.
Commandes & code
ASGI : internals et frameworks async
Le protocole bas niveau derriere FastAPI, Starlette et Uvicorn.
# --- ASGI vs WSGI : pourquoi un nouveau protocole ---
# WSGI (Django classique, Flask) : synchrone, une seule requete a la fois par worker,
# pas de WebSocket, pas de streaming natif.
# ASGI : asynchrone, gere HTTP + WebSocket + evenements de cycle de vie (lifespan),
# concu pour asyncio des le depart.
# --- Une application ASGI minimale, ecrite SANS framework ---
# Signature universelle : async def app(scope, receive, send)
async def application(scope, receive, send):
assert scope["type"] == "http"
# "receive" est un callable qui renvoie les evenements ENTRANTS (corps de requete, deconnexion)
evenement = await receive()
if evenement["type"] == "http.request":
corps = evenement.get("body", b"")
# "send" est un callable pour ENVOYER la reponse, en 2 etapes minimum
await send({
"type": "http.response.start",
"status": 200,
"headers": [(b"content-type", b"application/json")],
})
await send({
"type": "http.response.body",
"body": b'{"message": "Bonjour depuis ASGI brut"}',
})
# Lancer avec : uvicorn module:application
# --- Le "scope" : dictionnaire decrivant la connexion (equivalent de environ en WSGI) ---
async def afficher_scope(scope, receive, send):
print(scope["type"]) # "http" ou "websocket" ou "lifespan"
print(scope["method"]) # "GET", "POST", ...
print(scope["path"]) # "/utilisateurs/42"
print(scope["query_string"]) # b"page=2&limit=10"
print(scope["headers"]) # [(b"host", b"example.com"), ...]
print(scope.get("client")) # (ip, port) du client
# --- Lifespan protocol : hooks de demarrage/arret de l'application ---
async def app_avec_lifespan(scope, receive, send):
if scope["type"] == "lifespan":
while True:
message = await receive()
if message["type"] == "lifespan.startup":
# ouvrir un pool de connexions BDD, charger un cache, etc.
print("Demarrage : initialisation des ressources")
await send({"type": "lifespan.startup.complete"})
elif message["type"] == "lifespan.shutdown":
print("Arret : liberation des ressources")
await send({"type": "lifespan.shutdown.complete"})
return
# --- Middleware ASGI : une fonction qui enveloppe une autre application ASGI ---
def middleware_timing(app):
async def wrapped(scope, receive, send):
import time
debut = time.perf_counter()
async def send_avec_timing(message):
if message["type"] == "http.response.start":
duree_ms = (time.perf_counter() - debut) * 1000
message["headers"].append(
(b"x-response-time", f"{duree_ms:.2f}ms".encode())
)
await send(message)
await app(scope, receive, send_avec_timing)
return wrapped
application_avec_timing = middleware_timing(application)
# --- Middleware d'authentification, pattern de chaine ---
def middleware_auth(app):
async def wrapped(scope, receive, send):
headers = dict(scope.get("headers", []))
if b"authorization" not in headers:
await send({
"type": "http.response.start",
"status": 401,
"headers": [(b"content-type", b"text/plain")],
})
await send({"type": "http.response.body", "body": b"Non authentifie"})
return
await app(scope, receive, send)
return wrapped
# Composition de middlewares : chacun enveloppe le suivant
pipeline = middleware_auth(middleware_timing(application))
# --- Streaming de reponse : envoyer plusieurs chunks "http.response.body" ---
async def app_streaming(scope, receive, send):
await send({
"type": "http.response.start",
"status": 200,
"headers": [(b"content-type", b"text/plain")],
})
for i in range(5):
await send({
"type": "http.response.body",
"body": f"chunk {i}\n".encode(),
"more_body": i < 4, # False sur le dernier chunk : ferme la reponse
})
# --- WebSocket au niveau ASGI brut ---
async def app_websocket(scope, receive, send):
assert scope["type"] == "websocket"
await receive() # attend "websocket.connect"
await send({"type": "websocket.accept"})
while True:
evenement = await receive()
if evenement["type"] == "websocket.receive":
texte = evenement.get("text", "")
await send({"type": "websocket.send", "text": f"echo: {texte}"})
elif evenement["type"] == "websocket.disconnect":
break
# --- Comment Starlette/FastAPI batissent la-dessus (simplifie) ---
# Un framework ASGI transforme le triplet (scope, receive, send) en objets ergonomiques :
# Request (encapsule scope+receive), Response (encapsule send), routage par path/method.
class RequeteSimplifiee:
def __init__(self, scope, receive):
self.method = scope["method"]
self.path = scope["path"]
self._receive = receive
async def body(self) -> bytes:
evenement = await self._receive()
return evenement.get("body", b"")
class ReponseSimplifiee:
def __init__(self, contenu: str, status: int = 200):
self.contenu = contenu.encode()
self.status = status
async def envoyer(self, send):
await send({
"type": "http.response.start",
"status": self.status,
"headers": [(b"content-type", b"text/plain")],
})
await send({"type": "http.response.body", "body": self.contenu})
class MiniFramework:
def __init__(self):
self.routes = {}
def route(self, path):
def decorateur(handler):
self.routes[path] = handler
return handler
return decorateur
async def __call__(self, scope, receive, send):
if scope["type"] != "http":
return
requete = RequeteSimplifiee(scope, receive)
handler = self.routes.get(requete.path)
if handler is None:
reponse = ReponseSimplifiee("Not Found", 404)
else:
reponse = await handler(requete)
await reponse.envoyer(send)
mini_app = MiniFramework()
@mini_app.route("/ping")
async def ping(requete):
return ReponseSimplifiee("pong")Résumé
- ASGI generalise WSGI pour l'asynchrone : HTTP, WebSocket et lifecycle (startup/shutdown) sous un seul protocole.
- Une app ASGI est
async def app(scope, receive, send):scopedecrit la connexion,receive/sendsont des callables asynchrones. - Un middleware ASGI enveloppe simplement l'application suivante, en interceptant
receive/send. - Starlette/FastAPI ajoutent une couche ergonomique (Request/Response, routage) au-dessus de ce protocole brut.
Exercices pratiques
Mission : retrouver pourquoi un en-tête disparaît en streaming
Objectif : Diagnostiquer pourquoi un middleware de timing FastAPI n'ajoute pas son en-tête sur une réponse en streaming, en manipulant directement scope/receive/send.
Contexte
Un middleware ASGI middleware_timing ajoute un en-tête X-Response-Time en interceptant send au moment où le message "type": "http.response.start" passe. Sur une route classique, l'en-tête apparaît bien dans la réponse. Sur une route qui streame sa réponse en plusieurs morceaux avec "more_body": True, l'en-tête est absent la moitié du temps, selon le timing exact du serveur.
Tu dois comprendre pourquoi send est appelé plusieurs fois pour une réponse en streaming, corriger le middleware, puis écrire une petite application ASGI minimale sans framework.