WSGI vs ASGI: la interfaz que define tu servidor de producción

La diferencia entre WSGI y ASGI no es solo una cuestión de “síncrono vs asíncrono”: es el contrato que tu aplicación firma con el servidor, y ese contrato determina qué tipos de conexiones puedes manejar, cómo escala bajo carga y qué tecnologías de tiempo real están disponibles o simplemente prohibidas.

WSGI (definido en PEP 3333) es un callable Python que recibe environ y start_response, procesa la request de forma síncrona, y devuelve un iterable con el body. Un worker WSGI maneja exactamente una request a la vez; cuando esa llamada bloquea esperando la base de datos, el worker está muerto para todo lo demás. Gunicorn compensa esto lanzando múltiples procesos worker (tipicamente 2 * CPUs + 1), así que el paralelismo es real pero tiene un coste: cada worker es un proceso Python completo con su propio heap y su propio GIL.

ASGI (PEP no oficial, especificación en el repositorio de asgiref) introduce un contrato diferente: la aplicación es un callable async que recibe scope, receive y send. El scope describe la conexión (HTTP, WebSocket, lifespan); receive y send son coroutines que permiten al servidor y a la aplicación intercambiar mensajes de forma incremental. Un worker ASGI con asyncio puede tener miles de conexiones en vuelo simultáneamente porque mientras espera I/O en una, el event loop avanza con otras.

Esto habilita cosas que WSGI hace imposibles por diseño: WebSockets necesitan una conexión persistente bidireccional; Server-Sent Events requieren que la respuesta quede abierta indefinidamente; HTTP/2 push necesita multiplexar streams sobre una sola conexión TCP. WSGI no tiene modelo mental para ninguno de esos casos porque su interfaz asume que la respuesta termina y el worker queda libre.

La pregunta relevante no es “¿cuál es mejor?” sino “¿cuándo el overhead de ASGI no se justifica?”. Si tu aplicación es mayoritariamente request/response sin I/O concurrente, Flask con Gunicorn es más predecible, tiene mejor ecosistema de middleware maduro y el modelo de procesos aislados te da fault isolation gratis. ASGI brilla cuando el I/O concurrente es real: muchas conexiones simultáneas a APIs externas, WebSockets, o cuando ya estás usando asyncio en toda la stack.

# wsgi_vs_asgi_demo.py
# Requiere: pip install flask fastapi uvicorn gunicorn httpx anyio

import asyncio
import time

import httpx
from flask import Flask
from fastapi import FastAPI
from fastapi.responses import StreamingResponse

# ── WSGI: Flask ──────────────────────────────────────────────────────────────

wsgi_app = Flask(__name__)

@wsgi_app.route("/wsgi/sequential")
def wsgi_sequential():
    """
    Cada llamada a time.sleep bloquea el worker completo.
    Con gunicorn -w 4, solo 4 requests pueden ejecutarse en paralelo.
    """
    time.sleep(0.1)   # simula I/O bloqueante (base de datos, API externa)
    time.sleep(0.1)
    return {"total_sleep": "200ms", "interface": "WSGI"}


# ── ASGI: FastAPI ────────────────────────────────────────────────────────────

asgi_app = FastAPI()

@asgi_app.get("/asgi/concurrent")
async def asgi_concurrent():
    """
    asyncio.sleep cede el control al event loop; el worker puede procesar
    otras requests mientras espera. Un solo proceso maneja miles de estas.
    """
    await asyncio.sleep(0.1)
    await asyncio.sleep(0.1)
    return {"total_sleep": "200ms", "interface": "ASGI"}


@asgi_app.get("/asgi/fan-out")
async def asgi_fan_out():
    """
    Fan-out a tres APIs externas en paralelo: imposible sin async en WSGI
    sin lanzar threads explícitamente.
    """
    async with httpx.AsyncClient() as client:
        results = await asyncio.gather(
            client.get("https://httpbin.org/delay/1"),
            client.get("https://httpbin.org/delay/1"),
            client.get("https://httpbin.org/delay/1"),
        )
    # Los tres requests salen al mismo tiempo; el tiempo total es ~1s, no ~3s.
    return {"statuses": [r.status_code for r in results]}


async def event_stream():
    """
    Server-Sent Events: la conexión permanece abierta y el servidor empuja
    mensajes. WSGI no puede modelar esto; ASGI lo hace de forma natural
    porque send/receive son streams, no un único par request/response.
    """
    for i in range(5):
        await asyncio.sleep(1)
        yield f"data: tick {i}\n\n"   # formato SSE estándar

@asgi_app.get("/asgi/sse")
async def asgi_sse():
    return StreamingResponse(event_stream(), media_type="text/event-stream")


# ── Inspección del contrato de cada interfaz ─────────────────────────────────

def inspect_wsgi_contract(app):
    """
    Un servidor WSGI llama a la aplicación así de simple.
    environ es un dict con los headers y metadatos HTTP.
    start_response es un callable que la app invoca para fijar status y headers.
    """
    responses = []

    def start_response(status, headers):
        responses.append({"status": status, "headers": headers})

    environ = {
        "REQUEST_METHOD": "GET",
        "PATH_INFO": "/wsgi/sequential",
        "SERVER_NAME": "localhost",
        "SERVER_PORT": "8000",
        "wsgi.input": __import__("io").BytesIO(b""),
        "wsgi.errors": __import__("sys").stderr,
        "wsgi.url_scheme": "http",
        "wsgi.multithread": False,
        "wsgi.multiprocess": True,
        "wsgi.run_once": False,
    }

    body = b"".join(app(environ, start_response))
    return responses[0]["status"], body


