消息推送系统

# 示例代码：消息发送接口（Django）

from rest_framework.views import APIView

from rest_framework.response import Response

from .models import Message

from .serializers import MessageSerializer

class SendMessageView(APIView):

def post(self, request):

serializer = MessageSerializer(data=request.data)

if serializer.is_valid():

serializer.save()

return Response({"status": "success", "message": "Message sent successfully."})

return Response(serializer.errors, status=400)

# 示例代码：WebSocket消息接收（Django Channels）

from channels.generic.websocket import WebsocketConsumer

import json

from .models import Message

class ChatConsumer(WebsocketConsumer):

def connect(self):

self.accept()

def receive(self, text_data):

data = json.loads(text_data)

message = Message.objects.create(content=data['content'], sender=self.scope['user'])

self.send(text_data=json.dumps({'content': message.content}))

def disconnect(self, close_code):

pass

# 示例代码：消息查询接口（Django）

from rest_framework.views import APIView

from rest_framework.response import Response

from .models import Message

from .serializers import MessageSerializer

class GetMessageView(APIView):

def get(self, request):

user = request.user

messages = Message.objects.filter(receiver=user).order_by('-timestamp')

serializer = MessageSerializer(messages, many=True)

return Response(serializer.data)

# 示例代码：消息删除接口（Django）

from rest_framework.views import APIView

from rest_framework.response import Response

from .models import Message

class DeleteMessageView(APIView):

def delete(self, request, message_id):

try:

message = Message.objects.get(id=message_id)

message.delete()

return Response({"status": "success", "message": "Message deleted successfully."})

except Message.DoesNotExist:

return Response({"status": "error", "message": "Message not found."}, status=404)