Skip to content

Diff

ChangeTypes

Bases: StrEnum

Enum for the types of changes that can be made to a State.

Source code in src/edgygraph/diff.py 
10
11
12
13
14
15
16
17
class ChangeTypes(StrEnum):
    """
    Enum for the types of changes that can be made to a State.
    """

    ADDED = auto()
    REMOVED = auto()
    UPDATED = auto()

Change

Bases: RichReprMixin, BaseModel

Represents a change made to a State.

Source code in src/edgygraph/diff.py 
19
20
21
22
23
24
25
26
class Change(RichReprMixin, BaseModel):
    """
    Represents a change made to a State.
    """

    type: ChangeTypes
    old: Any
    new: Any

Diff

Utility class for computing differences between states.

Source code in src/edgygraph/diff.py 
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
class Diff:
    """
    Utility class for computing differences between states.
    """


    @classmethod
    def find_conflicts(cls, changes: list[dict[tuple[Hashable, ...], Change]]) -> dict[tuple[Hashable, ...], list[Change]]:
        """
        Finds conflicts in a list of changes.

        Args:
           changes: A list of dictionaries representing changes to a state.

        Returns:
            A dictionary mapping a path in the state as a list of keys to lists of conflicting changes directly under that path.
        """

        if len(changes) <= 1:
            return {}

        counts = Counter(key for d in changes for key in d)

        duplicate_keys = [k for k, count in counts.items() if count > 1]

        conflicts: dict[tuple[Hashable, ...], list[Change]] = {}        
        for key in duplicate_keys:
            conflicts[key] = [d[key] for d in changes if key in d]

        return conflicts


    @classmethod
    def recursive_diff(cls, old: Any, new: Any, path: tuple[Hashable, ...] | None = None) -> dict[tuple[Hashable, ...], Change]:
        """
        Recursively computes the differences between two dictionaries.


        Args:
            old: Part of the old dictionary.
            new: Part of the new dictionary.
            path: The current path of the parts in the full dictionary as a list of keys from least to most specific.

        Returns:
            A mapping of the path to the changes directly on that level.
        """

        path = path or ()
        changes: dict[tuple[Hashable, ...], Change] = {}

        if isinstance(old, dict) and isinstance(new, dict):
            all_keys: set[str] = set(old.keys()) | set(new.keys()) #type: ignore

            for key in all_keys:
                current_path: tuple[Hashable, ...] = (*path, key)

                if key in old and not key in new:
                    changes[current_path] = Change(type=ChangeTypes.REMOVED, old=old[key], new=None)
                elif key in new and not key in old:
                    changes[current_path] = Change(type=ChangeTypes.ADDED, old=None, new=new[key])
                else:
                    sub_changes = cls.recursive_diff(old[key], new[key], current_path)
                    changes.update(sub_changes)

        elif old != new:
            changes[path] = Change(type=ChangeTypes.UPDATED, old=old, new=new)

        return changes


    @classmethod
    def apply_changes(cls, target: dict[Hashable, Any], changes: dict[tuple[Hashable, ...], Change]) -> None:
        """
        Applies a set of changes to the target dictionary.


        Args:
            target: The dictionary to apply the changes to.
            changes: A mapping of paths to changes. The paths are tuples of keys that lead to the value that needs to changes. The changes are applied in the dictionary on that level.
        """

        for path, change in changes.items():
            cursor = target

            # Navigate down the dictionary
            for part in path[:-1]:
                if part not in cursor:
                    cursor[part] = {} # If the path was created because of ADDED
                cursor = cursor[part]

            last_key = path[-1]

            if change.type == ChangeTypes.REMOVED:
                print("DELETE KEY:")
                print(last_key)
                if last_key in cursor:
                    del cursor[last_key]
                else:
                    raise KeyError(f"Unable to remove key: {last_key} not found in target dictionary under path {path} from {target}")

            else:
                # UPDATED or ADDED
                cursor[last_key] = change.new

find_conflicts(changes) classmethod

Finds conflicts in a list of changes.

Parameters:

Name Type Description Default
changes list[dict[tuple[Hashable, ...], Change]]

A list of dictionaries representing changes to a state.

 required

Returns:

Type Description
dict[tuple[Hashable, ...], list[Change]]

