xdol.updating#

mapping using customizable policies to determine which items to update. It offers a fine-grained approach to dictionary updates beyond the standard dict.update() method.

Key Features:

  • Multiple built-in update policies (always update, update if different, prefer target, prefer source)

  • Support for custom update decision functions

  • Lazy value retrieval for performance optimization

  • Detailed statistics about update operations

  • Specialized update strategies (content hash comparison, timestamp-based updates)

  • File system-specific convenience functions

Custom update logic with a hash function:

>>> target = {"a": "hello", "b": "world"}
>>> source = {"a": "hello!", "c": "python"}
>>> update_by_content_hash(target, source, hash_function=lambda x: len(x))
{'examined': 2, 'updated': 1, 'added': 1, 'unchanged': 0, 'deleted': 0}
>>> target
{'a': 'hello!', 'b': 'world', 'c': 'python'}

Using convenience methods:

>>> update_with_policy.missing_only(target, source)
{'examined': 2, 'updated': 0, 'added': 0, 'unchanged': 2, 'deleted': 0}
>>> target
{'a': 'hello!', 'b': 'world', 'c': 'python'}

Functions for controlled mapping updates with customizable policies.

This module provides flexible functions to update a target mapping from a source mapping using customizable policies to determine which items to update.

Functions

add_as_attribute_of(obj[, name])

Decorator to add the function as an attribute of the object.

local_file_timestamp(store, key)

Get the modified timestamp of a file in a local file store.

print_all_but_skips(key, decision, *[, ...])

Print all decisions except SKIP.

union_iter(*iterables)

Generator yielding unique hashable items from an iterable of iterables.

update_all(target, source, *[, ...])

Update target with all source values, equivalent to dict.update().

update_by_content_hash(target, source, *, ...)

Update target with source values only if their hash differs.

update_files_by_timestamp(target, source, *)

Update a target file store with files from a source store based on modification times.

update_if_different(target, source, *[, ...])

Update target with source values only if they differ.

update_missing_only(target, source, *[, ...])

Update target with source values only for keys not in target.

update_newer(target, source, *, ...[, ...])

Update target with source values only if source has a newer timestamp.

update_with_policy(target, source, *[, ...])

Update a target mapping with values from a source mapping using a customizable policy.

Classes

DefaultPolicy(*values)

Standard policies for updating mappings.

KeyDecision(*values)

Decisions for individual keys during update.

KeyInfoExtractor(*args, **kwargs)

Protocol for functions that extract comparison information from values.

NotRetrieved()

Sentinel object indicating a value that hasn't been retrieved yet.

StringEnum(new_class_name, /, names, *[, ...])

Enum where members are also (and must be) strings.

UpdateDecider(*args, **kwargs)

Protocol for functions that decide whether to update a key.

UpdateStats([examined, updated, added, ...])

Statistics for an update operation.
class xdol.updating.DefaultPolicy(*values)[source]#

Bases: StringEnum

Standard policies for updating mappings.

class xdol.updating.KeyDecision(*values)[source]#

Bases: StringEnum

Decisions for individual keys during update.

class xdol.updating.KeyInfoExtractor(*args, **kwargs)[source]#

Bases: Protocol

Protocol for functions that extract comparison information from values.

class xdol.updating.NotRetrieved[source]#

Bases: object

Sentinel object indicating a value that hasn’t been retrieved yet.

class xdol.updating.StringEnum(new_class_name, /, names, *, module=None, qualname=None, type=None, start=1, boundary=None)[source]#

Bases: str, Enum

Enum where members are also (and must be) strings.

class xdol.updating.UpdateDecider(*args, **kwargs)[source]#

Bases: Protocol

Protocol for functions that decide whether to update a key.

class xdol.updating.UpdateStats(examined=0, updated=0, added=0, unchanged=0, deleted=0)[source]#

Bases: object

Statistics for an update operation.

as_dict()[source]#

Convert stats to a dictionary.

Return type:

dict[str, int]

xdol.updating.add_as_attribute_of(obj, name=None)[source]#

Decorator to add the function as an attribute of the object.

Parameters:

  • obj – The object to which the function will be added

  • name – The name of the attribute to be added, if None, uses the function name

Returns:

The decorator function

Examples

>>> def foo():
...     return "bar"
...
>>> @add_as_attribute_of(foo)
... def baz():
...     return "qux"
...
>>> foo.baz()
'qux'
xdol.updating.local_file_timestamp(store, key)[source]#

Get the modified timestamp of a file in a local file store.

Uses inner_most_key to handle relative paths correctly, resolving to the full path before getting the timestamp.

