Flask Notifications
Real-time notifications have become an essential feature in modern web applications. From social media alerts to order updates in e-commerce platforms, notifications keep users engaged and informed about relevant events. In this tutorial, we'll learn how to implement real-time notifications in Flask using WebSockets.
Prerequisites
Before you begin, make sure you have:
- Basic knowledge of Flask
- Understanding of WebSockets (covered in previous sections)
- Python 3.7+ installed
- Flask and Flask-SocketIO installed
If you haven't installed Flask-SocketIO yet, you can do so with pip:
pip install flask flask-socketio
Understanding Real-Time Notifications
Traditional web applications require users to refresh the page to see new information. With WebSockets, we can push notifications to users instantly whenever an event occurs.
Here's what we'll cover in this tutorial:
- Setting up a Flask application with SocketIO
- Creating notification components
- Sending notifications to specific users
- Handling notification interactions
- Building a complete notification system
Setting Up Your Flask Application
Let's start by creating a basic Flask application with Flask-SocketIO:
from flask import Flask, render_template, session
from flask_socketio import SocketIO, emit
app = Flask(__name__)
app.config['SECRET_KEY'] = 'your-secret-key'
socketio = SocketIO(app, cors_allowed_origins="*")
@app.route('/')
def index():
return render_template('index.html')
if __name__ == '__main__':
socketio.run(app, debug=True)
Creating the HTML Structure
Next, let's create a basic HTML template with a notification area. Create a templates
folder and add 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>Flask Notifications</title>
<style>
.notification-container {
position: fixed;
top: 20px;
right: 20px;
width: 300px;
z-index: 1000;
}
.notification {
background-color: #f8f9fa;
border-left: 4px solid #007bff;
margin-bottom: 10px;
padding: 15px;
box-shadow: 0 2px 4px rgba(0,0,0,0.2);
transition: all 0.3s ease;
opacity: 1;
}
.notification.success {
border-left-color: #28a745;
}
.notification.warning {
border-left-color: #ffc107;
}
.notification.error {
border-left-color: #dc3545;
}
.notification-title {
font-weight: bold;
margin-bottom: 5px;
}
.notification-body {
font-size: 0.9em;
}
.notification-close {
float: right;
cursor: pointer;
font-weight: bold;
}
</style>
</head>
<body>
<h1>Flask Notifications Demo</h1>
<button id="sendNotification">Send Test Notification</button>
<div class="notification-container" id="notification-container">
<!-- Notifications will be appended here -->
</div>
<script src="https://cdnjs.cloudflare.com/ajax/libs/socket.io/4.0.1/socket.io.js"></script>
<script>
const socket = io();
// Socket.IO event listeners will go here
document.getElementById('sendNotification').addEventListener('click', function() {
socket.emit('test_notification', {
title: 'Test Notification',
message: 'This is a test notification from the client.',
type: 'info'
});
});
</script>
</body>
</html>
Implementing WebSocket Notifications
Now, let's add the WebSocket event handlers to our Flask app to handle notifications:
# Add these handlers to your Flask application
@socketio.on('connect')
def handle_connect():
print('Client connected')
@socketio.on('disconnect')
def handle_disconnect():
print('Client disconnected')
@socketio.on('test_notification')
def handle_test_notification(data):
# Broadcast the notification to all clients (including sender)
emit('notification', {
'title': data['title'],
'message': data['message'],
'type': data['type']
}, broadcast=True)
Now let's add the JavaScript code to receive and display notifications in our HTML file. Add this inside the <script>
tag after the Socket.IO initialization:
// Add this to your script section in index.html
socket.on('notification', function(data) {
const notification = document.createElement('div');
notification.className = `notification ${data.type}`;
const closeBtn = document.createElement('span');
closeBtn.className = 'notification-close';
closeBtn.innerHTML = '×';
closeBtn.onclick = function() {
notification.style.opacity = '0';
setTimeout(() => notification.remove(), 300);
};
const title = document.createElement('div');
title.className = 'notification-title';
title.textContent = data.title;
const body = document.createElement('div');
body.className = 'notification-body';
body.textContent = data.message;
notification.appendChild(closeBtn);
notification.appendChild(title);
notification.appendChild(body);
document.getElementById('notification-container').appendChild(notification);
// Auto-remove notification after 5 seconds
setTimeout(() => {
notification.style.opacity = '0';
setTimeout(() => notification.remove(), 300);
}, 5000);
});
User-Specific Notifications
In a real application, you often need to send notifications to specific users. Let's enhance our system to support user-specific notifications:
# Add these imports and modify your Flask application
from flask import Flask, render_template, session, request
from flask_socketio import SocketIO, emit, join_room, leave_room
from flask_login import current_user, login_required
# Add these routes and handlers
@app.route('/login/<username>')
def login(username):
session['username'] = username
return f"Logged in as {username}. <a href='/'>Go to home</a>"
@socketio.on('connect')
def handle_connect():
if 'username' in session:
username = session['username']
# Use the username as the room name
join_room(username)
print(f"User {username} connected and joined room {username}")
@app.route('/send_notification/<username>')
def send_user_notification(username):
socketio.emit('notification', {
'title': 'New Message',
'message': f'This notification is specifically for {username}',
'type': 'success'
}, room=username)
return f"Notification sent to {username}"
Building a Complete Notification System
Now, let's build a more complete system that stores notifications and allows users to mark them as read. First, we'll add a simple in-memory storage for notifications:
# Add this to your Flask application
notifications_store = {}
@app.route('/notifications')
def get_notifications():
username = session.get('username')
if not username:
return {'notifications': []}
user_notifications = notifications_store.get(username, [])
return {'notifications': user_notifications}
@app.route('/send_notification_to/<username>')
def send_notification_to_user(username):
message = request.args.get('message', 'Default notification message')
title = request.args.get('title', 'Notification')
notification_type = request.args.get('type', 'info')
# Store notification
if username not in notifications_store:
notifications_store[username] = []
notification_id = len(notifications_store[username]) + 1
notification = {
'id': notification_id,
'title': title,
'message': message,
'type': notification_type,
'read': False,
'timestamp': datetime.now().isoformat()
}
notifications_store[username].append(notification)
# Send real-time notification if user is online
socketio.emit('notification', notification, room=username)
return {'success': True, 'notification_id': notification_id}
@app.route('/mark_notification_read/<int:notification_id>')
def mark_notification_read(notification_id):
username = session.get('username')
if not username or username not in notifications_store:
return {'success': False, 'error': 'User not found'}
for notification in notifications_store[username]:
if notification['id'] == notification_id:
notification['read'] = True
return {'success': True}
return {'success': False, 'error': 'Notification not found'}
Real-World Example: Chat Application Notifications
Let's implement notifications in a chat application context:
# Add this to your Flask application
from datetime import datetime
# Mock database for users and chat messages
users_db = {}
chat_messages = []
@app.route('/register/<username>')
def register(username):
if username in users_db:
return f"Username {username} is already taken"
users_db[username] = {
'username': username,
'online': False,
'last_seen': None
}
return f"User {username} registered. <a href='/login/{username}'>Login</a>"
@socketio.on('connect')
def handle_connect():
if 'username' in session:
username = session['username']
users_db[username]['online'] = True
users_db[username]['last_seen'] = datetime.now().isoformat()
join_room(username)
# Notify other users that this user is online
for user in users_db:
if user != username and users_db[user]['online']:
socketio.emit('notification', {
'title': 'User Online',
'message': f"{username} is now online",
'type': 'info'
}, room=user)
@socketio.on('disconnect')
def handle_disconnect():
if 'username' in session:
username = session['username']
users_db[username]['online'] = False
users_db[username]['last_seen'] = datetime.now().isoformat()
leave_room(username)
@socketio.on('send_message')
def handle_message(data):
if 'username' not in session:
return
sender = session['username']
recipient = data['recipient']
message = data['message']
timestamp = datetime.now().isoformat()
# Store message
chat_messages.append({
'sender': sender,
'recipient': recipient,
'message': message,
'timestamp': timestamp,
'read': False
})
# Send real-time notification if recipient is online
if recipient in users_db and users_db[recipient]['online']:
socketio.emit('notification', {
'title': f'New Message from {sender}',
'message': message[:50] + ('...' if len(message) > 50 else ''),
'type': 'success',
'sender': sender,
'timestamp': timestamp
}, room=recipient)
socketio.emit('new_message', {
'sender': sender,
'message': message,
'timestamp': timestamp
}, room=recipient)
return {'success': True}
Update the JavaScript in your HTML file to handle these new events:
// Additional JavaScript for the chat notification example
socket.on('new_message', function(data) {
// Create or update the chat window for this sender
let chatWindow = document.getElementById('chat-' + data.sender);
if (!chatWindow) {
// Create new chat window logic here
console.log('New message from', data.sender);
}
// Add message to the chat window
const message = document.createElement('div');
message.className = 'message received';
message.textContent = data.message;
// You would append this to the actual chat window in a real app
console.log('Message content:', data.message);
});
function sendMessage(recipient, message) {
socket.emit('send_message', {
recipient: recipient,
message: message
});
}
Summary
In this tutorial, we've learned how to implement real-time notifications in Flask using WebSockets. We've covered:
- Setting up a Flask application with Flask-SocketIO
- Creating the UI for displaying notifications
- Implementing basic notification functionality
- Sending user-specific notifications
- Building a more complete notification system
- Implementing notifications in a chat application context
Real-time notifications significantly enhance user experience by keeping users informed about relevant events without requiring them to refresh the page. This is particularly valuable for social networks, messaging applications, collaborative tools, and e-commerce platforms.
Additional Resources
Exercises
- Notification Preferences: Extend the application to allow users to configure which types of notifications they want to receive.
- Notification History: Create a page where users can view their notification history with filtering options.
- Desktop Notifications: Add support for browser desktop notifications using the Notification API.
- Read Receipts: Implement read receipts for notifications so senders can know when their messages have been seen.
- Notification Badges: Add notification badges to your application's navigation bar to show the count of unread notifications.
By completing these exercises, you'll have a robust notification system that you can integrate into any Flask application!
If you spot any mistakes on this website, please let me know at [email protected]. I’d greatly appreciate your feedback! :)