[ django使用daphne框架开发聊天功能 ]

发表评论
55 views

点击查看channels框架文档

配置settings.py

INSTALLED_APPS = [
    "daphne",


ASGI_APPLICATION = "djangoblog.asgi.application"

CHANNEL_LAYERS = {
    "default": {
        "BACKEND": "channels_redis.core.RedisChannelLayer",
        "CONFIG": {
            "hosts": [REDIS_URL + "/1"],
        },
    },
}

asgi.py文件注意事项

这个django_asgi_app必须要放在， 导入的app前面执行

import os
from django.core.asgi import get_asgi_application
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "djangoblog.settings")
django_asgi_app = get_asgi_application()

# Initialize Django ASGI application early to ensure the AppRegistry
# is populated before importing code that may import ORM models.

from channels.routing import ProtocolTypeRouter
from channels.routing import ProtocolTypeRouter, URLRouter
from channels.security.websocket import AllowedHostsOriginValidator
from chat.routing import websocket_urlpatterns

application = ProtocolTypeRouter({
    "http": django_asgi_app,
    "websocket": AllowedHostsOriginValidator(
        URLRouter(websocket_urlpatterns)
    ),
})

我的chat App

1.路由routing.py文件， 我是传jwt 的token来接受

# chat/routing.py
from django.urls import re_path

from . import consumers

websocket_urlpatterns = [
    re_path(r"ws/chat/(?P<token>[\w\-\.]+)/$", consumers.ChatConsumer.as_asgi()),
]

2. models.py文件还未完善

# Create your models here.
from django.db import models
from django.utils.translation import gettext_lazy as _
from accounts.models import BlogUser as User

############
## Choices
############

class ToppingChoices(models.IntegerChoices):
    NO = 0, _("关闭")
    YES = 1, _("开启")

class DisturbChoices(models.IntegerChoices):
    NO = 0, _("关闭")
    YES = 1, _("开启")

class WithdrawChoices(models.IntegerChoices):
    NO = 0, _("未撤回")
    YES = 1, _("撤回")

class AckChoices(models.IntegerChoices):
    NO = 0, _("No")
    YES = 1, _("Yes")


############
## 基础模型
############
class BaseModel(models.Model):
    version = models.IntegerField(default=1)
    created = models.DateTimeField(auto_now_add=True)
    updated = models.DateTimeField(auto_now=True)
    deleted = models.BooleanField(default=False)

    class Meta:
        abstract = True  # 使这个模型成为抽象基类


############
## Models
############

class Message(BaseModel):
    # 消息
    content = models.TextField(verbose_name="消息内容")
    withdraw = models.SmallIntegerField(choices=WithdrawChoices.choices, default=0, verbose_name="撤回")
    sender = models.ForeignKey(User, on_delete=models.CASCADE, verbose_name="发送者")
    created_at = models.DateTimeField(auto_now_add=True, verbose_name="时间戳")
    ack = models.BooleanField(default=False, verbose_name="发送成功确认")

    class Meta:
        db_table = "message"


class GroupChat(BaseModel):
    # 群
    name = models.CharField(max_length=100, verbose_name="群名")
    announcement = models.CharField(max_length=254, verbose_name="群公告", default="")
    members = models.ManyToManyField(User, verbose_name="成员", through="GroupChatMembers")

    class Meta:
        db_table = "group_chat"


class GroupProfile(BaseModel):
    # 个人在群里的配置
    group = models.ForeignKey(GroupChat, on_delete=models.CASCADE, verbose_name="群组")
    remarks_group_name = models.CharField(max_length=50, verbose_name="群备注名", default="")
    remarks_user_name = models.CharField(max_length=50,verbose_name="我在群里的名字", default="")
    topping = models.BooleanField(verbose_name="群置顶", default=False)
    disturb = models.BooleanField(verbose_name="免打扰", default=False)

    class Meta:
        db_table = "group_profile"


class GroupChatMembers(BaseModel):
    groupchat = models.ForeignKey(GroupChat, on_delete=models.CASCADE)
    bloguser = models.ForeignKey(User, on_delete=models.CASCADE)


class Friendship(BaseModel):
    # 好友
    self = models.ForeignKey(User, on_delete=models.CASCADE, verbose_name="自己")
    friend = models.ForeignKey(User, related_name="friend",on_delete=models.CASCADE, verbose_name="好友")
    remarks_name = models.CharField(max_length=50,verbose_name="好友备注名", default="")
    topping = models.BooleanField(verbose_name="群置顶", default=False)
    disturb = models.BooleanField(verbose_name="免打扰", default=False)

    class Meta:
        db_table = "friendship"


