Merge pull request #252 from andrewgodwin/binding

Data Binding

channels/apps.py

@@ -1,6 +1,8 @@
 from django.apps import AppConfig
 from django.core.exceptions import ImproperlyConfigured
 
+from .binding.base import BindingMetaclass
+
 
 class ChannelsConfig(AppConfig):
 
@@ -18,3 +20,5 @@ class ChannelsConfig(AppConfig):
         # Do django monkeypatches
         from .hacks import monkeypatch_django
         monkeypatch_django()
+        # Instantiate bindings
+        BindingMetaclass.register_all()

channels/binding/__init__.py

@@ -0,0 +1 @@
+from .base import Binding  # NOQA isort:skip

channels/binding/base.py

@@ -0,0 +1,187 @@
+from __future__ import unicode_literals
+
+import six
+
+from django.apps import apps
+from django.db.models.signals import post_save, post_delete
+
+from ..channel import Group
+
+
+class BindingMetaclass(type):
+    """
+    Metaclass that tracks instantiations of its type.
+    """
+
+    binding_classes = []
+
+    def __new__(cls, name, bases, body):
+        klass = type.__new__(cls, name, bases, body)
+        if bases != (object, ):
+            cls.binding_classes.append(klass)
+        return klass
+
+    @classmethod
+    def register_all(cls):
+        for binding_class in cls.binding_classes:
+            binding_class.register()
+
+
+@six.add_metaclass(BindingMetaclass)
+class Binding(object):
+    """
+    Represents a two-way data binding from channels/groups to a Django model.
+    Outgoing binding sends model events to zero or more groups.
+    Incoming binding takes messages and maybe applies the action based on perms.
+
+    To implement outbound, implement:
+     - group_names, which returns a list of group names to send to
+     - serialize, which returns message contents from an instance + action
+
+    To implement inbound, implement:
+     - deserialize, which returns pk, data and action from message contents
+     - has_permission, which says if the user can do the action on an instance
+     - create, which takes the data and makes a model instance
+     - update, which takes data and a model instance and applies one to the other
+
+    Outbound will work once you implement the functions; inbound requires you
+    to place one or more bindings inside a protocol-specific Demultiplexer
+    and tie that in as a consumer.
+    """
+
+    model = None
+
+    @classmethod
+    def register(cls):
+        """
+        Resolves models.
+        """
+        # If model is None directly on the class, assume it's abstract.
+        if cls.model is None:
+            if "model" in cls.__dict__:
+                return
+            else:
+                raise ValueError("You must set the model attribute on Binding %r!" % cls)
+        # Optionally resolve model strings
+        if isinstance(cls.model, six.string_types):
+            cls.model = apps.get_model(cls.model)
+        cls.model_label = "%s.%s" % (
+            cls.model._meta.app_label.lower(),
+            cls.model._meta.object_name.lower(),
+        )
+        # Connect signals
+        post_save.connect(cls.save_receiver, sender=cls.model)
+        post_delete.connect(cls.delete_receiver, sender=cls.model)
+
+    # Outbound binding
+
+    @classmethod
+    def save_receiver(cls, instance, created, **kwargs):
+        """
+        Entry point for triggering the binding from save signals.
+        """
+        cls.trigger_outbound(instance, "create" if created else "update")
+
+    @classmethod
+    def delete_receiver(cls, instance, **kwargs):
+        """
+        Entry point for triggering the binding from save signals.
+        """
+        cls.trigger_outbound(instance, "delete")
+
+    @classmethod
+    def trigger_outbound(cls, instance, action):
+        """
+        Triggers the binding to possibly send to its group.
+        """
+        self = cls()
+        self.instance = instance
+        # Check to see if we're covered
+        for group_name in self.group_names(instance, action):
+            group = Group(group_name)
+            group.send(self.serialize(instance, action))
+
+    def group_names(self, instance, action):
+        """
+        Returns the iterable of group names to send the object to based on the
+        instance and action performed on it.
+        """
+        raise NotImplementedError()
+
+    def serialize(self, instance, action):
+        """
+        Should return a serialized version of the instance to send over the
+        wire (return value must be a dict suitable for sending over a channel -
+        e.g., to send JSON as a WebSocket text frame, you must return
+        {"text": json.dumps(instance_serialized_as_dict)}
+        """
+        raise NotImplementedError()
+
+    # Inbound binding
+
+    @classmethod
+    def trigger_inbound(cls, message, **kwargs):
+        """
+        Triggers the binding to see if it will do something.
+        Also acts as a consumer.
+        """
+        # Late import as it touches models
+        from django.contrib.auth.models import AnonymousUser
+        self = cls()
+        self.message = message
+        # Deserialize message
+        self.action, self.pk, self.data = self.deserialize(self.message)
+        self.user = getattr(self.message, "user", AnonymousUser())
+        # Run incoming action
+        self.run_action(self.action, self.pk, self.data)
+
+    consumer = trigger_inbound
+
+    def deserialize(self, message):
+        """
+        Returns action, pk, data decoded from the message. pk should be None
+        if action is create; data should be None if action is delete.
+        """
+        raise NotImplementedError()
+
+    def has_permission(self, user, action, pk):
+        """
+        Return True if the user can do action to the pk, False if not.
+        User may be AnonymousUser if no auth hooked up/they're not logged in.
+        Action is one of "create", "delete", "update".
+        """
+        raise NotImplementedError()
+
+    def run_action(self, action, pk, data):
+        """
+        Performs the requested action. This version dispatches to named
+        functions by default for update/create, and handles delete itself.
+        """
+        # Check to see if we're allowed
+        if self.has_permission(self.user, action, pk):
+            if action == "create":
+                self.create(data)
+            elif action == "update":
+                self.update(pk, data)
+            elif action == "delete":
+                self.delete(pk)
+            else:
+                raise ValueError("Bad action %r" % action)
+
+    def create(self, data):
+        """
+        Creates a new instance of the model with the data.
+        """
+        raise NotImplementedError()
+
+    def update(self, pk, data):
+        """
+        Updates the model with the data.
+        """
+        raise NotImplementedError()
+
+    def delete(self, pk):
+        """
+        Deletes the model instance.
+        """
+        self.model.objects.filter(pk=pk).delete()

