Django Channels: ilova ochiq turgan foydalanuvchiga xabar bir soniyada

uv add channels channels-redis uvicorn

config/
├── settings.py
├── urls.py
└── asgi.py          ← WebSocket marshrutlash shu yerda
apps/alerts/
├── __init__.py
├── models.py        ← Alert modeli
├── consumers.py     ← WebSocket "view"lari
├── routing.py       ← WebSocket URL'lari
├── signals.py       ← saqlanganda broadcast
├── apps.py
└── tests.py

# config/settings.py
INSTALLED_APPS = [
    "daphne",  # runserver'ni ASGI rejimga o'tkazadi — ro'yxat BOSHIDA tursin
    # ...
    "channels",
    "apps.alerts",
]

ASGI_APPLICATION = "config.asgi.application"

# Channel layer — jarayonlar orasidagi "pochta": worker'lar ko'p bo'lsa ham
# xabar hammaga yetadi. Production'da doim Redis, testda InMemory.
CHANNEL_LAYERS = {
    "default": {
        "BACKEND": "channels_redis.core.RedisChannelLayer",
        "CONFIG": {"hosts": [("127.0.0.1", 6379)]},
    }
}

# config/asgi.py
import os

from django.core.asgi import get_asgi_application

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "config.settings")
# MUHIM: get_asgi_application() import'lardan OLDIN chaqiriladi —
# aks holda "Apps aren't loaded yet" xatosi chiqadi
django_asgi_app = get_asgi_application()

from channels.routing import ProtocolTypeRouter, URLRouter  # noqa: E402

from apps.alerts.routing import websocket_urlpatterns  # noqa: E402

application = ProtocolTypeRouter(
    {
        "http": django_asgi_app,  # oddiy Django o'z yo'lida davom etadi
        "websocket": URLRouter(websocket_urlpatterns),
    }
)

# apps/alerts/models.py
from django.db import models


class Alert(models.Model):
    class Level(models.TextChoices):
        INFO = "info", "Ma'lumot"
        WARNING = "warning", "Ogohlantirish"
        CRITICAL = "critical", "Jiddiy"

    title = models.CharField(max_length=200)
    body = models.TextField(blank=True)
    level = models.CharField(max_length=8, choices=Level.choices, default=Level.INFO)
    line = models.ForeignKey(
        "metro.Line",
        null=True,
        blank=True,
        on_delete=models.CASCADE,
        help_text="Bo'sh qolsa — butun tarmoq bo'yicha ogohlantirish",
    )
    is_active = models.BooleanField(default=True)
    created_at = models.DateTimeField(auto_now_add=True)

# apps/alerts/consumers.py
from channels.generic.websocket import AsyncJsonWebsocketConsumer

ALERTS_GROUP = "alerts"


class AlertsConsumer(AsyncJsonWebsocketConsumer):
    async def connect(self):
        # har ulanishning o'z kanali (self.channel_name) bor;
        # uni umumiy guruhga qo'shamiz — broadcast shu guruh orqali
        await self.channel_layer.group_add(ALERTS_GROUP, self.channel_name)
        await self.accept()

    async def disconnect(self, code):
        await self.channel_layer.group_discard(ALERTS_GROUP, self.channel_name)

    # group_send'dagi {"type": "alert_created"} shu metodga keladi:
    # type'dagi nuqta/pastki chiziq -> metod nomi
    async def alert_created(self, event):
        await self.send_json(event["payload"])

# apps/alerts/routing.py
from django.urls import re_path

from .consumers import AlertsConsumer

websocket_urlpatterns = [
    re_path(r"^ws/alerts/$", AlertsConsumer.as_asgi()),
]

# apps/alerts/signals.py
from asgiref.sync import async_to_sync
from channels.layers import get_channel_layer
from django.db.models.signals import post_save
from django.dispatch import receiver

from .consumers import ALERTS_GROUP
from .models import Alert


@receiver(post_save, sender=Alert)
def broadcast_alert(sender, instance: Alert, created: bool, **kwargs):
    if not created or not instance.is_active:
        return
    channel_layer = get_channel_layer()
    # signal sinxron kontekstda ishlaydi, channel layer esa async —
    # ko'prik: async_to_sync
    async_to_sync(channel_layer.group_send)(
        ALERTS_GROUP,
        {
            "type": "alert_created",  # -> AlertsConsumer.alert_created
            "payload": {
                "id": instance.id,
                "title": instance.title,
                "body": instance.body,
                "level": instance.level,
                "line_id": instance.line_id,
            },
        },
    )

# apps/alerts/apps.py
from django.apps import AppConfig


class AlertsConfig(AppConfig):
    name = "apps.alerts"

    def ready(self):
        from . import signals  # noqa: F401  — signal ulanishi uchun import shart

# 1-terminal: ASGI server
python manage.py runserver

# 2-terminal: WebSocket mijoz
python -m websockets ws://127.0.0.1:8000/ws/alerts/
# endi admin'dan Alert qo'shing — JSON shu yerga bir zumda tushadi

# apps/alerts/tests.py
import pytest
from channels.layers import get_channel_layer
from channels.testing import WebsocketCommunicator

from apps.alerts.consumers import ALERTS_GROUP, AlertsConsumer


@pytest.mark.asyncio
async def test_connected_client_receives_group_broadcast(settings):
    # testda Redis shart emas — xotiradagi layer yetadi
    settings.CHANNEL_LAYERS = {"default": {"BACKEND": "channels.layers.InMemoryChannelLayer"}}
    communicator = WebsocketCommunicator(AlertsConsumer.as_asgi(), "/ws/alerts/")
    connected, _ = await communicator.connect()
    assert connected

    channel_layer = get_channel_layer()
    await channel_layer.group_send(
        ALERTS_GROUP,
        {"type": "alert_created", "payload": {"title": "Chilonzor: uzilish"}},
    )

    message = await communicator.receive_json_from()
    assert message["title"] == "Chilonzor: uzilish"
    await communicator.disconnect()

uv add --dev pytest-asyncio
uv run pytest apps/alerts/ -q