Server-Sent Events (SSE) es una tecnología server push que permite a un servidor enviar actualizaciones en tiempo real a clientes web a través de una única conexión HTTP.

Resulta especialmente útil para aplicaciones que requieren actualizaciones en directo, como cuadros de mando, notificaciones o análisis en tiempo real.

Introducción a SSE

• SSE es un estándar que permite a los servidores iniciar la transmisión de datos hacia los clientes del navegador una vez establecida la conexión inicial con el cliente.
  
• A diferencia de los WebSockets, que permiten la comunicación bidireccional, SSE es unidireccional: los datos fluyen del servidor al cliente.
  
• Esto convierte a SSE en una alternativa más sencilla para escenarios en los que sólo el servidor necesita enviar actualizaciones al cliente.

¿Por qué utilizar la ESS?

SSE es ventajoso en escenarios donde:

1. Actualizaciones en tiempo real: Aplicaciones que necesitan enviar actualizaciones en tiempo real al cliente, como resultados deportivos en directo, actualizaciones bursátiles o aplicaciones de chat.
   
2. Monitorización de servidores y cuadros de mando: Los cuadros de mando de monitorización en tiempo real pueden beneficiarse de SSE para actualizar métricas y estados sin necesidad de que el cliente sondee el servidor constantemente.
   
3. Notificaciones: La implementación de notificaciones enviadas por el servidor puede gestionarse eficientemente con SSE, reduciendo la necesidad de sondeo constante del lado del cliente.
   
4. Transmisión de datos: El streaming de grandes conjuntos de datos o registros en tiempo real se simplifica con SSE.

Implementación de SSE en FastAPI

FastAPI, conocida por su velocidad y facilidad de uso, se integra bien con la biblioteca sse-starlette para implementar SSE. A continuación se explica cómo configurarlo.

Ejemplo de implementación

Paso 1: Instalar las bibliotecas necesarias

En primer lugar, instale FastAPI y sse-starlette:

pip install fastapi sse-starlette

Paso 2: Crear la aplicación FastAPI

from fastapi import FastAPI
from sse_starlette.sse import EventSourceResponse
import asyncio

app = FastAPI()
async def event_generator():
    while True:
        await asyncio.sleep(1)
        yield {"data": "This is a server-sent event!"}
@app.get("/events")
async def sse_endpoint():
    return EventSourceResponse(event_generator())

En este ejemplo, event_generator es un generador asíncrono que produce un nuevo evento cada segundo. El endpoint sse_endpoint utiliza EventSourceResponse para manejar la conexión SSE.

Casos de uso de SSE en FastAPI

Actualizaciones en directo para cuadros de mando:

• SSE proporciona una forma sencilla de enviar actualizaciones sin una gestión compleja de WebSocket para aplicaciones que muestran métricas en directo, como sistemas de supervisión o paneles de análisis en tiempo real.
  

Notificaciones en tiempo real:

• Las aplicaciones que necesitan notificar eventos a los usuarios (como aplicaciones de mensajería o sistemas de alerta) pueden utilizar SSE para enviar estas actualizaciones sin problemas.
  

Registros en tiempo real:

• SSE es una solución ligera que se adapta bien a las aplicaciones que necesitan mostrar registros en tiempo real u otros datos en streaming.
  

Ventajas de SSE:

1. Simplicidad: SSE es más sencillo de implementar y gestionar que WebSockets.
   
2. Reconexión automática: Los navegadores gestionan automáticamente las reconexiones, lo que hace que SSE sea robusto para las actualizaciones en tiempo real.
   
3. Compatibilidad HTTP: SSE funciona sobre conexiones HTTP/2 estándar, lo que facilita el trabajo con la infraestructura existente.
   

Alternativas a SSE:

WebSockets

• Comunicación bidireccional: Si la aplicación requiere comunicación bidireccional, son preferibles los WebSockets.
  
• Menor latencia: Los WebSockets proporcionan una latencia más baja que SSE, lo que los hace adecuados para aplicaciones altamente interactivas como los juegos en línea.

Sondeo largo

• Compatibilidad: El sondeo largo es compatible con navegadores antiguos y puede utilizarse cuando no se admiten WebSockets o SSE.
  
• Sobrecarga: Introduce más sobrecarga en comparación con SSE o WebSockets debido a las peticiones repetidas.

Ejemplos reales de uso de SSE

Los eventos enviados por el servidor (SSE) son ampliamente adoptados en aplicaciones del mundo real que requieren actualizaciones en tiempo real. He aquí algunos ejemplos notables:

1. Resultados deportivos en directo
   
2. Datos bursátiles y financieros
   
3. Plataformas de medios sociales, etc.
   

Profundicemos en el ejemplo de una aplicación de actualización de resultados deportivos en directo, como las que ofrecen ESPN o BBC Sports.

Imaginemos una aplicación de resultados deportivos en directo que ofrezca actualizaciones en tiempo real de partidos de fútbol en curso. La aplicación debe enviar actualizaciones como los goles marcados y las estadísticas del partido. Aquí está el enlace GitHub.

Implementación SSE

1. Implementación del lado del servidor:
   

El siguiente archivo contiene un resumen del partido.

// scores.py

[
    {
        "time": "00:01",
        "scores":"1:0",
        "event": "Goal! Team A scores!"
    },
    {
        "time": "00:05",
        "scores":"1:1",
        "event": "Goal! Team B scores!"
    },
    {
        "time": "00:10",
        "scores":"2:1",
        "event": "Goal! Team A scores again!"
    }
]

• El servidor mantiene una transmisión en directo de los datos del partido, que envía a los clientes conectados mediante SSE.
  
• Cada vez que se produce un evento (gol, sustitución, etc.), el servidor envía esta actualización a todos los clientes conectados.

# main.py

from fastapi import FastAPI
from sse_starlette.sse import EventSourceResponse
import asyncio
import json

app = FastAPI()

async def event_generator():
    with open("scores.json", "r") as file:
        scores = json.load(file)
    for score in scores:
        await asyncio.sleep(5)  
        yield f"Match Summary: {json.dumps(score)}\n\n"

@app.get("/live-scores")
async def live_scores_endpoint():
    return EventSourceResponse(event_generator())

# Adding CORS middleware
from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

Implementación del lado del cliente:

• El cliente (una aplicación web) establece una conexión SSE con el servidor y escucha las actualizaciones.

<!-- index.html -->

<!DOCTYPE html>
<html>
<head>
    <title>Live Sports Scores</title>
</head>
<body>
    <h1>Live Football Scores</h1>
    <div id="score-updates"></div>

    <script>
        const eventSource = new EventSource("http://127.0.0.1:8000/live-scores");

        eventSource.onmessage = function(event) {
            const newElement = document.createElement("div");
            newElement.innerHTML = event.data;
            document.getElementById("score-updates").appendChild(newElement);
        };

        eventSource.onerror = function(event) {
            console.error("EventSource failed:", event);
            eventSource.close();
        };
    </script>
</body>
</html>

Referencia

• https://fastapi.tiangolo.com/
  
• https://pypi.org/project/sse-starlette/
  

Conclusión

Con sse-starlette, integrar SSE en aplicaciones FastAPI es sencillo y eficaz. Tanto si está desarrollando cuadros de mando en directo, sistemas de notificación o aplicaciones de transmisión de datos, SSE le ofrece una solución robusta y sencilla.

Fuente