Tutorial: Django Channels — WebSockets en tiempo real

WebSockets con Django Channels

Django Channels extiende Django para manejar WebSockets, protocolos de larga duración y tareas async.

Instalación

pip install channels channels-redis

Configuración

# settings.py
INSTALLED_APPS = [
    'daphne',
    'channels',
    # ... rest
]

ASGI_APPLICATION = 'config.asgi.application'
CHANNEL_LAYERS = {
    'default': {
        'BACKEND': 'channels_redis.core.RedisChannelLayer',
        'CONFIG': {'hosts': [('127.0.0.1', 6379)]},
    },
}

Consumer (el "view" de WebSockets)

# consumers.py
import json
from channels.generic.websocket import AsyncWebsocketConsumer

class NotificationConsumer(AsyncWebsocketConsumer):
    async def connect(self):
        self.user = self.scope['user']
        self.group_name = f'notifications_{self.user.id}'

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

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

    async def send_notification(self, event):
        await self.send(text_data=json.dumps({
            'type': event['notification_type'],
            'message': event['message'],
            'timestamp': event['timestamp'],
        }))

Enviar notificación desde cualquier view

from channels.layers import get_channel_layer
from asgiref.sync import async_to_sync

def notify_user(user_id, message, notification_type='info'):
    channel_layer = get_channel_layer()
    async_to_sync(channel_layer.group_send)(
        f'notifications_{user_id}',
        {
            'type': 'send_notification',
            'notification_type': notification_type,
            'message': message,
            'timestamp': str(timezone.now()),
        }
    )