channels/binding/websockets.py

@@ -0,0 +1,88 @@
+import json
+
+from django.core import serializers
+
+from .base import Binding
+from ..generic.websockets import WebsocketDemultiplexer
+
+
+class WebsocketBinding(Binding):
+    """
+    Websocket-specific outgoing binding subclass that uses JSON encoding
+    and the built-in JSON/WebSocket multiplexer.
+
+    To implement outbound, implement:
+     - group_names, which returns a list of group names to send to
+
+    To implement inbound, implement:
+     - has_permission, which says if the user can do the action on an instance
+
+    Optionally also implement:
+     - serialize_data, which returns JSON-safe data from a model instance
+     - create, which takes incoming data and makes a model instance
+     - update, which takes incoming data and a model instance and applies one to the other
+    """
+
+    # Mark as abstract
+
+    model = None
+
+    # Stream multiplexing name
+
+    stream = None
+
+    # Outbound
+
+    def serialize(self, instance, action):
+        payload = {
+            "action": action,
+            "pk": instance.pk,
+            "data": self.serialize_data(instance),
+        }
+        # Encode for the stream
+        assert self.stream is not None
+        return WebsocketDemultiplexer.encode(self.stream, payload)
+
+    def serialize_data(self, instance):
+        """
+        Serializes model data into JSON-compatible types.
+        """
+        data = serializers.serialize('json', [instance])
+        return json.loads(data)[0]['fields']
+
+    # Inbound
+
+    def deserialize(self, message):
+        """
+        You must hook this up behind a Deserializer, so we expect the JSON
+        already dealt with.
+        """
+        action = message['action']
+        pk = message.get('pk', None)
+        data = message.get('data', None)
+        return action, pk, data
+
+    def _hydrate(self, pk, data):
+        """
+        Given a raw "data" section of an incoming message, returns a
+        DeserializedObject.
+        """
+        s_data = [
+            {
+                "pk": pk,
+                "model": self.model_label,
+                "fields": data,
+            }
+        ]
+        # TODO: Avoid the JSON roundtrip by using encoder directly?
+        return list(serializers.deserialize("json", json.dumps(s_data)))[0]
+
+    def create(self, data):
+        self._hydrate(None, data).save()
+
+    def update(self, pk, data):
+        instance = self.model.objects.get(pk=pk)
+        hydrated = self._hydrate(pk, data)
+        for name in data.keys():
+            setattr(instance, name, getattr(hydrated.object, name))
+        instance.save()

