Introducing events sent to the Inpython server | Looking at the data science

nimda August 4, 2025

0 3 9 minutes read

Introducing events sent to the Inpython server | Looking at the data science

Developer, I always look for ways to make my apps powerful and work together. Users today are waiting for real-time features, such as live notifications, broadcasting reviews, and automatic renewal dashboards. The tool that often climbs in the mind of web developers when processing these types of applications with WebSockets, and are very powerful.

There are times, however, when the crockets can be more, and their full function is often unwanted. They provide a complex communication channel, but many times, everything I need for pushing updates above the client. Through these familiar conditions, the exact and elegant remedy built directly from modern Web platforms are known as events sent to the server (SSE).

In this article, I will introduce events sent to the server. We will discuss what they are, how they compare with the WebSockets, and why it is usually the appropriate tool for work. Then, we will enter into a series of practical examples, using the python and the Fastapapi frame to create real easy time programs but are powerful.

What are the events sent by server?

Seven events are a web technology that allows the server to press the data from the Asynchronously client if the first customer's connection has been established. It provides a single-way broadcast, for data customer server over one, lasting connection. Client, usually a web browser, subscribed to this distribution and respond to the instructions available.

Some important features of the server sent server includes:

Simple protocol. SSE is a direct protocol, based on the text. Events is just a chunks of text sent over HTTP, making it easier to pollute usual tools like the curl.
Standard http. SSE serves on top of HTTP / Normal HTTPS. This means that it is usually very compatible with existing firewall and proxy servers.
Recycling automatic. This is a risk factor. When the connection to the server is lost, the Dukeke Shopour Source API will try re-contact. You get this stronger FREE, without writing any additional JavaScript code.
One-way communication. SSE is strong in Server-to-Client Data. If you need a full connection of dumplex, customer communication-to-server, websokets appropriate choices.
Support for traditional browser. All modern web browsers set out in the built-in events (SSE) by Intercrenger Source, eliminating the need for customer libraries.

Why the SSEs Dide / Commendo Appeals

The main opportunity of SSE is its easy. In the main class of real-time problems, it provides the necessary operations and part of the difficulty of websites, both on the server and the client. This means quick improvement, simple correction, and few things that can go well.

SSE is a suitable thing in any situation where the server needs to start communicating and send updates to the client. For example …

Live notifications programs. Pressing notifications to the user when a new message comes or an important event takes place.
Supply of real-time jobs. Spreading user feed updates, such as Twitter or Facebook time line.
Dash Desert in Live Data. Sending Continuous updates of Stickers of Stickers, Games Schools, or Methodicism States in Live Dashboard.
Streaming outgoing output. Displaying live out from the background process running long in the user's browser.
Revival of progress. Real-time progress to load the file, data processing work, or any longer operational work established.

That's a simple idea; Let's see how easy it is to use these ideas with Python.

Setting up the environmental environment

We will use Fastapi, a Python Perfect Web draft. Its traditional support of Asyncio and the broadcasting answers makes a total game for using server events. You will also need the Uvicorn ASGI server to use the program.

As usual, we will place the developmental environment for the storage of our projects. I suggest using the minicondo of this, but feel free to use any tool that is affected.

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

Now, enter the outdoor libraries we need.

# Install FastAPI and Uvicorn
(sse-env) $ pip install fastapi uvicorn

That's the setup we need. Now, we can start installing codes.

An example of code 1 – Python Backend. SSE EDPOINT EDPOINT

Let's build our first fate of SSE. Will send current message to client every second.

Create a file called UPP.PY and type the following.

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

app = FastAPI()

# Allow requests from  (where index.html is served)
app.add_middleware(
    CORSMiddleware,
    allow_origins=[""],
    allow_methods=["GET"],
    allow_headers=["*"],
)

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

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

I hope you agree that this code is straight.

We describe the_the_stream () work. This loop repeats endless, producing string every second.
The allowed string is setup according to SSE Spec: It should start with Data: and end with two newlines ( n n).
Our ENDPOINT TIME / DISTION TIME returns the broadcast, exceeding our generator to and placing Media_type in the EXTY / EXY / EXTER-stream-stream. The Fastapi treats all rest, keeping the communication open and one export allowed on the client.

Using the code, do not use normal Python App.py order as you were unusual. Instead, do this.

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

INFO:     Will watch for changes in these directories: ['/home/tom']
INFO:     Uvicorn running on  (Press CTRL+C to quit)
INFO:     Started reloader process [4109269] using WatchFiles
INFO:     Started server process [4109271]
INFO:     Waiting for application startup.
INFO:     Application startup complete.

Now, type this address in your browser …

/stream-time

… and you should see something like this.

Photo by the writer

The screen must show a revised time record every second.

An example of code 2. The actual dashboard is a real time

In this example, we will look at our PP or CPU CPU and Memoria use in real time.