class FriendshipOfflineMessage(BaseModel):
    # 好友离线消息
    # user = models.ForeignKey(User, on_delete=models.CASCADE, verbose_name="接受消息用户") 
    message = models.ForeignKey(Message, on_delete=models.CASCADE, verbose_name="消息")
    friendship = models.ForeignKey(Friendship, on_delete=models.CASCADE, verbose_name="好友关系id")
    # ack = models.SmallIntegerField(choices=AckChoices.choices, default=0, verbose_name="接收成功确认")

    class Meta:
        db_table = "friendship_offline_message"


class GroupChatOfflineMessage(BaseModel):
    # 群离线消息
    # user = models.ForeignKey(User, on_delete=models.CASCADE, verbose_name="接受消息用户")# 这个字段应该没用
    message = models.ForeignKey(Message, on_delete=models.CASCADE, verbose_name="消息")
    group = models.ForeignKey(GroupChat, on_delete=models.CASCADE, verbose_name="接受消息群")
    # ack = models.SmallIntegerField(choices=AckChoices.choices, default=0, verbose_name="接收成功确认")

    class Meta:
        db_table = "group_chat_offline_message"

3. repo.py 数据库异步封装

# 封装对数据库的访问。
from channels.db import database_sync_to_async
from accounts.models import BlogUser
from .models import GroupChat, FriendshipOfflineMessage, GroupChatOfflineMessage, Message, Friendship


class UserRepo:
    @staticmethod
    @database_sync_to_async
    def get_user(user_id):
        return BlogUser.objects.get(id=user_id)

    @staticmethod
    @database_sync_to_async
    def get_group_member_ids(group_id:int):
        group_chat = GroupChat.objects.get(id=group_id)
        member_ids = group_chat.members.all().values_list("id", flat=True)
        return member_ids

    @staticmethod
    @database_sync_to_async
    def get_group_obj_menbers(group: GroupChat):
        return list(group.members.all().values("id", "username", "email"))


class MessageRepo:

    @staticmethod
    @database_sync_to_async
    def create_message(content: str, sender_id: int):
        return Message.objects.create(content=content, sender_id=sender_id)


    @staticmethod
    @database_sync_to_async
    def ack_message(msg_ids: int):
        return Message.objects.filter(id__in=msg_ids).update(ack=1)


class FriendshipOfflineMsgRepo:

    @staticmethod
    @database_sync_to_async
    def get_friendship_messges_query(friendship_id: int):
        return list(FriendshipOfflineMessage.objects.filter(friendship_id=friendship_id).select_related("message").order_by("id"))


    @staticmethod
    @database_sync_to_async
    def create_friendship(msg_obj: Message, friendship_id: int):
        """
        Args:
            user_id (int): 接受用户
            msg_obj (Message): 消息
            friendship_id (int): 当前聊天窗口id
        """
        return FriendshipOfflineMessage.objects.create(message=msg_obj, friendship_id=friendship_id)


class FriendshipRepo:

    @staticmethod
    @database_sync_to_async
    def get_friendships_query(user_id: int):
        return list(Friendship.objects.filter(self_id=user_id).select_related("self", "friend").order_by("id"))

    @staticmethod
    @database_sync_to_async
    def create_friendship(data: dict):
        """创建好友关系"""
        return Friendship.objects.create(**data)


class GroupOfflineMsgRepo:

    @staticmethod
    @database_sync_to_async
    def get_group_messages(group_id: int):
        return list(GroupChatOfflineMessage.objects.filter(group=group_id).select_related("message").order_by("id"))


    @staticmethod
    @database_sync_to_async
    def create_group_offline_message(msg_obj: Message, group_id: int):
        """
        Args:
            user_id (int): 接收用户
            msg_obj (Message): 消息
            group_id (int): 当前群组id
        """
        return GroupChatOfflineMessage.objects.create(message=msg_obj, group_id=group_id)


class GroupRepo:
    @staticmethod
    @database_sync_to_async
    def get_user_groups(user_id: int):
        return list(GroupChat.objects.filter(members__id=user_id))

    @staticmethod
    @database_sync_to_async
    def add_member(user_id: int, group_id: int):
        """群组添加成员"""
        user: BlogUser = UserRepo.get_user(user_id)
        group: GroupChat = GroupRepo.get_group(group_id)
        group.members.add(user)
        return group

    @staticmethod
    @database_sync_to_async
    def create_group_chat(data: dict):
        menbers = data.pop("members")
        blog_users = BlogUser.objects.filter(id__in=menbers)
        group_chat =  GroupChat.objects.create(**data)
        group_chat.members.set(blog_users)
        return group_chat

    @staticmethod
    @database_sync_to_async
    def get_group(group_id: int):
        return GroupChat.objects.get(id=group_id)

