Create a subscription via model fields

[vCra]

Hi there,
Currently, the subscriptions only allow subscriptions for any resource, or a specific resource by it's PK. Is it possible to implement functionally to subscribe to a set of resources that fulfil some criteria, other than PK.
For example, we might have a field called name, and we may want to subscribe to any creations in which the name == "John". The current work around is to subscribe for all creations, and then filter the stream client side.
Just an example of how this could be done
{
"stream": "people",
"payload": {
"action":"subscribe",
"data": {
"action": "create"
}
"filter":{
"name": "John"
}
}

It would probably also make sense to move the existing PK field to the new filter dict.

This is obviously a big design change, but I personally feel that keeping as much of the processing on the server side is better than the current workaround.

[maxblax]

Hi All,
I'm desperatly in need for this feature too. I need it for scaling issues over my multiface app architecture. I want the client side to register on notification only concerning its company.
Could you please give visibility on this feature on the repo side?
If not on the pipe, I will need to develop something custom probably..
Cheers
Maxime

[maxblax]

Ok, ain't perfect, but here is a specific implementation of that feature, for all object you want to subscribe from, register throw a sub_field model called "company". But it's working for my use case.
Remaining problem is that there are no proper session security implemented on channels_api, which I need from rest_token authentication. For the moment it's still hackabled... So only to use for non-critical notifications objects. Checking #31
Anyway, here is my implementation:

notification/bindings.py

import json
import logging
import traceback
from pprint import pformat

from channels import Group
from channels_api import detail_action
from channels_api.bindings import ResourceBindingBase
from rest_framework.exceptions import APIException, ValidationError
from rest_framework.status import  HTTP_404_NOT_FOUND
from termcolor import colored

from accounts.models import Company
from accounts.serializers import User
from .models import Notification
from .serializers import ASGINotificationSerializer

logger = logging.getLogger(__name__)

class CompanySubscribeResourceBindingBase(ResourceBindingBase):
    # mark as abstract
    lookup_field = 'company__name'
    permission_classes = None # /!\ Only company based permissions

    # mark as abstract
    model = None

    @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
        self = cls()
        self.message = message
        self.kwargs = kwargs
        # Deserialize message
        try:
            self.action, self.company_name, self.username, self.data = self.deserialize(self.message)
        except KeyError:
            return super(CompanySubscribeResourceBindingBase, self).trigger_inbound(message, **kwargs)
        logger.debug(colored(pformat(self.message.items()), "cyan"))
        logger.debug(colored("{} - {} - {} - {}".format(self.action, self.company_name, self.username, self.data), "cyan"))
        try:
            self.user = User.objects.get(username=self.username)
        except :
            self.reply(self.action, errors=["User not found from,"
                                            " please precise under 'payload' {'username':<username>}"
                                            ""], status=HTTP_404_NOT_FOUND)
            return
        # Run incoming action
        self.run_action(self.action, self.company_name, self.data)

    @classmethod
    def group_names(cls, instance, action):
        self = cls()
        groups = [self._group_name(action)]
        if instance.pk:
            groups.append(self._group_name(action, company=instance.company))
        return groups

    def _group_name(self, action, company=None):
        """Formatting helper for group names."""
        if company:
            return "{}-{}-{}".format(self.model_label, action, company.id)
        else:
            return "{}-{}".format(self.model_label, action)


    def deserialize(self, message):
        body = json.loads(message['text'])
        self.request_id = body.get("request_id")
        action = body['action']
        company_name = body.get('company_name', None)
        username = body.get('username', None)
        data = body.get('data', None)
        return action, company_name, username, data

    def has_permission(self, user, action, company):
        # TODO: Not generic enough, but permission signature in channels differ from DRF... so tmp hardcoding
        members = company.get_members()
        return not members or user.profile in members

    def run_action(self, action, company_name, data):
        try:
            company=Company.objects.get(name=company_name)
        except :
            logger.error(colored(traceback.format_exc(),'red'))
            logger.error(colored("Could not fetch company with name {}".format(company_name),'red'))
            self.reply(action, errors=['Company not found'], status=HTTP_404_NOT_FOUND)

        try:
            if not self.has_permission(self.user, action, company):
                self.reply(action, errors=['Permission Denied'], status=401)
            elif action not in self.available_actions:
                self.reply(action, errors=['Invalid Action'], status=400)
            else:
                methodname = self.available_actions[action]
                method = getattr(self, methodname)
                detail = getattr(method, 'detail', True)
                if detail:
                    rv = method(company, data=data)
                else:
                    rv = method(data=data)
                data, status = rv
                self.reply(action, data=data, status=status, request_id=self.request_id)
        except APIException as ex:
            self.reply(action, errors=self._format_errors(ex.detail), status=ex.status_code, request_id=self.request_id)



class NotificationBinding(CompanySubscribeResourceBindingBase):
# class NotificationBinding(ResourceBinding):
    model = Notification
    stream = "notifications"
    serializer_class = ASGINotificationSerializer


    @detail_action()
    def filter_subscribe(self, company, data, **kwargs):

        if 'action' not in data:
            raise ValidationError('action required')
        action = data['action']
        group_name = self._group_name(action, company=company)
        Group(group_name).add(self.message.reply_channel)
        return {'action': action}, 200

routing.py

from channels.generic.websockets import WebsocketDemultiplexer
from channels.routing import route_class

from notification.bindings import NotificationBinding

class APIDemultiplexer(WebsocketDemultiplexer):
    # TODO: Track repo evolution for rest_token support to have proper security here.
    # https://github.com/linuxlewis/channels-api/issues/31
    consumers = {
      'notifications': NotificationBinding.consumer,
    }

application = [
    route_class(APIDemultiplexer)
]

[washingtoncostaa]

I'm looking for this feature too. After searching a lot, no success, I'm trying override the subscriber method because I'm creating a chat, so the user just can receiver messages that was send to him.

Do you have any ideias how to apply this filter?