Flask与Django如何构建实时通知系统

发布时间：2024-11-14 19:02:13 作者：小樊
来源：亿速云阅读：79

构建实时通知系统是一个复杂但有趣的任务，可以使用Flask或Django这样的Python框架来实现。下面我将分别介绍如何使用Flask和Django来构建一个基本的实时通知系统。

使用Flask构建实时通知系统

Flask本身并不直接支持WebSocket，但可以通过集成Flask-SocketIO来实现实时通信。以下是使用Flask-SocketIO构建实时通知系统的步骤：

安装Flask-SocketIO和Flask-SQLAlchemy：

pip install flask-socketio flask-sqlalchemy

创建Flask应用：

from flask import Flask, render_template
from flask_socketio import SocketIO, emit
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///notifications.db'
db = SQLAlchemy(app)
socketio = SocketIO(app)

class Notification(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    message = db.Column(db.String(200), nullable=False)
    user_id = db.Column(db.Integer, nullable=False)

    def __repr__(self):
        return f'<Notification {self.message}>'

@app.route('/')
def index():
    notifications = Notification.query.all()
    return render_template('index.html', notifications=notifications)

@socketio.on('send_notification')
def handle_send_notification(data):
    notification = Notification(message=data['message'], user_id=data['user_id'])
    db.session.add(notification)
    db.session.commit()
    emit('notification_sent', {'message': notification.message}, room=str(notification.user_id))

if __name__ == '__main__':
    socketio.run(app, debug=True)

创建HTML模板（templates/index.html）：

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>Real-time Notifications</title>
    <script src="https://cdnjs.cloudflare.com/ajax/libs/socket.io/4.0.1/socket.io.js"></script>
    <script src="https://code.jquery.com/jquery-3.6.0.min.js"></script>
</head>
<body>
    <h1>Notifications</h1>
    <ul id="notifications"></ul>

    <script>
        $(document).ready(function() {
            var socket = io();
            socket.on('notification_sent', function(data) {
                $('#notifications').append('<li>' + data.message + '</li>');
            });

            $('#send_notification_form').submit(function(event) {
                event.preventDefault();
                var message = $('#message').val();
                var user_id = $('#user_id').val();
                socket.emit('send_notification', {message: message, user_id: user_id});
            });
        });
    </script>
</body>
</html>

使用Django构建实时通知系统

Django本身也不直接支持WebSocket，但可以通过集成Django Channels来实现实时通信。以下是使用Django Channels构建实时通知系统的步骤：

安装Django Channels和Django ORM：
```
pip install channels django
```

创建Django项目和应用：

django-admin startproject myproject
cd myproject
python manage.py startapp notifications

配置Django Channels：在myproject/settings.py中添加以下内容：

INSTALLED_APPS = [
    ...
    'channels',
    'notifications',
]

ASGI_APPLICATION = 'myproject.asgi.application'

CHANNEL_LAYERS = {
    'default': {
        'BACKEND': 'channels_redis.core.RedisChannelLayer',
        'CONFIG': {
            'hosts': [('127.0.0.1', 6379)],
        },
    },
}

创建ASGI文件（myproject/asgi.py）：

import os
from django.core.asgi import get_asgi_application
from channels.routing import ProtocolTypeRouter, URLRouter
from channels.auth import AuthMiddlewareStack
import notifications.routing

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myproject.settings')

application = ProtocolTypeRouter({
    "http": get_asgi_application(),
    "websocket": AuthMiddlewareStack(
        URLRouter(
            notifications.routing.websocket_urlpatterns
        )
    ),
})

创建WebSocket路由（notifications/routing.py）：

from django.urls import re_path
from . import consumers

websocket_urlpatterns = [
    re_path(r'ws/notifications/$', consumers.NotificationConsumer.as_asgi()),
]

创建WebSocket消费者（notifications/consumers.py）：

import json
from channels.generic.websocket import AsyncWebsocketConsumer

class NotificationConsumer(AsyncWebsocketConsumer):
    async def connect(self):
        self.user = self.scope['user']
        await self.accept()

    async def disconnect(self, close_code):
        pass

    async def receive(self, text_data):
        text_data_json = json.loads(text_data)
        message = text_data_json['message']
        await self.send(text_data=json.dumps({
            'message': message
        }))

创建Django视图和模板：在notifications/views.py中添加以下内容：

from django.shortcuts import render
from .models import Notification

def index(request):
    notifications = Notification.objects.all()
    return render(request, 'notifications/index.html', {'notifications': notifications})

创建模板文件notifications/templates/notifications/index.html：

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>Real-time Notifications</title>
    <script src="https://cdnjs.cloudflare.com/ajax/libs/socket.io/4.0.1/socket.io.js"></script>
    <script src="https://code.jquery.com/jquery-3.6.0.min.js"></script>
</head>
<body>
    <h1>Notifications</h1>
    <ul id="notifications"></ul>

    <script>
        $(document).ready(function() {
            var socket = io('ws://' + window.location.host + '/ws/notifications/');
            socket.on('message', function(data) {
                $('#notifications').append('<li>' + data.message + '</li>');
            });
        });
    </script>
</body>
</html>

配置URL路由（notifications/urls.py）：

from django.urls import path
from . import views

urlpatterns = [
    path('', views.index, name='index'),
]

配置项目URL路由（myproject/urls.py）：

from django.contrib import admin
from django.urls import path, include

urlpatterns = [
    path('admin/', admin.site.urls),
    path('', include('notifications.urls')),
]

通过以上步骤，你可以使用Flask或Django构建一个基本的实时通知系统。根据具体需求，你可以进一步扩展和优化这个系统。

Flask与Django如何构建实时通知系统

使用Flask构建实时通知系统

使用Django构建实时通知系统

相关阅读