4. schemas.py数据模板定义字典的类型限制

检查用户发送的send的数据， 根据method来执行不同的处理函数

#############
## TypedDict
#############

class MessageRequestSchema(BaseModel):
    msg: str = ""
    method: Literal["send_friend", "send_group", "get_friend", "get_group", "ack"]
    id: int = 0 # 该聊天的id
    friend_id: int = 0 # friend时候的好友id
    ack_ids: list[int] = []


class MessageRespSchema(BaseModel):
    code: int
    msg: str
    data: Any

class SuccessRespMessage(MessageRespSchema):
    code: int
    msg: str = "ok"

class ErrorRespMessage(MessageRespSchema):
    code: int
    data: str = ""
    msg: Any

5. 主要的consumer.py 连接websocket连接处理的类

# chat/consumers.py
import logging
from .schemas import SuccessRespMessage, ErrorRespMessage, MessageRequestSchema
from channels.generic.websocket import AsyncWebsocketConsumer
from .models import GroupChat, FriendshipOfflineMessage, GroupChatOfflineMessage, Message, Friendship
from pydantic import ValidationError
from accounts.models import BlogUser
from django.core.cache import cache
from asgiref.sync import sync_to_async
from common.jwt import check_account_token
from .repo import UserRepo, GroupRepo, MessageRepo, FriendshipRepo, GroupOfflineMsgRepo, FriendshipOfflineMsgRepo


logger = logging.getLogger(__name__)

CACHE_TIMEOUT = 3600 


