def _evaluate(self, viewset_results=None):
        """Evaluate query observer.

        :param viewset_results: Objects returned by the viewset query
        """
        if viewset_results is None:
            viewset_results = self._viewset_results()

        try:
            observer = models.Observer.objects.get(id=self.id)
            # Do not evaluate the observer if there are no subscribers
            if observer.subscribers.count() == 0:
                return (None, None, None)

            # Update last evaluation time.
            models.Observer.objects.filter(id=self.id).update(
                last_evaluation=timezone.now()
            )

            # Log viewsets with too much output.
            max_result = get_queryobserver_settings()['warnings']['max_result_length']
            if len(viewset_results) > max_result:
                self._warning(
                    "Observed viewset returns too many results",
                    results=len(viewset_results),
                )
            new_results = collections.OrderedDict()
            for order, item in enumerate(viewset_results):
                if not isinstance(item, dict):
                    raise ValueError(
                        "Observable views must return a dictionary or a list of dictionaries!"
                    )

                item = {'order': order, 'data': item}

                try:
                    new_results[str(item['data'][self._meta.primary_key])] = item
                except KeyError:
                    raise KeyError(
                        "Observable view did not return primary key field '{}'!".format(
                            self._meta.primary_key
                        )
                    )

            # Process difference between old results and new results.
            added, changed = [], []
            new_ids = list(new_results.keys())

            removed_qs = observer.items.exclude(primary_key__in=new_results.keys())
            removed = list(removed_qs.values('order', 'data'))

            maybe_changed_qs = observer.items.filter(primary_key__in=new_results.keys())

            with transaction.atomic():
                # Removed items.
                removed_qs.delete()

                # Defer unique ordering constraint before processing order updates.
                # NOTE: The name of the constrait is generated by Django ORM.
                with connection.cursor() as cursor:
                    cursor.execute(
                        "SET CONSTRAINTS rest_framework_reactive_item_observer_id_order_9b8adde6_uniq DEFERRED"
                    )

                # Changed items.
                for item_id, old_order, old_data in maybe_changed_qs.values_list(
                    'primary_key', 'order', 'data'
                ):
                    new_item = new_results[item_id]
                    new_ids.remove(item_id)

                    if new_item['data'] != old_data:
                        changed.append(new_item)
                        observer.items.filter(primary_key=item_id).update(
                            data=new_item['data'], order=new_item['order']
                        )
                    elif new_item['order'] != old_order:
                        # TODO: If only order has changed, don't transmit
                        # full data (needs frontend support).
                        changed.append(new_item)
                        observer.items.filter(primary_key=item_id).update(
                            order=new_item['order']
                        )

                # Added items.
                for item_id in new_ids:
                    item = new_results[item_id]
                    added.append(item)
                    observer.items.create(
                        primary_key=item_id, order=item['order'], data=item['data']
                    )

            return (added, changed, removed)

        except models.Observer.DoesNotExist:
            # Observer removed, ignore evaluation
            return (None, None, None)
def _evaluate ( self , viewset_results = None ) : if viewset_results is None : viewset_results = self . _viewset_results ( ) try : observer = models . Observer . objects . get ( id = self . id ) if observer . subscribers . count ( ) == 0 : return ( None , None , None ) models . Observer . objects . filter ( id = self . id ) . update ( last_evaluation = timezone . now ( ) ) max_result = get_queryobserver_settings ( ) [ 'warnings' ] [ 'max_result_length' ] if len ( viewset_results ) > max_result : self . _warning ( "Observed viewset returns too many results" , results = len ( viewset_results ) , ) new_results = collections . OrderedDict ( ) for order , item in enumerate ( viewset_results ) : if not isinstance ( item , dict ) : raise ValueError ( "Observable views must return a dictionary or a list of dictionaries!" ) item = { 'order' : order , 'data' : item } try : new_results [ str ( item [ 'data' ] [ self . _meta . primary_key ] ) ] = item except KeyError : raise KeyError ( "Observable view did not return primary key field '{}'!" . format ( self . _meta . primary_key ) ) added , changed = [ ] , [ ] new_ids = list ( new_results . keys ( ) ) removed_qs = observer . items . exclude ( primary_key__in = new_results . keys ( ) ) removed = list ( removed_qs . values ( 'order' , 'data' ) ) maybe_changed_qs = observer . items . filter ( primary_key__in = new_results . keys ( ) ) with transaction . atomic ( ) : removed_qs . delete ( ) with connection . cursor ( ) as cursor : cursor . execute ( "SET CONSTRAINTS rest_framework_reactive_item_observer_id_order_9b8adde6_uniq DEFERRED" ) for item_id , old_order , old_data in maybe_changed_qs . values_list ( 'primary_key' , 'order' , 'data' ) : new_item = new_results [ item_id ] new_ids . remove ( item_id ) if new_item [ 'data' ] != old_data : changed . append ( new_item ) observer . items . filter ( primary_key = item_id ) . update ( data = new_item [ 'data' ] , order = new_item [ 'order' ] ) elif new_item [ 'order' ] != old_order : changed . append ( new_item ) observer . items . filter ( primary_key = item_id ) . update ( order = new_item [ 'order' ] ) for item_id in new_ids : item = new_results [ item_id ] added . append ( item ) observer . items . create ( primary_key = item_id , order = item [ 'order' ] , data = item [ 'data' ] ) return ( added , changed , removed ) except models . Observer . DoesNotExist : return ( None , None , None )
Evaluate query observer.

        :param viewset_results: Objects returned by the viewset query