I'll walk you through implementing Server-Sent Events (SSE) in FastAPI from beginning to end, making it easy for beginners to understand.
Understanding SSE basics
Server-Sent Events (SSE) is a technology that allows one-way communication (server to client only), uses a simple text-based protocol with messages separated by double newlines, handles reconnection automatically in browsers, and is perfect for live updates like logs, tracking, and dashboard metrics.
Step 1: Setting up your FastAPI project
First, install the required packages:
pip install fastapi uvicornStep 2: Creating the basic FastAPI application
Create a new Python file (e.g., main.py) and set up the basic FastAPI application:
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
import uvicorn
# Initialize the FastAPI application
app = FastAPI()
# Add CORS middleware to allow requests from web browsers
app.add_middleware(
CORSMiddleware,
allow_origins=["*"], # For development; restrict in production
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
if __name__ == "__main__":
uvicorn.run(app, host="127.0.0.1", port=8000)Step 3: Creating the waypoints JSON file
Create a file named waypoints.json in the same directory with some sample coordinate data:
[
{"lat": 37.7749, "lng": -122.4194},
{"lat": 37.7750, "lng": -122.4195},
{"lat": 37.7751, "lng": -122.4196},
{"lat": 37.7752, "lng": -122.4197},
{"lat": 37.7753, "lng": -122.4198},
{"lat": 37.7754, "lng": -122.4199},
{"lat": 37.7755, "lng": -122.4200},
{"lat": 37.7756, "lng": -122.4201},
{"lat": 37.7757, "lng": -122.4202},
{"lat": 37.7758, "lng": -122.4203}
]Step 4: Implementing the SSE generator function
Add the generator function to your main.py file:
import json
from asyncio import sleep
from fastapi.responses import StreamingResponse
# Generator function that yields waypoints as SSE events
async def waypoints_generator():
# Open and load the waypoints JSON file
waypoints = open('waypoints.json')
waypoints = json.load(waypoints)
# Iterate through the first 10 waypoints
for waypoint in waypoints[0:10]:
# Convert waypoint to JSON string
data = json.dumps(waypoint)
# Format as SSE event with event type and data
yield f"event: locationUpdate\ndata: {data}\n\n"
# Wait 1 second between events
await sleep(1)Step 5: Creating the SSE endpoint
Add the endpoint to your FastAPI application:
@app.get("/get-waypoints")
async def root():
# Return a streaming response with our generator
# The media_type is critical for SSE to work properly
return StreamingResponse(
waypoints_generator(),
media_type="text/event-stream"
)Step 6: Creating the frontend HTML page
Create an index.html file:
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>Live Vehicle Coordinates</title>
<style>
body {
font-family: Arial, sans-serif;
margin: 20px;
}
#coordinates-container {
border: 1px solid #ccc;
padding: 20px;
border-radius: 5px;
}
h2 {
color: #333;
}
</style>
</head>
<body>
<div id="coordinates-container">
<h2>Live Vehicle Coordinates</h2>
<div id="coordinates"></div>
</div>
<script src="script.js"></script>
</body>
</html>Step 7: Implementing the frontend javaScript
Create a script.js file:
// Get the element where coordinates will be displayed
const coordinatesElement = document.getElementById('coordinates');
let coords;
// Create an EventSource pointing to our FastAPI endpoint
const eventSource = new EventSource('http://127.0.0.1:8000/get-waypoints');
// When the connection opens
eventSource.onopen = () => {
console.log('EventSource connected');
// Clear previous data when connection is established
coordinatesElement.innerText = '';
}
// Listen for 'locationUpdate' events
eventSource.addEventListener('locationUpdate', function(event) {
// Parse the JSON data from the event
coords = JSON.parse(event.data);
console.log('LocationUpdate', coords);
// Update the UI with the new coordinates
updateCoordinates(coords);
});
// Handle errors
eventSource.onerror = (error) => {
console.error('EventSource failed', error);
// Close the connection on error to prevent endless retry loops
eventSource.close();
}
// Function to update and display coordinates
function updateCoordinates(coordinates) {
// Create a new paragraph element for each coordinate
const paragraph = document.createElement('p');
paragraph.textContent = `Latitude: ${coordinates.lat}, Longitude: ${coordinates.lng}`;
// Add it to the coordinates container
coordinatesElement.appendChild(paragraph);
}Start your FastAPI server with python main.py, then open the HTML file in your browser. You should see coordinates appearing one by one with a 1-second delay between each.
Key concepts that make SSE work
On the server side, the StreamingResponse handles the streaming mechanism, setting media_type="text/event-stream" tells the browser this is an SSE connection, the \n\n at the end of each message separates events, and the event: locationUpdate line categorizes the event type.
On the client side, the EventSource object connects to the SSE endpoint, event listeners process different event types, the browser handles reconnection automatically, and errors can be managed through the onerror handler.
Troubleshooting common issues
If coordinates don't appear, check browser console for errors. Ensure CORS is properly configured if testing on different domains. Make sure the waypoints.json file exists and has valid JSON. Verify your server is running on the expected port (8000).
This guide covers everything you need to implement Server-Sent Events with Python FastAPI, from setting up the server to creating the client interface.
