Build a Real-Time Notification System with Server-Sent Events
Build a Real-Time Notification System with Server-Sent Events
Your users are waiting for that "new message" alert, but your current polling strategy is burning database CPU and draining battery life. Every few seconds, your frontend fires an HTTP request asking, "Is there anything new?" and the server replies, "Nope." Do this a thousand times a minute, and you’ve built a performance nightmare.
The solution isn’t always WebSockets, which introduce complex connection management and bi-directional state. For the specific case of pushing notifications from server to client, there’s a simpler, standardized, and often more efficient tool: Server-Sent Events (SSE).
SSE allows the server to push updates to the client over a single, long-lived HTTP connection. It’s unidirectional (server → client), lightweight, and works natively in browsers via the EventSource API. Let’s build a real-time notification system you can deploy today using Python.
Why SSE Beats Polling (and Sometimes WebSockets)
Before we write code, let’s clarify why SSE is the right choice here.
| Feature | HTTP Polling | WebSockets | Server-Sent Events (SSE) |
|---|---|---|---|
| Direction | Client → Server (request) | Bi-directional | Server → Client only |
| Connection | Short-lived, repeated | Persistent, complex | Persistent, simple HTTP |
| Browser Support | Universal | Universal | Native (EventSource) |
| Overhead | High (headers per request) | Medium (frame headers) | Low (chunked HTTP) |
| Complexity | Low | High | Low |
SSE is perfect for notifications because you rarely need the client to send data back over the same channel. You just need to push: "User X sent a message," "Your order shipped," or "System alert."
According to the MDN documentation, the browser initiates the connection using EventSource, and the server keeps it open to stream updates [15]. The server sets the Content-Type to text/event-stream and leaves the connection open until events are done or a timeout occurs [3].
Setting Up the Python Backend
We’ll use Flask and the flask-streaming pattern to create an SSE endpoint. While Flask doesn’t have native SSE support like NestJS’s @Sse() decorator [12], we can implement it easily using generators or the Response object with stream_with_context.
First, install the dependencies:
pip install flask
Now, let’s build the server. We need two endpoints:
-
/notifications/stream: The SSE endpoint where clients connect to listen. -
/notifications/send: A regular POST endpoint to trigger a new notification.
We’ll use a simple in-memory list to store active connections. In production, you’d swap this for Redis Pub/Sub to handle multiple server instances [2][6].
from flask import Flask, Response, request, stream_with_context
import json
import time
import threading
app = Flask(__name__)
# In-memory store for active client connections
# In production: use Redis Pub/Sub for scalability [2][6]
active_connections = []
# Lock to manage thread-safe connection updates
connection_lock = threading.Lock()
def generate_events():
"""Generator that streams events to connected clients."""
while True:
# Wait for new notifications (simulated with a short sleep)
# In a real app, you'd use a queue or Redis listener
time.sleep(0.5)
# Check if there are new notifications to send
# This is a simplified example; use a real queue for production
with connection_lock:
if active_connections:
# Simulate a notification event
event_data = {
"type": "notification",
"message": "New message received from Alice!",
"timestamp": time.time()
}
# Format as SSE: data: {...}\n\n
yield f"data: {json.dumps(event_data)}\n\n"
@app.route('/notifications/stream')
def stream_notifications():
"""SSE endpoint: Clients connect here to receive real-time updates."""
# Register this request as a subscriber
# Note: In a real app, you'd store the response object or a queue reference
# For this demo, we'll use a global generator approach
return Response(
stream_with_context(generate_events()),
mimetype='text/event-stream'
)
@app.route('/notifications/send', methods=['POST'])
def send_notification():
"""API endpoint to trigger a new notification."""
data = request.get_json()
message = data.get('message', 'New notification!')
# In a real app, you'd push this to a Redis queue or database
# and the SSE generator would listen to that queue [2][6]
print(f"Notification sent: {message}")
return {"status": "success", "message": message}, 200
if __name__ == '__main__':
# Run with threaded=True to handle multiple connections
app.run(port=5000, threaded=True)
This code sets the Content-Type to text/event-stream, which is required for SSE to work [3]. The generate_events function yields formatted strings in the SSE protocol: data: <json>\n\n.
Connecting the Frontend
The browser side is incredibly simple. You don’t need a heavy library like Socket.IO. Just use the native EventSource API [15].
Create an index.html file:
<!DOCTYPE html>
<html>
<head>
<title>SSE Notifications</title>
<style>
#notification-box {
border: 1px solid #ccc;
padding: 15px;
margin: 20px;
background: #f9f9f9;
min-height: 50px;
}
.toast {
background: #4a90e2;
color: white;
padding: 10px 15px;
margin: 5px 0;
border-radius: 4px;
animation: fadeIn 0.3s;
}
@keyframes fadeIn {
from { opacity: 0; transform: translateY(-10px); }
to { opacity: 1; transform: translateY(0); }
}
</style>
</head>
<body>
<h1>Real-Time Notifications (SSE)</h1>
<button onclick="sendNotification()">Send Test Notification</button>
<div id="notification-box"></div>
<script>
const notificationBox = document.getElementById('notification-box');
// Connect to the SSE endpoint
const eventSource = new EventSource('/notifications/stream');
eventSource.onmessage = function(event) {
const data = JSON.parse(event.data);
const toast = document.createElement('div');
toast.className = 'toast';
toast.textContent = data.message;
notificationBox.appendChild(toast);
// Auto-remove after 5 seconds
setTimeout(() => toast.remove(), 5000);
};
eventSource.onerror = function(err) {
console.error('EventSource failed:', err);
// Optional: Implement reconnection logic [2]
};
function sendNotification() {
fetch('/notifications/send', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ message: 'Hello from the frontend!' })
});
}
</script>
</body>
</html>
Open http://localhost:5000 in your browser. Click "Send Test Notification," and watch the toast appear instantly without a page reload.
Scaling for Production: What’s Missing?
The example above is a great starting point, but it’s not production-ready. Here’s what you need to add to make it robust:
1. Authentication & User Scoping
In a real app, you shouldn’t broadcast to everyone. You need to scope connections to authenticated users. Add a token or userId to the /notifications/stream?userId=123 URL and validate it before adding the connection to your list [2][4].
2. Redis Pub/Sub for Multi-Instance Scaling
If you run multiple server instances (e.g., in a Kubernetes cluster), the in-memory active_connections list won’t work across nodes. Use Redis Pub/Sub to publish notifications to a channel, and have each server instance subscribe to that channel to forward events to its local clients [2][6].
3. Heartbeats & Reconnection
Intermediaries like proxies or load balancers might close idle connections. Send a "ping" event every 30 seconds to keep the connection alive [4]. Also, implement a reconnection UI to show users if they
If you found this helpful, consider buying me a coffee ☕ — it keeps these articles coming!
Also check out my AI tools collection: AI 次元世界 — free AI tools for developers.
💡 Related: **Content Creator Ultimate Bundle (Save 33%)* — $29.99*
Top comments (0)