diffsync.helpers

DiffSync helper classes for calculating and performing diff and sync operations.

class diffsync.helpers.DiffSyncDiffer(src_diffsync: Adapter, dst_diffsync: Adapter, flags: ~diffsync.enum.DiffSyncFlags, diff_class: ~typing.Type[~diffsync.diff.Diff] = <class 'diffsync.diff.Diff'>, callback: ~typing.Callable[[str, int, int], None] | None = None)

Bases: object

Helper class implementing diff calculation logic for DiffSync.

Independent from Diff and DiffElement as those classes are purely data objects, while this stores some state.

__init__(src_diffsync: Adapter, dst_diffsync: Adapter, flags: ~diffsync.enum.DiffSyncFlags, diff_class: ~typing.Type[~diffsync.diff.Diff] = <class 'diffsync.diff.Diff'>, callback: ~typing.Callable[[str, int, int], None] | None = None)

Create a DiffSyncDiffer for calculating diffs between the provided DiffSync instances.

calculate_diffs() → Diff

Calculate diffs between the src and dst DiffSync objects and return the resulting Diff.

diff_child_objects(diff_element: DiffElement, src_obj: DiffSyncModel | None, dst_obj: DiffSyncModel | None) → DiffElement

For all children of the given DiffSyncModel pair, diff recursively, adding diffs to the given diff_element.

Helper method to calculate_diffs, usually doesn’t need to be called directly.

These helper methods work in a recursive cycle: diff_object_list -> diff_object_pair -> diff_child_objects -> diff_object_list -> etc.

diff_object_list(src: List[DiffSyncModel], dst: List[DiffSyncModel]) → List[DiffElement]

Calculate diffs between two lists of like objects.

Helper method to calculate_diffs, usually doesn’t need to be called directly.

These helper methods work in a recursive cycle: diff_object_list -> diff_object_pair -> diff_child_objects -> diff_object_list -> etc.

diff_object_pair(src_obj: DiffSyncModel | None, dst_obj: DiffSyncModel | None) → DiffElement | None

Diff the two provided DiffSyncModel objects and return a DiffElement or None.

Helper method to calculate_diffs, usually doesn’t need to be called directly.

These helper methods work in a recursive cycle: diff_object_list -> diff_object_pair -> diff_child_objects -> diff_object_list -> etc.

incr_models_processed(delta: int = 1) → None

Increment self.models_processed, then call self.callback if present.

static validate_objects_for_diff(object_pairs: Iterable[Tuple[DiffSyncModel | None, DiffSyncModel | None]]) → None

Check whether all DiffSyncModels in the given dictionary are valid for comparison to one another.

Helper method for diff_object_list.

Raises:

• TypeError – If any pair of objects in the dict have differing get_type() values.

• ValueError – If any pair of objects in the dict have differing get_shortname() or get_identifiers() values.

class diffsync.helpers.DiffSyncSyncer(diff: Diff, src_diffsync: Adapter, dst_diffsync: Adapter, flags: DiffSyncFlags, callback: Callable[[str, int, int], None] | None = None)

Bases: object

Helper class implementing data synchronization logic for DiffSync.

Independent from DiffSync and DiffSyncModel as those classes are purely data objects, while this stores some state.

__init__(diff: Diff, src_diffsync: Adapter, dst_diffsync: Adapter, flags: DiffSyncFlags, callback: Callable[[str, int, int], None] | None = None)

Create a DiffSyncSyncer instance, ready to call perform_sync() against.

incr_elements_processed(delta: int = 1) → None

Increment self.elements_processed, then call self.callback if present.

log_sync_status(action: str | None, status: DiffSyncStatus, message: str) → None

Log the current sync status at the appropriate verbosity with appropriate context.

Helper method to sync_diff_element/sync_model.

perform_sync() → bool

Perform data synchronization based on the provided diff.

Returns:

True if any changes were actually performed, else False.

sync_diff_element(element: DiffElement, parent_model: DiffSyncModel | None = None) → bool

Recursively synchronize the given DiffElement and its children, if any, into the dst_diffsync.

Helper method to perform_sync.

Returns:

True if this element or any of its children resulted in actual changes, else False.

Return type:

bool

sync_model(src_model: DiffSyncModel | None, dst_model: DiffSyncModel | None, ids: Dict, attrs: Dict) → Tuple[bool, DiffSyncModel | None]

Create/update/delete the current DiffSyncModel with current ids/attrs, and update self.status and self.message.

Helper method to sync_diff_element.

Returns:

(changed, model) where model may be None if an error occurred