Skip to content

anymerge

anymerge

Reducer

Reducer(reducer: ReducerCallable) -> ReducerInfo
Reducer(*, deep: Literal[True]) -> ReducerInfo
Reducer(reducer: EllipsisType) -> ReducerInfo
Reducer(
    reducer: ReducerCallable | EllipsisType | None = None,
    *,
    deep: bool | None = None,
) -> ReducerInfo

Create a ReducerInfo object with the given reducer callable or deep set to True.

PARAMETER DESCRIPTION
reducer

The reducer callable to apply to the field. If set to ..., the reducer will be applied deeply.

TYPE: ReducerCallable | EllipsisType | None DEFAULT: None

deep

Whether to apply the reducer deeply.

TYPE: bool | None DEFAULT: None

RETURNS DESCRIPTION
ReducerInfo

A ReducerInfo object with the given reducer callable or deep set to True.

Examples:

Define a reducer callable:

>>> Reducer(operator.add)

To apply a reducer deeply, either use ... as the reducer argument or set deep to True:

>>> Reducer(...)
>>> Reducer(deep=True)
Source code in src/anymerge/_api.py
def Reducer(  # noqa: N802
    reducer: ReducerCallable | types.EllipsisType | None = None,
    *,
    deep: bool | None = None,
) -> models.ReducerInfo:
    """Create a [`ReducerInfo`][anymerge.models.ReducerInfo] object with the given `reducer` callable or `deep` set to True.

    Args:
        reducer: The reducer callable to apply to the field. If set to `...`, the reducer will be applied deeply.
        deep: Whether to apply the reducer deeply.

    Returns:
        A [`ReducerInfo`][anymerge.models.ReducerInfo] object with the given `reducer` callable or `deep` set to True.

    Examples:
        Define a reducer callable:

            >>> Reducer(operator.add)

        To apply a reducer deeply, either use `...` as the `reducer` argument or set `deep` to `True`:

            >>> Reducer(...)
            >>> Reducer(deep=True)
    """
    if reducer is ...:
        if deep is False:
            msg = "deep cannot be False when reducer is ..."
            raise AnyMergeValueError(msg)
        return models.ReducerInfo(None, deep=True)

    if reducer is None:
        if deep is False:
            msg = "deep cannot be False when reducer is None"
            raise AnyMergeValueError(msg)
        return models.ReducerInfo(None, deep=True)

    if deep is True:
        msg = "deep cannot be True when reducer is provided"
        raise AnyMergeValueError(msg)

    return models.ReducerInfo(reducer=reducer, deep=False)

merge

merge(
    a: T,
    b: Any,
    *,
    default_reducer: ReducerInfo = DEFAULT_REDUCER,
) -> T

Merge two instances of data models.

PARAMETER DESCRIPTION
a

The first instance to merge. This should contain all the annotations with the reducer information.

TYPE: T

b

The second instance to merge.

TYPE: Any

default_reducer

The default reducer to apply to fields.

TYPE: ReducerInfo DEFAULT: DEFAULT_REDUCER

RETURNS DESCRIPTION
T

The merged instance.

Source code in src/anymerge/_merger.py
def merge(
    a: T,
    b: typing.Any,
    *,
    default_reducer: ReducerInfo = DEFAULT_REDUCER,
) -> T:
    """
    Merge two instances of data models.

    Args:
        a: The first instance to merge. This should contain all the annotations with the reducer information.
        b: The second instance to merge.
        default_reducer: The default reducer to apply to fields.

    Returns:
        The merged instance.
    """
    adapted_a = maybe_adapt_value(a)
    adapted_b = maybe_adapt_value(b)
    fields = adapted_a.adapter.get_fields()
    a_values = adapted_a.get_values()
    b_values = adapted_b.get_values()
    changes = {
        key: apply_reducers(
            a_values[key],
            value,
            field_info=fields[key],
            default_reducer=default_reducer,
        )
        for key, value in b_values.items()
        if key in fields
    }

    return adapted_a.copy(changes=changes)

reduce

reduce(
    initial: T,
    iterable: Iterable[Any],
    *,
    default_reducer: ReducerInfo = DEFAULT_REDUCER,
) -> T

Reduce an iterable of instances of data models.

PARAMETER DESCRIPTION
initial

The initial instance to reduce. This should contain all the annotations with the reducer information.

TYPE: T

iterable

The iterable of instances to reduce.

TYPE: Iterable[Any]

RETURNS DESCRIPTION
T

The reduced instance.

Source code in src/anymerge/_reducer.py
def reduce(
    initial: T,
    iterable: typing.Iterable[typing.Any],
    *,
    default_reducer: ReducerInfo = DEFAULT_REDUCER,
) -> T:
    """Reduce an iterable of instances of data models.

    Args:
        initial: The initial instance to reduce. This should contain all the annotations with the reducer information.
        iterable: The iterable of instances to reduce.

    Returns:
        The reduced instance.
    """
    value = initial
    for item in iterable:
        value = merge(value, item, default_reducer=default_reducer)
    return value