arunseohub.in

Python Server-Sent Events Sse Web App

Written by

arun

in

Server‑Sent Events (SSE) give web developers a simple, efficient way to push real‑time updates from a Python backend directly to a browser without the overhead of WebSockets. In this guide we’ll explore how to build a robust Python SSE web app, why SSE can be the perfect fit for live dashboards, notifications, and streaming data, and we’ll walk through a complete example using Flask and FastAPI. By the end of this post you’ll have a production‑ready template that you can adapt to any project that needs low‑latency, one‑way server communication.

What Are Server‑Sent Events?

Server‑Sent Events are part of the HTML5 EventSource API. Unlike WebSockets, which establish a full‑duplex channel, SSE creates a unidirectional stream from server to client. The browser automatically reconnects if the connection drops, and the data is sent as plain text following a simple line‑based format.

  • Lightweight: No binary framing, just UTF‑8 text.
  • Built‑in reconnection: The client retries automatically with an exponential back‑off.
  • Easy to implement: A single HTTP endpoint that returns text/event-stream.
  • SEO‑friendly: Since the page itself is still served via normal HTTP, search engines can crawl the static content while the dynamic part runs in the background.

When to Choose SSE Over WebSockets

Both SSE and WebSockets can deliver real‑time data, but they excel in different scenarios. Use SSE when you need:

  1. Simple, one‑way updates (e.g., live scores, stock tickers, chat notifications).
  2. Broad browser support without polyfills (all modern browsers support EventSource).
  3. Low server resource consumption – SSE works over standard HTTP/1.1 or HTTP/2.
  4. Automatic reconnection and event ID tracking out of the box.

If you need bidirectional communication, binary data, or sub‑millisecond latency, WebSockets remain the better choice.

Setting Up the Python Environment

Required Packages

For this tutorial we’ll use two popular Python web frameworks. You can pick one based on your existing stack:

  • Flask – lightweight, easy to get started.
  • FastAPI – async‑first, built on Starlette, excellent for high‑concurrency.

Install the dependencies with pip:

pip install flask fastapi uvicorn

Building an SSE Endpoint with Flask

Basic Flask App

Below is a minimal Flask app that streams the current server time every second.

from flask import Flask, Response, stream_with_context
import time

app = Flask(__name__)

def event_stream():
    """Generator that yields a new SSE message every second."""
    counter = 0
    while True:
        counter += 1
        data = f"data: Server time {time.strftime('%Y-%m-%d %H:%M:%S')} (tick {counter})\n\n"
        yield data
        time.sleep(1)

@app.route('/stream')
def stream():
    # Set the correct MIME type for SSE
    return Response(stream_with_context(event_stream()),
                    mimetype='text/event-stream')

if __name__ == '__main__':
    app.run(debug=True, threaded=True)

Key points:

  • Use stream_with_context to keep Flask’s request context alive.
  • Yield strings that follow the SSE format: data: <payload>\n\n.
  • Set mimetype='text/event-stream' so the browser knows to treat the response as an event stream.

Client‑Side JavaScript

On the front‑end, the EventSource object handles the connection automatically.

<script>
    const source = new EventSource('/stream');

    source.onmessage = function(event) {
        console.log('Received:', event.data);
        const log = document.getElementById('log');
        const entry = document.createElement('div');
        entry.textContent = event.data;
        log.prepend(entry);
    };

    source.onerror = function(err) {
        console.error('SSE error:', err);
    };
</script>

<div id="log"></div>

Building an SSE Endpoint with FastAPI

Why FastAPI?

FastAPI leverages Python’s asyncio library, allowing thousands of concurrent connections with minimal overhead. Its built‑in support for StreamingResponse makes SSE implementation straightforward.

Async SSE Example

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import asyncio
import datetime

app = FastAPI()

async def event_generator():
    """Async generator that yields a timestamp every second."""
    counter = 0
    while True:
        counter += 1
        now = datetime.datetime.utcnow().isoformat()
        # SSE format: data: ...\n\n
        yield f"data: {{\"time\": \"{now}\", \"tick\": {counter}}}\n\n"
        await asyncio.sleep(1)

@app.get("/sse")
async def sse_endpoint():
    return StreamingResponse(event_generator(),
                             media_type="text/event-stream")

FastAPI automatically runs on an ASGI server (e.g., uvicorn). Start it with:

uvicorn myapp:app --reload

Client Code for FastAPI SSE

The JavaScript remains identical because the browser only cares about the text/event-stream MIME type.

<script>
    const source = new EventSource('/sse');

    source.addEventListener('message', (e) => {
        const data = JSON.parse(e.data);
        console.log('Server time:', data.time, 'Tick:', data.tick);
    });
</script>

Best Practices for Production‑Ready SSE

1. Use a Reverse Proxy That Supports Streaming

  • Nginx: Set proxy_buffering off; and proxy_cache off; for the SSE location.
  • Apache: Enable mod_proxy_wstunnel and disable buffering with ProxyPass ... ws:// (even though it’s not WebSocket, the same directives apply).

2. Keep Connections Alive

Browsers may close idle connections after 30‑60 seconds. Send a comment line (:) as a heartbeat:

def event_stream():
    while True:
        yield ": keep‑alive\\n\\n"   # comment line, ignored by client
        # then your actual data...

3. Set Appropriate CORS Headers

If your front‑end lives on a different domain, add CORS support:

# Flask example
from flask_cors import CORS
CORS(app)

# FastAPI example
from fastapi.middleware.cors import CORSMiddleware
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],   # Adjust for security
    allow_methods=["GET"],
)

4. Limit Message Size

Browsers impose a default limit (~64 KB). If you need larger payloads, split them into multiple events or switch to WebSockets.

5. Graceful Shutdown

When the server stops, close all generator loops cleanly to avoid “broken pipe” warnings.

try:
    while running:
        yield data
except GeneratorExit:
    # Cleanup logic here
    pass

Real‑World Use Cases for Python SSE

  • Live dashboards: Stream sensor data, KPI metrics, or log tails directly into a web UI.
  • Chat notifications: Push “user is typing” or “new message” alerts without opening a full WebSocket channel.
  • Progress monitoring: Show real‑time job progress for background tasks (e.g., Celery workers).
  • IoT telemetry: Feed lightweight device updates to a monitoring console.

Testing and Debugging SSE

Browser DevTools

Open the Network tab, filter by “event‑source”, and you’ll see the continuous stream. Look for:

  • Correct Content-Type: text/event-stream.
  • Absence of Transfer‑Encoding: chunked errors (some proxies mishandle chunked streams).
  • Heartbeat comments keeping the connection alive.

Command‑Line Testing

Use curl to verify the raw stream:

curl -N http://localhost:5000/stream

The -N flag disables buffering, letting you see each event as it arrives.

Scaling SSE with Multiple Workers

Because each SSE connection holds a thread (Flask) or an async task (FastAPI), you’ll eventually need a process manager:

  • Gunicorn with Gevent: Enables cooperative multitasking for Flask.
  • Uvicorn workers: Use uvicorn myapp:app --workers 4 for FastAPI.

For truly massive scale, consider a message broker (Redis

Comments

Leave a Reply

Your email address will not be published. Required fields are marked *

More posts