:: _using-rxdjango:

Using RxDjango

Backend

The rxdjango.channels.ContextChannel is the core of the RxDjango. Every ContextChannel subclass must declare a Meta class containing a state property

from rxdjango.channels import ContextChannel
from myapp.serializers import MyNestedSerializer

class MyContextChannel(ContextChannel):

    class Meta:
        state = MyNestedSerializer()

A ContextChannel has to be either a single or a multiple instance channel. A single instance channel, like the one above, must define the has_permission static method, to verify if user has permission on an instance:

from rxdjango.channels import ContextChannel
from myapp.serializers import MyNestedSerializer

class MyContextChannel(ContextChannel):

    class Meta:
        state = MyNestedSerializer()

    @staticmethod
    def has_permission(user, **kwargs)

    def get_instance_id(self, **kwargs):
        # this is optional, by default returns the first kwarg
        return kwargs['my_model_id']

To declare a channel that has several instances, many=True has to be passed to the serializer. In this case, the method list_instances() must be implemented, and it should return either a queryset or a list of instance ids.

from rxdjango.channels import ContextChannel
from myapp.serializers import MyNestedSerializer
from mayapp.models import MyModel()

class MyContextListChannel(ContextChannel):

    class Meta:
        state = MyNestedSerializer(many=True)


    async def list_instances(self, **kwargs):
        # Return a queryset of visible objects
        # You may need to use @database_sync_to_async

In both cases, **kwargs comes from the paths registered in asgi.py:

from myapp.channels import MyContextChannel

websocket_urlpatterns = [
    path('ws/myapp/instance/<str:mymodel_id>/', MyContextChannel.as_asgi()),
    path('ws/myapp/list', MyContextListChannel.as_asgi()),
]

application = ProtocolTypeRouter({
    "http": app,
    "websocket": URLRouter(
        websocket_urlpatterns
    ),
})

Frontend

Once the channels and routes are created, you can run the makefrontend command to generate the typescript interfaces and classes at the frontend. Make sure you have configured settings.RX_FRONTEND_DIR and settings.RX_WEBSOCKET_URL.

./manage.py makefrontend

The output of the command will be a diff of the frontend files. If you want to automatically build frontend as you develop, you can use the –makefrontend options for runserver:

./manage.py runserver --makefrontend

Make sure you have installed @rxdjango/react dependency in the frontend.

import { useChannelState } from "@rxdjango/react";
import { MyNestedType } from "my-rx-frontend-dir/myapp.interfaces";
import { MyContextChannel } from "my-rx-frontend-dir/myapp.channels";

const MyPage = () => {
  const channel = new MyContextChannel(instanceId, auth.token);
  const state = useChannelState<MyNestedType>(channel);
}

Now the state variable will be automatically updated with the state of the instance as it is updated in the models.

Actions

Actions operate on both backend and frontend. With actions, methods can be registered on the backend to be called directly from the frontend.

from rxdjango.actions import action

class MyContextListChannel(ContextChannel):

    ...

    @action
    async def change_instance_state(self, some_var: int) -> bool:
        # do something, changes in state will automatically be broadcast
        return result

When creating actions, it’s important to use typehints, so the typings can automatically be generated for the frontend.

In channels with a list of instances, actions can be used to change the instances in the context, for example to create a search:

from rxdjango.actions import action

class MyContextListChannel(ContextChannel):

    search_term = None

    @action
    async def search(self, term: str) -> None:
        self.search_term = term
        instances = self._list_instances()
        self.clear()
        for instance in instances:
            self.add_instance(instance)

add_instance, remove_instance and clear methods can be used to change the instances in the context, for list channels.

On the frontend side, a method will be created in the channel class. As you call the method in the frontend, it will be asynchronously called in the backend, and the results will be returned in the frontend.

const channel = new MyContextChannel(instanceId, auth.token);

await channel.search(searchTerm);

Consumers

RxDjango is build on top of Django Channels, which implements the concept of consumers. Each ContextChannel instance has a private instance of AsyncWebsocketConsumer, and provides an api to it.

You can implement consumer functionality by using the rxdjango.consumers.consumer decorator on your ContextChannel:

from rxdjango.channels import ContextChannel
from rxdjango.consumers import consumer

class MyChannel(ContextChannel):

    @consumer('some.event.type')
    def my_consumer(self, event):
        # handle event

    async def on_connect(tstamp):
        # Join a group to receive events
        await self.group_add('some-group')

For this to work, you will probably want to join some group, as shown above. The group_add works like in a Django Channels consumer.

See Channels Layers documentation for information on how to send messages to groups and general consumer functionality.