Parameters:

  • store – A mapping whose keys resolve to file paths

  • key – A key in the store

Returns:

The modification timestamp of the file

Return type:

float

xdol.updating.print_all_but_skips(key, decision, *, print_func=<built-in function print>)[source]#

Print all decisions except SKIP.

xdol.updating.union_iter(*iterables)[source]#

Generator yielding unique hashable items from an iterable of iterables.

Parameters:

iterables (Iterable[TypeVar(T)]) – An iterable where each element is itself an iterable of hashable items.

Yields:

Unique items from the nested iterables, in order of first appearance.

Return type:

Generator[T, None, None]

Example

>>> list(union_iter(['a', 'b'], ['b', 'c']))
['a', 'b', 'c']
>>> list(union_iter({'a': 1, 'b': 2}.keys(), {'b': 3, 'c': 4}.values()))
['a', 'b', 3, 4]
xdol.updating.update_all(target, source, *, keys_to_consider=<function _just_source_keys>, verbose=False)[source]#

Update target with all source values, equivalent to dict.update().

Parameters:
Return type:

dict[str, int]

Returns:

Dictionary with statistics about the update operation

xdol.updating.update_by_content_hash(target, source, *, hash_function, keys_to_consider=<function _just_source_keys>, verbose=False)[source]#

Update target with source values only if their hash differs.

Parameters:
Return type:

dict[str, int]

Returns:

Dictionary with statistics about the update operation

xdol.updating.update_files_by_timestamp(target, source, *, keys_to_consider=<function _just_source_keys>, verbose=False)[source]#

Update a target file store with files from a source store based on modification times.

This is a convenience wrapper around update_newer that uses local_file_timestamp to compare file modification times.

Parameters:
Return type:

dict[str, int]

Returns:

Dictionary with statistics about the update operation

xdol.updating.update_if_different(target, source, *, key_info=None, target_key_info=None, source_key_info=None, keys_to_consider=<function _just_source_keys>, verbose=False)[source]#

Update target with source values only if they differ.

Parameters:

  • target (MutableMapping[TypeVar(K), TypeVar(V)]) – The mapping to be updated (modified in-place)

  • source (Mapping[TypeVar(K), TypeVar(V)]) – The mapping containing items to potentially copy to target

  • key_info – (Deprecated) Function to extract comparison info from values

  • target_key_info – Function to extract comparison info from target keys

  • source_key_info – Function to extract comparison info from source keys

  • keys_to_consider (Union[Iterable[TypeVar(K)], Callable[[Mapping, MutableMapping], Iterable[TypeVar(K)]]]) – Specific iterable of keys to consider or a callable to make this from source and target. The default will consider source keys only.

  • verbose (bool | Callable[[TypeVar(K), KeyDecision], Any]) – If True, print debug information, if callable, call with (key, decision) on each key that is looped over

Return type:

dict[str, int]

Returns:

Dictionary with statistics about the update operation

xdol.updating.update_missing_only(target, source, *, keys_to_consider=<function _just_source_keys>, verbose=False)[source]#

Update target with source values only for keys not in target.

Parameters:
Return type:

dict[str, int]

Returns:

Dictionary with statistics about the update operation

xdol.updating.update_newer(target, source, *, target_timestamp, source_timestamp, keys_to_consider=<function _just_source_keys>, verbose=False)[source]#

Update target with source values only if source has a newer timestamp.

Parameters:
Return type:

dict[str, int]

Returns:

Dictionary with statistics about the update operation

xdol.updating.update_with_policy(target, source, *, policy=DefaultPolicy.UPDATE_IF_DIFFERENT, target_key_info=None, source_key_info=None, keys_to_consider=<function _just_source_keys>, verbose=False)[source]#

Update a target mapping with values from a source mapping using a customizable policy.

Parameters:
Return type:

dict[str, int]

Returns:

Dictionary with statistics about the update operation

Examples

>>> target = {"a": 1, "b": 2}
>>> source = {"a": 10, "c": 30}
>>> update_with_policy(target, source)
{'examined': 2, 'updated': 1, 'added': 1, 'unchanged': 0, 'deleted': 0}
>>> target
{'a': 10, 'b': 2, 'c': 30}

>>> # Using PREFER_TARGET policy
>>> target = {"a": 1, "b": 2}
>>> source = {"a": 10, "c": 30}
>>> update_with_policy(target, source, policy=DefaultPolicy.PREFER_TARGET)
{'examined': 2, 'updated': 0, 'added': 1, 'unchanged': 1, 'deleted': 0}
>>> target
{'a': 1, 'b': 2, 'c': 30}