channels/generic/websockets.py

@@ -1,6 +1,6 @@
 import json
 
-from ..channel import Group
+from ..channel import Group, Channel
 from ..auth import channel_session_user_from_http
 from ..sessions import enforce_ordering
 from .base import BaseConsumer
@@ -98,11 +98,12 @@ class WebsocketConsumer(BaseConsumer):
         else:
             raise ValueError("You must pass text or bytes")
 
-    def group_send(self, name, text=None, bytes=None):
+    @classmethod
+    def group_send(cls, name, text=None, bytes=None):
         if text is not None:
-            Group(name, channel_layer=self.message.channel_layer).send({"text": text})
+            Group(name).send({"text": text})
         elif bytes is not None:
-            Group(name, channel_layer=self.message.channel_layer).send({"bytes": bytes})
+            Group(name).send({"bytes": bytes})
         else:
             raise ValueError("You must pass text or bytes")
 
@@ -153,5 +154,61 @@ class JsonWebsocketConsumer(WebsocketConsumer):
         """
         super(JsonWebsocketConsumer, self).send(text=json.dumps(content))
 
-    def group_send(self, name, content):
-        super(JsonWebsocketConsumer, self).group_send(name, json.dumps(content))
+    @classmethod
+    def group_send(cls, name, content):
+        WebsocketConsumer.group_send(name, json.dumps(content))
+
+
+class WebsocketDemultiplexer(JsonWebsocketConsumer):
+    """
+    JSON-understanding WebSocket consumer subclass that handles demultiplexing
+    streams using a "stream" key in a top-level dict and the actual payload
+    in a sub-dict called "payload". This lets you run multiple streams over
+    a single WebSocket connection in a standardised way.
+
+    Incoming messages on streams are mapped into a custom channel so you can
+    just tie in consumers the normal way. The reply_channels are kept so
+    sessions/auth continue to work. Payloads must be a dict at the top level,
+    so they fulfill the Channels message spec.
+
+    Set a mapping from streams to channels in the "mapping" key. We make you
+    whitelist channels like this to allow different namespaces and for security
+    reasons (imagine if someone could inject straight into websocket.receive).
+    """
+
+    mapping = {}
+
+    def receive(self, content, **kwargs):
+        # Check the frame looks good
+        if isinstance(content, dict) and "stream" in content and "payload" in content:
+            # Match it to a channel
+            stream = content['stream']
+            if stream in self.mapping:
+                # Extract payload and add in reply_channel
+                payload = content['payload']
+                if not isinstance(payload, dict):
+                    raise ValueError("Multiplexed frame payload is not a dict")
+                payload['reply_channel'] = self.message['reply_channel']
+                # Send it onto the new channel
+                Channel(self.mapping[stream]).send(payload)
+            else:
+                raise ValueError("Invalid multiplexed frame received (stream not mapped)")
+        else:
+            raise ValueError("Invalid multiplexed frame received (no channel/payload key)")
+
+    def send(self, stream, payload):
+        self.message.reply_channel.send(self.encode(stream, payload))
+
+    @classmethod
+    def group_send(cls, name, stream, payload):
+        Group(name).send(cls.encode(stream, payload))
+
+    @classmethod
+    def encode(cls, stream, payload):
+        """
+        Encodes stream + payload for outbound sending.
+        """
+        return {"text": json.dumps({
+            "stream": stream,
+            "payload": payload,
+        })}

docs/asgi.rst

@@ -348,6 +348,11 @@ A channel layer implementing the ``groups`` extension must also provide:
 * ``group_discard(group, channel)``, a callable that removes the ``channel``
   from the ``group`` if it is in it, and does nothing otherwise.
 
+* ``group_channels(group)``, a callable that returns an iterable which yields
+  all of the group's member channel names. The return value should be serializable
+  with regards to local adds and discards, but best-effort with regards to
+  adds and discards on other nodes.
+
 * ``send_group(group, message)``, a callable that takes two positional
   arguments; the group to send to, as a unicode string, and the message
   to send, as a serializable ``dict``. It may raise MessageTooLarge but cannot

docs/binding.rst

@@ -0,0 +1,149 @@
+Data Binding
+============
+
+.. warning::
+
+    Data Binding is new and might change slightly in the
+    upcoming weeks, and so don't consider this API totally stable yet.
+
+The Channels data binding framework automates the process of tying Django
+models into frontend views, such as javascript-powered website UIs. It provides
+a quick and flexible way to generate messages on Groups for model changes
+and to accept messages that chanes models themselves.
+
+The main target for the moment is WebSockets, but the framework is flexible
+enough to be used over any protocol.
+
+What does data binding allow?
+-----------------------------
+
+Data binding in Channels works two ways:
+
+* Outbound, where model changes made through Django are sent out to listening
+  clients. This includes creation, update and deletion of instances.
+
+* Inbound, where a standardised message format allow creation, update and
+  deletion of instances to be made by clients sending messages.
+
+Combined, these allow a UI to be designed that automatically updates to
+reflect new values and reflects across clients. A live blog is easily done
+using data binding against the post object, for example, or an edit interface
+can show data live as it's edited by other users.
+
+It has some limitations:
+
+* Signals are used to power outbound binding, so if you change the values of
+  a model outside of Django (or use the ``.update()`` method on a QuerySet),
+  the signals are not triggered and the change will not be sent out. You
+  can trigger changes yourself, but you'll need to source the events from the
+  right place for your system.
+
+* The built-in serializers are based on the built-in Django ones and can only
+  handle certain field types; for more flexibility, you can plug in something
+  like the Django REST Framework serializers.
+
+Getting Started
+---------------
+
+A single Binding subclass will handle outbound and inbound binding for a model,
+and you can have multiple bindings per model (if you want different formats
+or permission checks, for example).
+
+You can inherit from the base Binding and provide all the methods needed, but
+we'll focus on the WebSocket JSON variant here, as it's the easiest thing to
+get started and likely close to what you want. Start off like this::
+
+    from django.db import models
+    from channels.binding.websockets import WebsocketBinding
+
+    class IntegerValue(models.Model):
+
+        name = models.CharField(max_length=100, unique=True)
+        value = models.IntegerField(default=0)
+
+    class IntegerValueBinding(WebsocketBinding):
+
+        model = IntegerValue
+        stream = "intval"
+
+        def group_names(self, instance, action):
+            return ["intval-updates"]
+
+        def has_permission(self, user, action, pk):
+            return True
+
+This defines a WebSocket binding - so it knows to send outgoing messages
+formatted as JSON WebSocket frames - and provides the two methods you must
+always provide:
+
+* ``group_names`` returns a list of groups to send outbound updates to based
+  on the model and action. For example, you could dispatch posts on different
+  liveblogs to groups that included the parent blog ID in the name; here, we
+  just use a fixed group name.
+
+* ``has_permission`` returns if an inbound binding update is allowed to actually
+  be carried out on the model. We've been very unsafe and made it always return
+  ``True``, but here is where you would check against either Django's or your
+  own permission system to see if the user is allowed that action.
+
+For reference, ``action`` is always one of the unicode strings ``"create"``,
+``"update"`` or ``"delete"``. You also supply the :ref:`multiplexing`
+stream name to provide to the client - you must use multiplexing if you
+use WebSocket data binding.
+
+Just adding the binding like this in a place where it will be imported will
+get outbound messages sending, but you still need a Consumer that will both
+accept incoming binding updates and add people to the right Groups when they
+connect. The WebSocket binding classes use the standard :ref:`multiplexing`,
+so you just need to use that::
+
+    from channels.generic.websockets import WebsocketDemultiplexer
+
+    class Demultiplexer(WebsocketDemultiplexer):
+
+        mapping = {
+            "intval": "binding.intval",
+        }
+
+        def connection_groups(self):
+            return ["intval-updates"]
+
+As well as the standard stream-to-channel mapping, you also need to set
+``connection_groups``, a list of groups to put people in when they connect.
+This should match the logic of ``group_names`` on your binding - we've used
+our fixed group name again.
+
+Tie that into your routing, and tie each demultiplexed channel into the
+``.consumer`` attribute of the Binding, and you're ready to go::
+
+    from channels import route_class
+    from .consumers import BindingConsumer
+    from .models import IntegerValueBinding
+
+    channel_routing = [
+        route_class(BindingConsumer, path="^binding/"),
+        route("binding.intval", IntegerValueBinding.consumer),
+    ]
+
+
+Frontend Considerations
+-----------------------
+
+You can use the standard Channels WebSocket wrapper **(not yet available)**
+to automatically run demultiplexing, and then tie the events you receive into
+your frontend framework of choice based on ``action``, ``pk`` and ``data``.
+
+.. note::
+
+    Common plugins for data binding against popular JavaScript frameworks are
+    wanted; if you're interested, please get in touch.
+
+
+Custom Serialization/Protocols
+------------------------------
+
+Rather than inheriting from the ``WebsocketBinding``, you can inherit directly
+from the base ``Binding`` class and implement serialization and deserialization
+yourself. Until proper reference documentation for this is written, we
+recommend looking at the source code in ``channels/bindings/base.py``; it's
+reasonably well-commented.

docs/generics.rst

@@ -175,6 +175,40 @@ that will do this for you if you call it explicitly.
 ``self.close()`` is also provided to easily close the WebSocket from the server
 end once you are done with it.
 
+.. _multiplexing:
+
+WebSocket Multiplexing
+----------------------
+
+Channels provides a standard way to multiplex different data streams over
+a single WebSocket, called a ``Demultiplexer``. You use it like this::
+
+    from channels.generic.websockets import WebsocketDemultiplexer
+
+    class Demultiplexer(WebsocketDemultiplexer):
+
+        mapping = {
+            "intval": "binding.intval",
+            "stats": "internal.stats",
+        }
+
+It expects JSON-formatted WebSocket frames with two keys, ``stream`` and
+``payload``, and will match the ``stream`` against the mapping to find a
+channel name. It will then forward the message onto that channel while
+preserving ``reply_channel``, so you can hook consumers up to them directly
+in the ``routing.py`` file, and use authentication decorators as you wish.
+
+You cannot use class-based consumers this way as the messages are no
+longer in WebSocket format, though. If you need to do operations on
+``connect`` or ``disconnect``, override those methods on the ``Demultiplexer``
+itself (you can also provide a ``connection_groups`` method, as it's just
+based on the JSON WebSocket generic consumer).
+
+The :doc:`data binding <binding>` code will also send out messages to clients
+in the same format, and you can encode things in this format yourself by
+using the ``WebsocketDemultiplexer.encode`` class method.
+
+
 Sessions and Users
 ------------------
 

docs/getting-started.rst

@@ -60,7 +60,7 @@ Here's what that looks like::
             "ROUTING": "myproject.routing.channel_routing",
         },
     }
-..
+
 ::
 
     # In routing.py

docs/index.rst

@@ -31,6 +31,7 @@ Contents:
    deploying
    generics
    routing
+   binding
    backends
    testing
    cross-compat