Server-Sent Events with Python FastAPI

SSE communication model

Many times when we have a requirement to send continuous stream of data from sever to the client we have majorly three options available of which are pooling the data at the client side, web-sockets or server-sent events(SSE). Each one of them have its own pros & cons. Pooling can put unnecessary load on server when there is no data to be sent whereas websockets can be harder to scale & at the same time SSE is unidirectional from server to client only.

SSE in a nutshell —

1. It is unidirectional in nature meaning only server can send data to frontend unlike traditional request-response architecture, where the client can send the data & receives response from server.
2. SSE is a pure text based protocol where everything is transfered in pure text format ending with new line indicator(\n\n) to distinguish between new events. Every message can be segregated into different type of events according to which the client can consume the message appropriately.
3. SSE includes automatic reconnection mechanisms, so if the connection is lost, the client will attempt to reconnect to the server without requiring explicit intervention.

Use cases —

1. displaying live log information in frontend as new logs created in backend
2. updating live locations of tracking objects in frontend by getting the updated coordinates from backend
3. Updating informations or graphs on real time basis on dashboards

Backend implementation in FastAPI —

Below I have a simple JSON file from which every 1 second I want to stream one coordinate to the client using FastAPI.

[
    {
        "lat": 22.09769,
        "lng": 87.24068
    },
    {
        "lat": 22.09776,
        "lng": 87.24075
    },
    {
        "lat": 22.09784,
        "lng": 87.24082
    },
    {
        "lat": 22.09811,
        "lng": 87.24098
    }
]

from fastapi import FastAPI, Response
from fastapi.responses import StreamingResponse
from fastapi.middleware.cors import CORSMiddleware
import json, uvicorn
from asyncio import sleep

app = FastAPI()

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

async def waypoints_generator():
    waypoints = open('waypoints.json')
    waypoints = json.load(waypoints)
    for waypoint in waypoints[0: 10]:
        data = json.dumps(waypoint)
        yield f"event: locationUpdate\ndata: {data}\n\n"
        await sleep(1)

@app.get("/get-waypoints")
async def root():
    return StreamingResponse(waypoints_generator(), media_type="text/event-stream")


if __name__ == "__main__":
    uvicorn.run(app, host="127.0.0.1", port=8000)

Here I have a generator function waypoints_generator() which yields a response of one coordinate from the JSON file every 1 second which is then streamed to the client side.

Two things to consider majorly here is that, in order to identify as an SSE response, the content type of the response should be “text/event-stream” and every response should end with \n\n as the response of SSE is always a pure text, so to identify response properly this distinction is a must.

I have used the StreamingResponse response type from FastAPI here which takes an generator function as an argument to stream the output in batches instead of a single response.

Since the response is a text response, so every event in the response should be separated by a new line of \n. There should be minimum one event for e.g a default event of data can be made available.

FrontEnd —

To work with server-sent events JS have the EventSource API for it.

The EventSource interface is web content's interface to server-sent events.

An EventSource instance opens a persistent connection to an HTTP server, which sends events in text/event-stream format. The connection remains open until closed by calling EventSource.close().

Once the connection is opened, incoming messages from the server are delivered to your code in the form of events. If there is an event field in the incoming message, the triggered event is the same as the event field value. If no event field is present, then a generic message event is fired.

Here I am just updating the received coordinates part of the SSE response into an HTML doc.

<!DOCTYPE html>
<html lang="en">

<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <link rel="stylesheet" href="style.css">
    <title>Live Vehicle Coordinates</title>
</head>

<body>
    <div id="coordinates-container">
        <h2>Live Vehicle Coordinates</h2>
        <div id="coordinates"></div>
    </div>
</body>
<script src="script.js"></script>
</html>

const coordinatesElement = document.getElementById('coordinates');
let coords;

// Create EventSource for SSE endpoint
const eventSource = new EventSource('http://127.0.0.1:8000/get-waypoints');

eventSource.onopen = () => {
    console.log('EventSource connected')
    //Everytime the connection gets extablished clearing the previous data from UI
    coordinatesElement.innerText = ''
}

//eventSource can have event listeners based on the type of event.
//Bydefault for message type of event it have the onmessage method which can be used directly or this same can be achieved through explicit eventlisteners
eventSource.addEventListener('locationUpdate', function (event) {
    coords = JSON.parse(event.data);
    console.log('LocationUpdate', coords);
    updateCoordinates(coords)
});

//In case of any error, if eventSource is not closed explicitely then client will retry the connection a new call to backend will happen and the cycle will go on.
eventSource.onerror = (error) => {
    console.error('EventSource failed', error)
    eventSource.close()
}

// Function to update and display coordinates
function updateCoordinates(coordinates) {
    // Create a new paragraph element for each coordinate and append it
    const paragraph = document.createElement('p');
    paragraph.textContent = `Latitude: ${coordinates.lat}, Longitude: ${coordinates.lng}`;
    coordinatesElement.appendChild(paragraph);
}

After creating the eventSource object, we can attach various eventListeners on the object based on the event on which we want to subscribe to. If no event is available, a generic onmessage method can be used to get the response.

Note — In case of any error or the server closing the connection after the response if completed, the client will auto retry and connect to server once again and the loop will continue. So in case of error, it can be handled and if required the eventSource needs to be closed explicitly to avoid retry.

Sample response image from PostMan

In case of SSE, browsers always try to keep the connection alive by retrying the connection in case of any failures, connection clousures form server or intermittent network issues etc. At the same time when browser gets closed the server can also close the connection thus by saving resource. The retry interval can be set from server side by using the retry event keyword.

The full source code can be found here — https://github.com/NandaGopal56/server-sent-events-fastapi.git

Conclusion —

SSE is a very good protocol for use cases where unidirectional communication is required thus by avoiding more resource heavy protocols like web-sockets. But through unit testing is key to achieving the best results.

resources —

EventSource - Web APIs | MDN

Thank you everyone for contributing your valuable time on this. Please correct me if any mistakes, comment your thoughts or any suggestions and I am always happy to learn from you.