class ChatConsumer(AsyncWebsocketConsumer):
    user: BlogUser = None
    async def connect(self):
        token = self.scope["url_route"]["kwargs"]["token"]
        payload = check_account_token(token)
        self.user = await UserRepo.get_user(int(payload["user_id"]))
        if not self.user:
            await self.send_error_message(data=ErrorRespMessage(code=1, msg="用户未登录"))
            await self.close()
            return 

        await self.cache_user_info()
        await self.accept()
        await self.init_connect()

    async def cache_user_info(self):
        """缓存用户信息"""
        self.user_info = {
            "user_id": self.user.id,
            "channel_name": self.channel_name
        }
        await sync_to_async(cache.set)(self.user.id, self.channel_name, CACHE_TIMEOUT)

    async def get_user_channel_name(self, user_id):
        user_channel = await sync_to_async(cache.get)(user_id)
        return user_channel

    async def get_user_channel_names(self, user_ids):
        user_channels = await sync_to_async(cache.get_many)(user_ids)
        return user_channels
    async def disconnect(self, close_code):
        await sync_to_async(cache.delete)(self.user.id)

    async def get_my_friendship_messages(self, friendship_id):
        """获取当前好友消息"""
        friend_offline_msg_queryset = await FriendshipOfflineMsgRepo.get_friendship_messges_query(friendship_id)
        result = []
        for offline_msg in friend_offline_msg_queryset:
            sender = "other"
            if offline_msg.message.sender_id == self.user.pk: # 消息接受者是自己, 是对方发送给我的消息
                sender = "me"
            result.append({
                "id": offline_msg.pk,
                "msg": offline_msg.message.content,
                "msg_id": offline_msg.message.id,
                "sender":sender,
                "send_type": "friend",
                "ack": offline_msg.message.ack
            })
        return result

    async def get_my_group_messages(self, group_id):
        """获取当前群组消息"""
        group_offline_msg_queryset = await GroupOfflineMsgRepo.get_group_messages(group_id=group_id)
        result = []
        for offline_msg in group_offline_msg_queryset:
            sender = "other"
            if offline_msg.message.sender_id == self.user.pk: # 消息接受者是自己, 是对方发送给我的消息
                sender = "me"
            result.append({
                "id": offline_msg.pk,
                "msg": offline_msg.message.content,
                "msg_id": offline_msg.message.id,
                "sender":sender,
                "send_type": "group",
                "ack": offline_msg.message.ack
            })
        return result

    async def init_connect(self):
        """用户连接时，发送初始化信息"""
        friendship_result = await self.get_my_friendships()
        group_result =await self.get_my_group_chats()
        await self.send(SuccessRespMessage(data=friendship_result+group_result, code=-2).model_dump_json())

    async def get_my_friendships(self):
        friendship_queryset: list[Friendship] =  await FriendshipRepo.get_friendships_query(self.user.pk)
        result = []
        for friendship in friendship_queryset:
            result.append({
                "id": friendship.id,
                "self_id": friendship.self.id,
                "friend_id": friendship.friend.id,
                "name": friendship.friend.username,
                "email": friendship.friend.email,
                "remarks_name": friendship.remarks_name,
                "topping": friendship.topping,
                "disturb": friendship.disturb,
                "created": friendship.created.strftime("%Y-%m-%d %H:%M:%S"),
                "updated": friendship.updated.strftime("%Y-%m-%d %H:%M:%S"),
                "type": "friend"
            })
        return result

    async def get_my_group_chats(self):
        group_chats = await GroupRepo.get_user_groups(self.user.pk)
        result = []
        for group in group_chats:
            result.append({
                "id": group.id,
                "name": group.name,
                "announcement": group.announcement,
                "members": await UserRepo.get_group_obj_menbers(group),
                "created": group.created.strftime("%Y-%m-%d %H:%M:%S"),
                "updated": group.updated.strftime("%Y-%m-%d %H:%M:%S"),
                "type": "group"
            })
        return result

    async def receive(self, text_data):
        ## 数据验证
        try:
            data = MessageRequestSchema.model_validate_json(text_data)
        except ValidationError as e:
            error_info = e.errors()
            msg = {i["loc"][0]:i["msg"] for i in error_info} if error_info[0]["loc"] else error_info[0]["msg"]
            await self.send_error_message(data=ErrorRespMessage(code=1, msg=msg))
            return 

        ## 获取消息
        if data.method == "get_friend":
            result = await self.get_my_friendship_messages(data.id)
            await self.channel_layer.send(self.channel_name, {"type": "send.messages", "data": result})
            return 

        elif data.method == "get_group":
            result = await self.get_my_group_messages(data.id)
            await self.channel_layer.send(self.channel_name, {"type": "send.messages", "data": result})
            return 

        ## 消息确认
        if data.method == "ack":
            await MessageRepo.ack_message(data.ack_ids)
            return

        ## 发送消息
        if data.method == "send_friend":
            await self.send_friendship_message(data)

        ## 群聊发送消息
        elif data.method == "send_group":
            await self.send_group_message(data)

    async def send_friendship_message(self, data: MessageRequestSchema):
        msg_obj = await MessageRepo.create_message(content=data.msg, sender_id=self.user.id)
        friendship_offline_msg = await FriendshipOfflineMsgRepo.create_friendship(msg_obj=msg_obj, friendship_id=data.id)
        channel_name = await self.get_user_channel_name(data.friend_id)
        # 如果在线给好友发
        data_dict = data.model_dump()
        if channel_name:
            data_dict["sender"] = "other"
            await self.channel_layer.send(
                channel_name, {"type": "send.message", "data": data_dict}
            )

        # 给自己发
        data_dict["sender"] = "me"
        await self.channel_layer.send(
            self.channel_name, {"type": "send.message", "data": data_dict}
        )


    async def send_group_message(self, data: MessageRequestSchema):
            msg_obj = await MessageRepo.create_message(content=data.msg, sender_id=self.user.id)
            group_offline_msg = await GroupOfflineMsgRepo.create_group_offline_message(msg_obj=msg_obj, group_id=data.id)
            member_ids = await UserRepo.get_group_member_ids(group_id=data.id)
            channel_name_dict = await self.get_user_channel_names(user_ids=member_ids)

            data_dict = data.model_dump()
            for channel_name in channel_name_dict.values():
                data_dict["sender"] = "other"
                if channel_name == self.channel_name:
                    data_dict["sender"] = "me"
                await self.channel_layer.send(channel_name, {"type": "send.message", "data": data_dict})


    async def ask_offline_message(message_id: int, send_to: int, send_type: int):
        """确认离线消息
            message_id: 消息id
            send_to: [ 好友离线id, 群组离线id ]
            send_type: [ 好友离线消息, 群组离线消息 ]
        """
        if send_type == 0:
            await sync_to_async(FriendshipOfflineMessage.objects.filter(id=send_to).update)(ack=1)
        elif send_type == 1:
            await sync_to_async(GroupChatOfflineMessage.objects.filter(id=send_to).update)(ack=1)

    async def send_message(self, event):
        data = event.get("data", None)
        await self.send(text_data=SuccessRespMessage(data=data, code=0).model_dump_json())

    async def send_messages(self, event):
        data = event.get("data", None)
        await self.send(text_data=SuccessRespMessage(data=data, code=-1).model_dump_json())


    async def send_user_info(self, event):
        data = event.get("data", None)
        await self.send(text_data=SuccessRespMessage(data=data, code=-2).model_dump_json())

    async def send_error_message(self, data: ErrorRespMessage):
        logger.error(f"user_id: {self.user.pk}, username: {self.user.username}, msg: {data.msg}")
        await self.send(text_data=data.model_dump_json())

启动命令

daphne -p 8000 djangoblog.asgi:application