Вызов функции AsyncWebsocketConsumer из команд

У меня есть этот скрипт, который получает сообщение и отправляет то же самое сообщение всем подключенным браузерам.

comsumer.py

import json

from channels.db import database_sync_to_async
from channels.generic.websocket import AsyncWebsocketConsumer

from .models import Message


class ChatConsumer(AsyncWebsocketConsumer):
  
    async def connect(self):
        self.room_group_name = 'chat'

        await self.channel_layer.group_add(
            self.room_group_name,
            self.channel_name
        )
        await self.accept()

    async def disconnect(self, close_code):
        await self.channel_layer.group_discard(
            self.room_group_name,
            self.channel_name
        )

    async def push(self,message):
        print("pushing message")
        await self.send(text_data=json.dumps({
            'message': message
        }))

    async def receive(self, text_data):

        text_data_json = json.loads(text_data)
        message = text_data_json['message']
      

        res = await self.save_message_to_db(message)
        await self.channel_layer.group_send(
            self.room_group_name,
            {
                'type': 'chat_message',
                'message': res['message_data'],
            }
        )

    async def chat_message(self, event):
        message = event['message']

        # WebSocketを介してメッセージを送信
        await self.send(text_data=json.dumps({
            'message': message
        }))

    @database_sync_to_async
    def save_message_to_db(self, message_text):

        message = Message.objects.create(content=message_text)
        return {
            'message_data': {
                'id': message.id,
                'content': message.content,
                'timestamp': message.timestamp.isoformat(),
            }
        }

И этот класс используется как

from django.urls import re_path

from . import consumers

websocket_urlpatterns = [
    re_path(r'ws/chat/', consumers.ChatConsumer.as_asgi()),
]

Теперь я хочу вызвать async def push() из представления или командного скрипта.

На данный момент я тестирую в management/commands/test_cmd.py

def handle(self, *args, **options):
    channel_layer = get_channel_layer()
    async_to_sync(channel_layer.group_send)(
        'public_room',
        {
            "type": "push",
            "message": "message"
        }
    )

Однако ничего не происходит.

Вернуться на верх