Introducing Server-Despatched Occasions in Python | In the direction of Information Science

a developer, I’m all the time on the lookout for methods to make my purposes extra dynamic and interactive. Customers at this time anticipate real-time options, similar to reside notifications, streaming updates, and dashboards that refresh robotically. The instrument that always involves thoughts for net builders when contemplating a lot of these purposes is WebSockets, and it’s extremely highly effective. 

There are occasions, although, when WebSockets will be overkill, and their full performance is usually not required. They supply a fancy, bi-directional communication channel, however many occasions, all I want is for the server to push updates to the shopper. For these frequent situations, a extra easy and stylish answer that’s constructed proper into fashionable net platforms is called Server-Despatched Occasions (SSE).

On this article, I’m going to introduce you to Server-Despatched Occasions. We’ll talk about what they’re, how they evaluate to WebSockets, and why they’re usually the right instrument for the job. Then, we’ll dive right into a sequence of sensible examples, utilizing Python and the FastAPI framework to construct real-time purposes which are surprisingly easy but highly effective.

What are Server-Despatched Occasions (SSE)?

Server-Despatched Occasions is an online know-how customary that permits a server to push information to a shopper asynchronously as soon as an preliminary shopper connection has been established. It offers a one-way, server-to-client stream of knowledge over a single, long-lived HTTP connection. The shopper, usually an online browser, subscribes to this stream and might react to the messages it receives.

Some key facets of Server-Despatched Occasions embody:

• Easy Protocol. SSE is a simple, text-based protocol. Occasions are simply chunks of textual content despatched over HTTP, making them simple to debug with customary instruments like curl.
• Normal HTTP. SSE works over common HTTP/HTTPS. This implies it’s typically extra suitable with present firewalls and proxy servers.
• Computerized Reconnection. It is a killer characteristic. If the connection to the server is misplaced, the browser’s EventSource API will robotically attempt to reconnect. You get this resilience free of charge, with out writing any further JavaScript code.
• One-Method Communication. SSE is strictly for server-to-client information pushes. For those who want full-duplex, client-to-server communication, WebSockets are the extra acceptable alternative.
• Native Browser Help. All fashionable net browsers have built-in assist for Server-Despatched Occasions (SSE) by the EventSource interface, eliminating the necessity for client-side libraries.

Why SSE Issues/Frequent Use Instances

The first benefit of SSE is its simplicity. For a big class of real-time issues, it offers all the required performance with a fraction of the complexity of WebSockets, each on the server and the shopper. This implies sooner growth, simpler upkeep, and fewer issues that may go fallacious.

SSE is an ideal match for any situation the place the server must provoke communication and ship updates to the shopper. For instance …

• Stay Notification Methods. Pushing notifications to a consumer when a brand new message arrives or an necessary occasion happens.
• Actual-Time Exercise Feeds. Streaming updates to a consumer’s exercise feed, just like a Twitter or Fb timeline.
• Stay Information Dashboards. Sending steady updates for inventory tickers, sports activities scores, or monitoring metrics to a reside dashboard.
• Streaming Log Outputs. Displaying the reside log output from a long-running background course of immediately within the consumer’s browser.
• Progress Updates. Exhibiting the real-time progress of a file add, a knowledge processing job, or some other long-running activity initiated by the consumer.

That’s sufficient idea; let’s see simply how simple it’s to implement these concepts with Python.

Organising the Improvement Setting

We’ll utilise FastAPI, a contemporary and high-performance Python net framework. Its native assist for asyncio and streaming responses makes it an ideal match for implementing Server-Despatched Occasions. You’ll additionally want the Uvicorn ASGI server to run the appliance.

As traditional, we’ll arrange a growth setting to maintain our initiatives separate. I recommend utilizing MiniConda for this, however be happy to make use of whichever instrument you’re accustomed to.

# Create and activate a brand new digital setting
(base) $ conda create -n sse-env python=3.13 -y
(base) $ activate sse-env

Now, set up the exterior libraries we want.

# Set up FastAPI and Uvicorn
(sse-env) $ pip set up fastapi uvicorn

That’s all of the setup we want. Now, we will begin coding.

Code Instance 1 — The Python Backend. A Easy SSE Endpoint

Let’s create our first SSE endpoint. It would ship a message with the present time to the shopper each second.

Create a file named app.py and sort the next into it.

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from fastapi.middleware.cors import CORSMiddleware
import time

app = FastAPI()

# Permit requests from http://localhost:8080 (the place index.html is served)
app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:8080"],
    allow_methods=["GET"],
    allow_headers=["*"],
)

def event_stream():
    whereas True:
        yield f"information: The time is {time.strftime('%X')}nn"
        time.sleep(1)

@app.get("/stream-time")
def stream():
    return StreamingResponse(event_stream(), media_type="textual content/event-stream")

I hope you agree that this code is simple.

