Source code for divera.triggers

import abc
import typing

import divera.models


class Trigger(abc.ABC):
    @abc.abstractmethod
    def __init__(self):
        pass


class ObjectInteraction(Trigger):
    def __init__(
            self,
            obj: divera.models.Model,
    ):
        self.object: divera.models.Model = obj


class ObjectCreated(ObjectInteraction):
    pass


class AttributeChanged(ObjectInteraction):
    def __init__(
            self,
            obj,
            attribute_name: str,
            old_value,
    ):
        super().__init__(obj=obj)
        self.attribute_name = attribute_name
        self.old_value = old_value


class AttributesChanged(ObjectInteraction):
    def __init__(
            self,
            obj,
            old_values: dict,
    ):
        super().__init__(obj=obj)
        self.old_values = old_values


class ObjectRemoved(ObjectInteraction):
    pass


async def get_object_changes(
        new_objects: typing.List[divera.models.Model],
        old_objects: typing.List[divera.models.Model],
):
    """
    Find the differences between two lists of objects.

    :param new_objects: The list of objects that comes fresh from the api (in most cases).
    :param old_objects: The list of objects that has been stored in memory since the previous API call.
    :return: Trigger (or subclass) objects that describe the type of change that has been detected.
    """
    for obj in new_objects:
        if obj not in old_objects:
            yield divera.triggers.ObjectCreated(obj=obj)
        else:
            i = old_objects.index(obj)
            old_values = {}
            for attribute in obj - old_objects[i]:
                old_values[attribute] = old_objects[i].data[attribute]
                yield divera.triggers.AttributeChanged(
                    obj=obj,
                    attribute_name=attribute,
                    old_value=old_objects[i].data[attribute] if attribute in old_objects[i].data else None,
                )
            yield divera.AttributesChanged(
                obj=obj,
                old_values=old_values,
            )
    for obj in old_objects:
        if obj not in new_objects:
            yield divera.triggers.ObjectRemoved(obj)