使用Postman测试Django Channels WebSocket时类型错误的解决方法

问题场景 #

在Django项目中使用Channels实现WebSocket功能，期望创建MyClass对象时将其名称发送至名为test_consumer_group_1的消费者组。

模型代码 #

class MyClass(models.Model):
    name = models.CharField(max_length=128, unique=True)
    members = models.ManyToManyField("Employee")
    def save(self, *args, **kwargs):
        super().save(*args,**kwargs)
        channel_layer = get_channel_layer()
        data = {"current_obj":self.name}
        async_to_sync(channel_layer.group_send)(
            "test_consumer_group_1",{
                'type':'send_notification',
                'value':json.dumps(data)
            }
    )

消费者代码 #

class TestConsumer(WebsocketConsumer):
    def connect(self):
        self.room_name="test_consumer"
        self.room_group_name = "test_consumer_group_1"
        async_to_sync(self.channel_layer.group_add)(
            self.channel_name, 
            self.room_group_name
        )
        self.accept()
        print('connected..')
        self.send(text_data=json.dumps({'status':'connected'}))

    def recieve(self, text_data):
        print(text_data)

    def disconnect(self, *args, **kwargs):
        print('disconnected')

    def send_notification(self, event):
        print("send_notification called")
        print(event)  

测试错误信息 #

raise TypeError(self.invalid_name_error.format("Group", name))
TypeError: Group name must be a valid unicode string with length < 100 containing only ASCII alphanumerics, hyphens, underscores, or periods, not specific.be2251de4bb647c1988845bd460d6971!564c92a792634237bcdba63290554557
WebSocket DISCONNECT /ws/test/ [127.0.0.1:35480]

错误原因 #

group_add方法的参数顺序完全颠倒：Channels规定group_add的第一个参数是要加入的组名称，第二个参数是当前消费者的通道名称。你现在把包含特殊字符的self.channel_name当成组名传入，违反了组名的格式规则，因此触发错误。

另外注意：消费者中的recieve是拼写错误，正确写法是receive，不修正的话后续无法处理客户端发送的消息。

修复后的消费者代码 #

class TestConsumer(WebsocketConsumer):
    def connect(self):
        self.room_name="test_consumer"
        self.room_group_name = "test_consumer_group_1"
        # 调换group_add的参数顺序：组名称在前，通道名称在后
        async_to_sync(self.channel_layer.group_add)(
            self.room_group_name,
            self.channel_name
        )
        self.accept()
        print('connected..')
        self.send(text_data=json.dumps({'status':'connected'}))

    # 修正拼写错误：recieve -> receive
    def receive(self, text_data):
        print(text_data)

    def disconnect(self, *args, **kwargs):
        print('disconnected')
        # 断开连接时移除组，避免资源泄漏
        async_to_sync(self.channel_layer.group_discard)(
            self.room_group_name,
            self.channel_name
        )

    def send_notification(self, event):
        print("send_notification called")
        print(event)
        # 将组消息推送给客户端
        self.send(text_data=event['value'])

额外优化建议 #

• 在disconnect方法中调用group_discard，将当前通道从组中移除，避免无效连接占用组资源。
• 在send_notification方法中添加self.send，把收到的组消息推送给客户端，实现完整的通知功能。

内容的提问来源于stack exchange，提问作者Waleed Farrukh