SSE, or Server-Sent Events, is a web technology that allows a server to push real-time, one-way updates to a client (usually a web browser) over a single, long-lived HTTP connection. It’s ideal for scenarios where the client needs to receive continuous streams of data, like stock tickers, live sports scores, or news feeds, without constantly polling the server.
The counterpart to real-time streaming (like SSE or WebSockets) is polling (or more specifically, long polling for a slightly more “real-time” feel than simple polling), which is the traditional model of HTTP communication. It’s worth noting WebSockets is also real-time streaming but bidirectional.
Server-Side Python Example (using Flask)
This example uses the Flask web framework to create a simple SSE endpoint.
import time
from flask import Flask, Response, render_template_string
app = Flask(__name__)
# A simple HTML page to host our SSE client
HTML_PAGE = """
<!DOCTYPE html>
<html>
<head>
<title>SSE Client</title>
</head>
<body>
<h1>Server-Sent Events Example</h1>
<div id="messages"></div>
<script>
const eventSource = new EventSource('/event_stream');
const messagesDiv = document.getElementById('messages');
eventSource.onopen = function() {
console.log("SSE connection opened.");
const p = document.createElement('p');
p.textContent = "Connection established.";
messagesDiv.appendChild(p);
};
eventSource.onmessage = function(event) {
console.log("Received message:", event.data);
const p = document.createElement('p');
p.textContent = `Message from server: ${event.data}`;
messagesDiv.appendChild(p);
};
eventSource.onerror = function(error) {
console.error("EventSource failed:", error);
const p = document.createElement('p');
p.textContent = `Error: ${error.message || 'Unknown error'}`;
messagesDiv.appendChild(p);
eventSource.close(); // Close on error, EventSource might try to reconnect
};
eventSource.addEventListener('custom_event', function(event) {
console.log("Received custom event:", event.data);
const p = document.createElement('p');
p.textContent = `Custom event from server: ${event.data}`;
messagesDiv.appendChild(p);
});
// Optional: Listen for the 'close' event (not standard EventSource, but useful for debugging)
// eventSource.addEventListener('close', function() {
// console.log("SSE connection closed.");
// });
</script>
</body>
</html>
"""
@app.route('/')
def index():
return render_template_string(HTML_PAGE)
@app.route('/event_stream')
def event_stream():
def generate_events():
i = 0
while True:
# Send a regular message
data = f"Server time: {time.ctime()} (message {i})"
yield f"data: {data}\n\n"
print(f"Sent: {data}")
# Send a custom event every 3 messages
if i % 3 == 0 and i > 0:
custom_data = f"This is a custom event {i/3}"
yield f"event: custom_event\ndata: {custom_data}\n\n"
print(f"Sent custom_event: {custom_data}")
i += 1
time.sleep(2) # Send an event every 2 seconds
return Response(generate_events(), mimetype='text/event-stream')
if __name__ == '__main__':
# Run the Flask app. You can access it at http://127.0.0.1:5000/
# The SSE stream will be available at http://127.0.0.1:5000/event_stream
app.run(debug=True, port=5000)In this example, '/event_stream' is the SSE URL.
Client-Side Python Example (using requests and sseclient)
While SSE is primarily designed for browser clients, you can also consume SSE streams from Python using libraries. The requests library is good for basic HTTP, but for SSE, sseclient is more suitable as it handles the event parsing and connection management.
import requests
from sseclient import SSEClient # You might need to install this: pip install sseclient
# The SSE URL from our Flask example (assuming it's running)
SSE_URL = "http://127.0.0.1:5000/event_stream"
print(f"Connecting to SSE URL: {SSE_URL}")
try:
# Create an SSEClient instance
# The 'stream=True' is crucial for requests to keep the connection open
messages = SSEClient(SSE_URL)
# Iterate over the incoming messages
for msg in messages:
# Each 'msg' object has attributes like 'data', 'event', 'id', 'retry'
print(f"Received (event: {msg.event}): {msg.data}")
# You can also check for specific event types
if msg.event == 'custom_event':
print(f" --> This was a CUSTOM EVENT: {msg.data}")
except requests.exceptions.ConnectionError as e:
print(f"Error connecting to SSE server: {e}")
print("Make sure the Flask SSE server (sse_server.py) is running.")
except Exception as e:
print(f"An unexpected error occurred: {e}")
print("Client finished.")
Naixian Zhang 