Django Channels Rest Framework (DCRF)

To notify users about changes to related objects (e.g., notifying room members when a new Message is created), use the @model_observer(ModelClass) decorator. This requires defining three specific methods on the observer to control how data is grouped and serialized:

1. The main observer method: The decorated function that executes for each subscribed consumer. It receives subscribing_request_ids which allows you to loop through and send personalized JSON payloads.
2. groups_for_signal: Defines which Channels groups the signal should be sent to (e.g., based on a foreign key like room_id).
3. groups_for_consumer: Defines which groups a consumer should join when they subscribe to the observer (e.g., passing a room ID).
4. serializer: A method to transform the model instance into a dictionary before it is sent to subscribers. This is efficient as it runs once per update rather than once per subscriber.

@model_observer(Message)
async def message_activity(self, message, observer=None, subscribing_request_ids=[], **kwargs):
    for request_id in subscribing_request_ids:
        message_body = dict(request_id=request_id)
        message_body.update(message)
        await self.send_json(message_body)

@message_activity.groups_for_signal
def message_activity(self, instance: Message, **kwargs):
    yield f'room__{instance.room_id}'

@message_activity.groups_for_consumer
def message_activity(self, room=None, **kwargs):
    if room is not None:
        yield f'room__{room}'

@message_activity.serializer
def message_activity(self, instance: Message, action, **kwargs):
    return dict(
        data=MessageSerializer(instance).data,
        action=action.value,
        pk=instance.pk
    )

You can subscribe WebSocket clients to real-time updates when specific model instances are created, updated, or deleted.

1. Instance Observation: Use the ObserverModelInstanceMixin to subscribe to changes in individual model instances.
2. Collection Observation: Use the @model_observer(ModelClass) decorator to define granular event listeners for collections of models. This tracks all changes made via the Django ORM, including those from the Django admin or CLI.

To implement a custom observer, you define:

• An async method decorated with @model_observer to handle the message.
• A method decorated with @method_name.groups_for_signal to define which groups the signal should be sent to.
• A method decorated with @method_name.groups_for_consumer to define which groups a specific consumer belongs to.
• An @action to trigger the subscription using await self.method_name.subscribe(request_id=..., ...).

from .models import User, Comment
from .serializers import UserSerializer
from djangochannelsrestframework.generics import GenericAsyncAPIConsumer
from djangochannelsrestframework.observer import model_observer

class MyConsumer(GenericAsyncAPIConsumer):
    queryset = User.objects.all()
    serializer_class = UserSerializer

    @model_observer(Comment)
    async def comment_activity(self, message, observer=None, subscribing_request_ids=[], **kwargs):
        for request_id in subscribing_request_ids:
            await self.send_json({"message": message, "request_id": request_id})

    @comment_activity.groups_for_signal
    def comment_activity(self, instance, **kwargs):
        yield f'comment__{instance.user_id}'

    @comment_activity.groups_for_consumer
    def comment_activity(self, user_pk, **kwargs):
        if user_pk:
            yield f'comment__{user_pk}'

    @action()
    async def subscribe_to_comment_activity(self, request_id, user_pk, **kwargs):
        await self.comment_activity.subscribe(request_id=request_id, user_pk=user_pk)

Use the @observer decorator to create a consumer action that reacts to Django Signal emissions. This allows you to push data to specific clients when a signal is triggered.

Implementation Pattern:

1. Define a standard Django Signal.
2. Use @observer(signal=...) on a consumer method to handle the signal.
3. Use helper decorators on the handler method to define:
   • .serializer(): How to transform the signal data.
   • .groups_for_signal(): Which groups the signal should be sent to (based on the signal instance).
   • .groups_for_consumer(): Which groups the consumer belongs to.
4. Use .subscribe() on the handler to allow clients to start listening to that signal.

# signals.py
from django.dispatch.dispatcher import Signal
joined_chat_signal = Signal()

# consumers.py
from djangochannelsrestframework.consumers import AsyncAPIConsumer
from djangochannelsrestframework.decorators import action
from djangochannelsrestframework.observer import observer
from rest_framework import status
from .signals import joined_chat_signal
from .serializers import UserSerializer


