def _update_from_diff(self, new, changed, deleted):
        msg_format = 'now updating from diff new: %s changed: %s deleted: %s'
        logger.debug(msg_format, *map(len, [new, changed, deleted]))
        to_remove, to_add = self._translate_diff(new, changed, deleted)

        # remove those which changed
        with self.get_timeline_batch_interface() as batch_interface:
            if to_remove:
                self.remove_many_aggregated(
                    to_remove, batch_interface=batch_interface)

        # now add the new ones
        with self.get_timeline_batch_interface() as batch_interface:
            if to_add:
                self.add_many_aggregated(
                    to_add, batch_interface=batch_interface)

        logger.debug(
            'removed %s, added %s items from feed %s', len(to_remove), len(to_add), self)

        # return the merge of these two
        new_aggregated = new[:]
        if changed:
            new_aggregated += list(zip(*changed))[1]

        self.on_update_feed(to_add, to_remove)
        return new_aggregated
Sends the add and remove commands to the storage layer based on a diff
dataset/ETHPy150Open tschellenbach/Stream-Framework/stream_framework/feeds/aggregated_feed/base.py/AggregatedFeed._update_from_diff