1. We outline an event_stream() perform. This loop repeats endlessly, producing a string each second.
2. The yielded string is formatted based on the SSE spec: it should begin with information: and finish with two newlines (nn).
3. Our endpoint /stream-time returns a StreamingResponse, passing our generator to it and setting the media_type to textual content/event-stream. FastAPI handles the remainder, maintaining the connection open and sending every yielded chunk to the shopper.

To run the code, don’t use the usual Python app.py command as you’d usually. As a substitute, do that.

(sse-env)$ uvicorn app:app --reload

INFO:     Will look ahead to adjustments in these directories: ['/home/tom']
INFO:     Uvicorn working on http://127.0.0.1:8000 (Press CTRL+C to stop)
INFO:     Began reloader course of [4109269] utilizing WatchFiles
INFO:     Began server course of [4109271]
INFO:     Ready for utility startup.
INFO:     Utility startup full.

Now, sort this deal with into your browser …

http://127.0.0.1:8000/stream-time

… and you need to see one thing like this.

The display ought to show an up to date time document each second.

Code instance 2. Actual-Time System Monitoring Dashboard

On this instance, we’ll monitor our PC or laptop computer’s CPU and reminiscence utilization in actual time.

Right here is the app.py code you want.

import asyncio
import json
import psutil
from fastapi import FastAPI, Request
from fastapi.responses import HTMLResponse, StreamingResponse
from fastapi.middleware.cors import CORSMiddleware
import datetime

# Outline app FIRST
app = FastAPI()

# Then add middleware
app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:8080"],
    allow_methods=["GET"],
    allow_headers=["*"],
)

async def system_stats_generator(request: Request):
    whereas True:
        if await request.is_disconnected():
            print("Consumer disconnected.")
            break

        cpu_usage = psutil.cpu_percent()
        memory_info = psutil.virtual_memory()

        stats = {
            "cpu_percent": cpu_usage,
            "memory_percent": memory_info.p.c,
            "memory_used_mb": spherical(memory_info.used / (1024 * 1024), 2),
            "memory_total_mb": spherical(memory_info.complete / (1024 * 1024), 2)
        }

        yield f"information: {json.dumps(stats)}nn"
        await asyncio.sleep(1)

@app.get("/system-stats")
async def stream_system_stats(request: Request):
    return StreamingResponse(system_stats_generator(request), media_type="textual content/event-stream")

@app.get("/", response_class=HTMLResponse)
async def read_root():
    with open("index.html") as f:
        return HTMLResponse(content material=f.learn())

This code constructs a real-time system monitoring service utilizing the FastAPI net framework. It creates an online server that repeatedly tracks and broadcasts the host machine’s CPU and reminiscence utilization to any linked net shopper.

First, it initialises a FastAPI utility and configures Cross-Origin Useful resource Sharing (CORS) middleware. This middleware is a safety characteristic that’s explicitly configured right here to permit an online web page served from http://localhost:8080 to make requests to this server, which is a typical requirement when the frontend and backend are developed individually.

The core of the appliance is the system_stats_generator asynchronous perform. This perform runs in an infinite loop, and in every iteration, it makes use of the psutil library to fetch the present CPU utilisation share and detailed reminiscence statistics, together with the share used, megabytes used, and complete megabytes. It packages this data right into a dictionary, converts it to a JSON string, after which yields it within the particular “textual content/event-stream” format (information: …nn). 

Using asyncio.sleep(1) introduces a one-second pause between updates, stopping the loop from consuming extreme sources. The perform can also be designed to detect when a shopper has disconnected and gracefully cease sending information for that shopper.

The script defines two net endpoints. The @app.get(“/system-stats”) endpoint creates a StreamingResponse that initiates the system_stats_generator. When a shopper makes a GET request to this URL, it establishes a persistent connection, and the server begins streaming the system stats each second. The second endpoint, @app.get(“/”), serves a static HTML file named index.html as the primary web page. This HTML file would usually include the JavaScript code wanted to connect with the /system-stats stream and dynamically show the incoming efficiency information on the internet web page.

Now, right here is the up to date (index.html) front-end code.

    
    
    


    

        
        
        

            
Reminiscence Utilization
            
            
0% (0 / 0 MB)
        
    
    

Run the app utilizing Uvicorn, as we did in Instance 1. Then, in a separate command window, sort the next to begin a Python server.

python3 -m http.server 8080

Now, open the URL http://localhost:8080/index.html in your browser, and you will note the output, which ought to replace repeatedly.

Code instance 3 — Background activity progress bar

On this instance, we provoke a activity and show a bar indicating the duty’s progress.

Up to date app.py

import asyncio
import json
import psutil
from fastapi import FastAPI, Request
from fastapi.responses import HTMLResponse, StreamingResponse
from fastapi.middleware.cors import CORSMiddleware
import datetime