class TestConsumer(AsyncAPIConsumer):

    @action()
    def join_chat(self, chat_id, **kwargs):
        serializer = UserSerializer(instance=self.scope['user'])
        joined_chat_signal.send(sender='join_chat', data=serializer.data, **kwargs)
        return {}, status.HTTP_204_NO_CONTENT

    @observer(signal=joined_chat_signal)
    async def joined_chat_handler(self, data, observer=None, action=None, subscribing_request_ids=[], **kwargs):
        for request_id in subscribing_request_ids:
            await self.reply(action='joined_chat', data=data, status=status.HTTP_200_OK, request_id=request_id)

    @joined_chat_handler.serializer
    def join_chat_handler(self, sender, data, **kwargs):
        return data

    @joined_chat_handler.groups_for_signal
    def joined_chat_handler(self, instance, **kwargs):
        yield f'chat__{instance}'

    @joined_chat_handler.groups_for_consumer
    def joined_chat_handler(self, chat, **kwargs):
        if chat:
            yield f'chat__{chat}'

    @action()
    async def subscribe_joined(self, chat_id, request_id, **kwargs):
        await self.joined_chat_handler.subscribe(chat_id, request_id=request_id)

In DCRF, you can create a GenericAsyncAPIConsumer that functions similarly to DRF's GenericAPIView. To enable specific CRUD operations, you must mix in specific action mixins. The consumer requires a queryset and a serializer_class to be defined.

Available mixins include:

• ListModelMixin: adds the list action (retrieve all instances).
• RetrieveModelMixin: adds the retrieve action (retrieve an object by pk).
• PatchModelMixin: adds the patch action (partial update).
• UpdateModelMixin: adds the update action (full update).
• CreateModelMixin: adds the create action (create a new instance).
• DeleteModelMixin: adds the delete action (delete an instance by pk).

from django_channels_rest_framework.generics import GenericAsyncAPIConsumer
from django_channels_rest_framework.mixins import (
    ListModelMixin,
    RetrieveModelMixin,
    PatchModelMixin,
    UpdateModelMixin,
    CreateModelMixin,
    DeleteModelMixin,
)

class UserConsumer(
        ListModelMixin, 
        RetrieveModelMixin,
        PatchModelMixin,
        UpdateModelMixin,
        CreateModelMixin,
        DeleteModelMixin,
        GenericAsyncAPIConsumer,
):
    queryset = User.objects.all()
    serializer_class = UserSerializer

To avoid subscribing to every single instance of a model, you can use a group-based filtering system. This involves three components:

1. @model_observer method: The handler that receives the signal.
2. .groups_for_signal: A decorator that defines which groups an instance belongs to when a change occurs. This is called frequently; do not perform database queries here.
3. .groups_for_consumer: A decorator that defines which groups a consumer should join when they call .subscribe(). This is called during subscription/unsubscription.

To trigger the subscription, use a custom @action that calls .subscribe() on the observer method, passing the necessary filtering parameters (like school or classroom).

class MyConsumer(AsyncAPIConsumer):
    @model_observer(models.Classroom)
    async def classroom_change_handler(
        self, 
        message, 
        observer=None, 
        action=None, 
        subscribing_request_ids=[], 
        **kwargs
    ):
        for request_id in subscribing_request_ids:
            await self.send_json(dict(body=message, action=action, request_id=request_id))

    @classroom_change_handler.groups_for_signal
    def classroom_change_handler(self, instance: models.Classroom, **kwargs):
        yield f'-school__{instance.school_id}'
        yield f'-pk__{instance.pk}'

    @classroom_change_handler.groups_for_consumer
    def classroom_change_handler(self, school=None, classroom=None, **kwargs):
        if school is not None:
            yield f'-school__{school.pk}'
        if classroom is not None:
            yield f'-pk__{classroom.pk}'

    @action()
    async def subscribe_to_classroom(self, classroom_pk, request_id, **kwargs):
        # Perform permission checks here
        await self.classroom_change_handler.subscribe(classroom=classroom, request_id=request_id)