async def inspect_asgi_contract(app):
    """
    Un servidor ASGI llama a la aplicación con scope/receive/send.
    receive y send son coroutines: el protocolo es un diálogo, no un monólogo.
    """
    scope = {
        "type": "http",
        "asgi": {"version": "3.0"},
        "http_version": "1.1",
        "method": "GET",
        "path": "/asgi/concurrent",
        "query_string": b"",
        "headers": [],
    }

    messages = []

    async def receive():
        # El servidor entrega el body de la request cuando la app lo pide
        return {"type": "http.request", "body": b"", "more_body": False}

    async def send(message):
        # La app empuja mensajes al servidor: primero headers, luego body
        messages.append(message)

    await app(scope, receive, send)
    return messages


if __name__ == "__main__":
    # Demostración del contrato WSGI
    status, body = inspect_wsgi_contract(wsgi_app.wsgi_app)
    print(f"WSGI status: {status}")

    # Demostración del contrato ASGI
    messages = asyncio.run(inspect_asgi_contract(asgi_app))
    response_start = next(m for m in messages if m["type"] == "http.response.start")
    print(f"ASGI status: {response_start['status']}")

Qué está pasando realmente aquí

El punto de entrada de inspect_wsgi_contract te muestra el contrato en su forma más desnuda: la aplicación es una función que el servidor llama una vez, de forma síncrona, y cuando retorna la respuesta está completa. No hay forma de que la app diga “espera, todavía tengo más datos”; la única salida es el iterable que devuelve.

inspect_asgi_contract revela por qué ASGI puede hacer más: la app recibe send como herramienta para empujar mensajes al servidor en cualquier momento. Fíjate en cómo event_stream usa yield en un generador async: cada yield llama a send internamente vía StreamingResponse, manteniendo la conexión abierta y enviando datos cuando están listos. Eso no es un truco de FastAPI, es lo que el protocolo ASGI hace posible por diseño.

El endpoint /asgi/fan-out ilustra el caso más común donde ASGI gana de forma objetiva. Con WSGI y Flask necesitarías concurrent.futures.ThreadPoolExecutor para hacer tres llamadas HTTP en paralelo, lo cual funciona pero añade complejidad y consume threads del OS. Con ASGI el fan-out es asyncio.gather: un solo thread del OS, sin context switches de kernel, con el mismo efecto.

La elección de servidor refleja directamente esta dicotomía. Gunicorn con --workers 4 arranca cuatro procesos Python; cada uno puede procesar una request bloqueante simultáneamente. Uvicorn con --workers 1 arranca un proceso con un event loop que maneja miles de conexiones concurrentes siempre que el I/O sea async. En producción se suele combinar Gunicorn como gestor de procesos con Uvicorn como worker class (gunicorn -k uvicorn.workers.UvicornWorker), obteniendo fault isolation de procesos más concurrencia dentro de cada uno.

El overhead de ASGI que mencioné antes es concreto: el event loop de asyncio tiene latencia de scheduling, el modelo mental de “todo debe ser async o bloqueas el loop” es una restricción real, y librerías síncronas como psycopg2 o el ORM clásico de Django bloquean el worker si las llamas sin asyncio.run_in_executor. Si tu app hace queries SQL, sirve HTML renderizado y no tiene WebSockets ni I/O masivamente paralelo, Flask + Gunicorn te evita esa clase de accidentes.

Errores que debes conocer

Error: llamar código síncrono bloqueante desde un handler async, matando la concurrencia que ASGI prometía.

# ❌ Wrong
import psycopg2

@asgi_app.get("/users")
async def get_users():
    conn = psycopg2.connect("dbname=mydb")   # bloquea el event loop entero
    cur = conn.cursor()
    cur.execute("SELECT * FROM users")
    return cur.fetchall()

# ✅ Right
import asyncpg

@asgi_app.get("/users")
async def get_users():
    conn = await asyncpg.connect("postgresql://localhost/mydb")
    rows = await conn.fetch("SELECT * FROM users")
    await conn.close()
    return [dict(r) for r in rows]

Usar un driver async (asyncpg, aiosqlite, aiomysql) garantiza que la llamada cede al event loop mientras espera la base de datos.

Error: desplegar una app ASGI con un servidor WSGI puro (o viceversa), lo que produce errores crípticos en runtime en lugar de un fallo claro al arrancar.

# ❌ Wrong: gunicorn no entiende el contrato ASGI
gunicorn myapp:asgi_app

# ✅ Right: uvicorn para ASGI, o gunicorn con el worker class correcto
uvicorn myapp:asgi_app --host 0.0.0.0 --port 8000
# o en producción multi-worker:
gunicorn myapp:asgi_app -k uvicorn.workers.UvicornWorker -w 4

El worker class de Uvicorn dentro de Gunicorn es el puente: Gunicorn gestiona los procesos y las señales del OS, Uvicorn implementa el protocolo ASGI dentro de cada proceso.

180

Dejar un comentario

Tu dirección de correo electrónico no será publicada. Los campos obligatorios están marcados con *

Scroll al inicio