# Outline app FIRST
app = FastAPI()

# Then add middleware
app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:8080"],
    allow_methods=["GET"],
    allow_headers=["*"],
)

async def training_progress_generator(request: Request):
    """
    Simulates a long-running AI coaching activity and streams progress.
    """
    total_epochs = 10
    steps_per_epoch = 100

    for epoch in vary(1, total_epochs + 1):
        # Simulate some preliminary processing for the epoch
        await asyncio.sleep(0.5)

        for step in vary(1, steps_per_epoch + 1):
            # Verify if shopper has disconnected
            if await request.is_disconnected():
                print("Consumer disconnected, stopping coaching activity.")
                return

            # Simulate work
            await asyncio.sleep(0.02)

            progress = (step / steps_per_epoch) * 100
            simulated_loss = (1 / epoch) * (1 - (step / steps_per_epoch)) + 0.1

            progress_data = {
                "epoch": epoch,
                "total_epochs": total_epochs,
                "progress_percent": spherical(progress, 2),
                "loss": spherical(simulated_loss, 4)
            }

            # Ship a named occasion "progress"
            yield f"occasion: progressndata: {json.dumps(progress_data)}nn"

    # Ship a last "full" occasion
    yield f"occasion: completendata: Coaching full!nn"

@app.get("/stream-training")
async def stream_training(request: Request):
    """SSE endpoint to stream coaching progress."""
    return StreamingResponse(training_progress_generator(request), media_type="textual content/event-stream")

@app.get("/", response_class=HTMLResponse)
async def read_root():
    """Serves the primary HTML web page."""
    with open("index.html") as f:
        return HTMLResponse(content material=f.learn())

The up to date index.html code is that this.

Cease your present uvicorn and Python server processes in the event that they’re nonetheless working, after which restart each.

Now, if you open the index.html web page, you need to see a display with a button. Urgent the button will begin a dummy activity, and a shifting bar will show the duty progress.

Code Instance 4— A Actual-Time Monetary Inventory Ticker

For our last instance, we’ll create a simulated inventory ticker. The server will generate random worth updates for a number of inventory symbols and ship them utilizing named occasions, the place the occasion identify corresponds to the inventory image (e.g., occasion: AAPL, occasion: GOOGL). It is a highly effective sample for multiplexing completely different varieties of knowledge over a single SSE connection, permitting the shopper to deal with every stream independently.

Right here is the up to date app.py code you’ll want.

import asyncio
import json
import random
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
from fastapi.middleware.cors import CORSMiddleware

# Step 1: Create app first
app = FastAPI()

# Step 2: Add CORS to permit requests from http://localhost:8080
app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:8080"],
    allow_methods=["GET"],
    allow_headers=["*"],
)

# Step 3: Simulated inventory costs
STOCKS = {
    "AAPL": 150.00,
    "GOOGL": 2800.00,
    "MSFT": 300.00,
}

# Step 4: Generator to simulate updates
async def stock_ticker_generator(request: Request):
    whereas True:
        if await request.is_disconnected():
            break

        image = random.alternative(checklist(STOCKS.keys()))
        change = random.uniform(-0.5, 0.5)
        STOCKS[symbol] = max(0, STOCKS[symbol] + change)

        replace = {
            "image": image,
            "worth": spherical(STOCKS[symbol], 2),
            "change": spherical(change, 2)
        }

        # Ship named occasions so the browser can hear by image
        yield f"occasion: {image}ndata: {json.dumps(replace)}nn"
        await asyncio.sleep(random.uniform(0.5, 1.5))

# Step 5: SSE endpoint
@app.get("/stream-stocks")
async def stream_stocks(request: Request):
    return StreamingResponse(stock_ticker_generator(request), media_type="textual content/event-stream")

And the up to date index.html

Cease then restart the uvicorn and Python processes as earlier than. This time, if you open http://localhost:8080/index.html in your browser, you need to see a display like this, which can regularly replace the dummy costs of the three shares.

Abstract

On this article, I demonstrated that for a lot of real-time use instances, Server-Despatched Occasions supply a less complicated various to WebSockets. We mentioned the core rules of SSE, together with its one-way communication mannequin and computerized reconnection capabilities. By a sequence of hands-on examples utilizing Python and FastAPI, we noticed simply how simple it’s to construct highly effective real-time options. We coated:

• A easy Python back-end and SSE endpoint
• A reside system monitoring dashboard streaming structured JSON information.
• An actual-time progress bar for a simulated long-running background activity.
• A multiplexed inventory ticker utilizing named occasions to handle completely different information streams.

Subsequent time it is advisable to push information out of your server to a shopper, I encourage you to pause earlier than reaching for WebSockets. Ask your self for those who really want bi-directional communication. If the reply isn’t any, then Server-Despatched Occasions are seemingly the extra easy, sooner, and extra strong answer you’ve been on the lookout for.