A dictionary mapping a path in the state as a list of keys to lists of conflicting changes directly under that path.
Source code in src/edgygraph/diff.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
@classmethod
def find_conflicts(cls, changes: list[dict[tuple[Hashable, ...], Change]]) -> dict[tuple[Hashable, ...], list[Change]]:
    """
    Finds conflicts in a list of changes.

    Args:
       changes: A list of dictionaries representing changes to a state.

    Returns:
        A dictionary mapping a path in the state as a list of keys to lists of conflicting changes directly under that path.
    """

    if len(changes) <= 1:
        return {}

    counts = Counter(key for d in changes for key in d)

    duplicate_keys = [k for k, count in counts.items() if count > 1]

    conflicts: dict[tuple[Hashable, ...], list[Change]] = {}        
    for key in duplicate_keys:
        conflicts[key] = [d[key] for d in changes if key in d]

    return conflicts

recursive_diff(old, new, path=None) classmethod

Recursively computes the differences between two dictionaries.

Parameters:

Name Type Description Default
old Any

Part of the old dictionary.

 required
new Any

Part of the new dictionary.

 required
path tuple[Hashable, ...] | None

The current path of the parts in the full dictionary as a list of keys from least to most specific.

 None

Returns:

Type Description
dict[tuple[Hashable, ...], Change]

A mapping of the path to the changes directly on that level.
Source code in src/edgygraph/diff.py 
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
@classmethod
def recursive_diff(cls, old: Any, new: Any, path: tuple[Hashable, ...] | None = None) -> dict[tuple[Hashable, ...], Change]:
    """
    Recursively computes the differences between two dictionaries.


    Args:
        old: Part of the old dictionary.
        new: Part of the new dictionary.
        path: The current path of the parts in the full dictionary as a list of keys from least to most specific.

    Returns:
        A mapping of the path to the changes directly on that level.
    """

    path = path or ()
    changes: dict[tuple[Hashable, ...], Change] = {}

    if isinstance(old, dict) and isinstance(new, dict):
        all_keys: set[str] = set(old.keys()) | set(new.keys()) #type: ignore

        for key in all_keys:
            current_path: tuple[Hashable, ...] = (*path, key)

            if key in old and not key in new:
                changes[current_path] = Change(type=ChangeTypes.REMOVED, old=old[key], new=None)
            elif key in new and not key in old:
                changes[current_path] = Change(type=ChangeTypes.ADDED, old=None, new=new[key])
            else:
                sub_changes = cls.recursive_diff(old[key], new[key], current_path)
                changes.update(sub_changes)

    elif old != new:
        changes[path] = Change(type=ChangeTypes.UPDATED, old=old, new=new)

    return changes

apply_changes(target, changes) classmethod

Applies a set of changes to the target dictionary.

Parameters:

Name Type Description Default
target dict[Hashable, Any]

The dictionary to apply the changes to.

 required
changes dict[tuple[Hashable, ...], Change]

A mapping of paths to changes. The paths are tuples of keys that lead to the value that needs to changes. The changes are applied in the dictionary on that level.

 required
Source code in src/edgygraph/diff.py 
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
@classmethod
def apply_changes(cls, target: dict[Hashable, Any], changes: dict[tuple[Hashable, ...], Change]) -> None:
    """
    Applies a set of changes to the target dictionary.


    Args:
        target: The dictionary to apply the changes to.
        changes: A mapping of paths to changes. The paths are tuples of keys that lead to the value that needs to changes. The changes are applied in the dictionary on that level.
    """

    for path, change in changes.items():
        cursor = target

        # Navigate down the dictionary
        for part in path[:-1]:
            if part not in cursor:
                cursor[part] = {} # If the path was created because of ADDED
            cursor = cursor[part]

        last_key = path[-1]

        if change.type == ChangeTypes.REMOVED:
            print("DELETE KEY:")
            print(last_key)
            if last_key in cursor:
                del cursor[last_key]
            else:
                raise KeyError(f"Unable to remove key: {last_key} not found in target dictionary under path {path} from {target}")

        else:
            # UPDATED or ADDED
            cursor[last_key] = change.new

ChangeConflictException

Bases: Exception

Exception raised when a conflict between changes to a state is detected.

Source code in src/edgygraph/diff.py 
135
136
137
138
139
class ChangeConflictException(Exception):
    """
    Exception raised when a conflict between changes to a state is detected.
    """
    pass