Here are App.py code you need.

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

# Define app FIRST
app = FastAPI()

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

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

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

        stats = {
            "cpu_percent": cpu_usage,
            "memory_percent": memory_info.percent,
            "memory_used_mb": round(memory_info.used / (1024 * 1024), 2),
            "memory_total_mb": round(memory_info.total / (1024 * 1024), 2)
        }

        yield f"data: {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="text/event-stream")

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

This code creates a real-time system monitoring service using the Fastapi web draft. It creates a continuous web server continuously and broadcasting the hosting machine CPU and memory on any connected Web Client.

First, it launches the Fastapi app and prepares the allocation of resources (CORTS). This dreamdler is a feature of safety prepared here to allow the webpage to work from that request, the common requirement when rebuilt.

The site of the application is a system_stats_Gegerator system asynchronut to work. This work is running out of endless loop, and each contalite, using the PSSU library to download the current CPU usage and used memory statistics, including percentage used, and megabytes used. It includes this information in the dictionary, changing it into JSON's string, and then shows it in a specific format saying “Text-Samsst” (Data: … n n).

The use of asyncio.Sleep (1) introduces one second suspension between updates, to prevent loop to eat additional resources. The work is also designed to find that the client has crossed and graciously to send the data to that customer.

The script describes two web endpoints. @ App.get (“/ system-stats”) endpoint creates starting starting starting program_stats_geegerator. When the client apples for this URL, it creates persistent communication, and the server starts to spread system statistics every second. The second endpoint, @ @ App.Tet (“/”), working with the Index.html file as the main page. This HTML file generally contain the JavaScript code required to connect to / system-stats-Stats-Stats Stream and displayed with incoming operating data on the Web page.

Now, here is the Front-End Code End (Index (Index.html)




    
    System Monitor
    


    
        
        
        
            Memory Usage
            
            0% (0 / 0 MB)

Run the app using Uvicorn, as we do by example 1. Then, in a different command window, type the following to start the Python server.

python3 -m http.server 8080

Now, open the URL /intex.html in your browser, and you will see the outgoing, which should update continuously.

Example of Code 3 – Better of Background Service Bar

In this example, we start work and show the bar to show the progress of the work.

Updated 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

# Define app FIRST
app = FastAPI()

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

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

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

        for step in range(1, steps_per_epoch + 1):
            # Check if client has disconnected
            if await request.is_disconnected():
                print("Client disconnected, stopping training task.")
                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": round(progress, 2),
                "loss": round(simulated_loss, 4)
            }

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

    # Send a final "complete" event
    yield f"event: completendata: Training complete!nn"

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

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

Index and Revised Index.html is.




    
    Live Task Progress

Stop your Uvicorn and Python Server processes when they run, and restart both.

Now, when you open the Histob.Html page, you should see the screen with the button. Pressing the button will start the dummy function, and the moving bar will show the progress of the work.

Example of code 4- General Financial Financial Study

Our final example, we will build a stock ticker made. The server will generate random price updates about several stock signs and send them using events called the word, where the event is in the stock sign (eg AAPL, event: An event: Google). This is a powerful pattern of different types of data by connecting one SSE, which allows the client to carry each river independently.

Here is a new app. The new investigation will require.

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 allow requests from 
app.add_middleware(
    CORSMiddleware,
    allow_origins=[""],
    allow_methods=["GET"],
    allow_headers=["*"],
)

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

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

        symbol = random.choice(list(STOCKS.keys()))
        change = random.uniform(-0.5, 0.5)
        STOCKS[symbol] = max(0, STOCKS[symbol] + change)

        update = {
            "symbol": symbol,
            "price": round(STOCKS[symbol], 2),
            "change": round(change, 2)
        }

        # Send named events so the browser can listen by symbol
        yield f"event: {symbol}ndata: {json.dumps(update)}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="text/event-stream")

And a revised index.html




    
    Live Stock Ticker

Stand and restart Uvicorn and Python processes as before. In the meantime, when you open an Epa /Ndex.html in your browser, you must see a screen like this, which will continue to reset three stock prices.

Summary

In this article, I showed that on many real-time charges, events sent to the server provides another easy way to websockets. We discussed the important principles of SSE, including one-way model and automatic re-connecting skills. With a series of emelons of Hands-on using Python and Fastapi, we saw how easy it is to build real powerful. Cover:

Python Back-End-End-End Endoind
The distribution of a demeanor dashboard formal is JSON's data.
Real-time progress bar for a long running work.
The solid stock ticker uses events called the iNentrelsions to manage different data streams.

Next time you need to push data from your server to client, I encourage you to stand a little before accessing WebSockets. Ask yourself if you really need the direction of the direction. If the answer is not, then the interns sent to the server may be directly straight, the fastest quick solution you want.

Source link

nimda August 4, 2025

0 3 9 minutes read