Skip to content

Dict

Bases: DictCommonMethods[K, V]

Wrapper for Python dictionaries with chainable methods.

Source code in src/pyochain/_dict/_main.py
 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
class Dict[K, V](DictCommonMethods[K, V]):
    """Wrapper for Python dictionaries with chainable methods."""

    __slots__ = ()

    @staticmethod
    def from_[G, I](
        data: Mapping[G, I] | Iterable[tuple[G, I]] | SupportsKeysAndGetItem[G, I],
    ) -> Dict[G, I]:
        """Create a `Dict` from a convertible value.

        Args:
            data (Mapping[G, I] | Iterable[tuple[G, I]] | SupportsKeysAndGetItem[G, I]): Object convertible into a Dict.

        Returns:
            Dict[G, I]: Instance containing the data from the input.

        Example:
        ```python
        >>> import pyochain as pc
        >>> class MyMapping:
        ...     def __init__(self):
        ...         self._data = {1: "a", 2: "b", 3: "c"}
        ...
        ...     def keys(self):
        ...         return self._data.keys()
        ...
        ...     def __getitem__(self, key):
        ...         return self._data[key]
        >>>
        >>> pc.Dict.from_(MyMapping()).inner()
        {1: 'a', 2: 'b', 3: 'c'}
        >>> pc.Dict.from_([("d", "e"), ("f", "g")]).inner()
        {'d': 'e', 'f': 'g'}

        ```
        """
        return Dict(dict(data))

    @staticmethod
    def from_object(obj: object) -> Dict[str, Any]:
        """Create a `Dict` from an object `__dict__` attribute.

        We can't know in advance the values types, so we use `Any`.

        Args:
            obj (object): The object whose `__dict__` attribute will be used to create the Dict.

        Returns:
            Dict[str, Any]: A new Dict instance containing the attributes of the object.

        ```python
        >>> import pyochain as pc
        >>> class Person:
        ...     def __init__(self, name: str, age: int):
        ...         self.name = name
        ...         self.age = age
        >>> person = Person("Alice", 30)
        >>> pc.Dict.from_object(person).inner()
        {'name': 'Alice', 'age': 30}

        ```
        """
        return Dict(obj.__dict__)

    def pivot(self, *indices: int) -> Dict[Any, Any]:
        """Pivot a nested dictionary by rearranging the key levels according to order.

        Syntactic sugar for `Dict.to_arrays().rearrange(*indices).to_records()`

        Args:
            *indices (int): Indices specifying the new order of key levels

        Returns:
            Dict[Any, Any]: Pivoted dictionary with keys rearranged

        Example:
        ```python
        >>> import pyochain as pc
        >>> d = {"A": {"X": 1, "Y": 2}, "B": {"X": 3, "Y": 4}}
        >>> pc.Dict(d).pivot(1, 0).inner()
        {'X': {'A': 1, 'B': 3}, 'Y': {'A': 2, 'B': 4}}
        """
        return self.to_arrays().rearrange(*indices).to_records()

diff

diff(other: Mapping[K, V]) -> Dict[K, tuple[V | None, V | None]]

Returns a dict of the differences between this dict and another.

Parameters:

Name Type Description Default
other Mapping[K, V]

The mapping to compare against.

required

Returns:

Type Description
Dict[K, tuple[V | None, V | None]]

Dict[K, tuple[V | None, V | None]]: Dict with differences as (self_value, other_value) tuples.

The keys of the returned dict are the keys that are not shared or have different values. The values are tuples containing the value from self and the value from other.

>>> import pyochain as pc
>>> d1 = {"a": 1, "b": 2, "c": 3}
>>> d2 = {"b": 2, "c": 4, "d": 5}
>>> pc.Dict(d1).diff(d2).sort().inner()
{'a': (1, None), 'c': (3, 4), 'd': (None, 5)}

Source code in src/pyochain/_dict/_joins.py
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
def diff(self, other: Mapping[K, V]) -> Dict[K, tuple[V | None, V | None]]:
    """Returns a dict of the differences between this dict and another.

    Args:
        other(Mapping[K, V]): The mapping to compare against.

    Returns:
        Dict[K, tuple[V | None, V | None]]: Dict with differences as (self_value, other_value) tuples.

    The keys of the returned dict are the keys that are not shared or have different values.
    The values are tuples containing the value from self and the value from other.
    ```python
    >>> import pyochain as pc
    >>> d1 = {"a": 1, "b": 2, "c": 3}
    >>> d2 = {"b": 2, "c": 4, "d": 5}
    >>> pc.Dict(d1).diff(d2).sort().inner()
    {'a': (1, None), 'c': (3, 4), 'd': (None, 5)}

    ```
    """

    def _diff(data: Mapping[K, V]) -> dict[K, tuple[V | None, V | None]]:
        all_keys: set[K] = data.keys() | other.keys()
        diffs: dict[K, tuple[V | None, V | None]] = {}
        for key in all_keys:
            self_val = data.get(key)
            other_val = other.get(key)
            if self_val != other_val:
                diffs[key] = (self_val, other_val)
        return diffs

    return self._new(_diff)

diff_keys

diff_keys(*others: Mapping[K, V]) -> Dict[K, V]

Keep only keys present in self but not in others mappings.

Parameters:

Name Type Description Default
*others Mapping[K, V]

Other mappings to exclude keys from.

()

Returns:

Type Description
Dict[K, V]

Dict[K, V]: A new Dict with only the differing keys.

>>> import pyochain as pc
>>> d1 = {"a": 1, "b": 2, "c": 3}
>>> d2 = {"b": 10, "d": 40}
>>> d3 = {"c": 30}
>>> pc.Dict(d1).diff_keys(d2, d3)
{'a': 1}
Source code in src/pyochain/_dict/_filters.py
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
def diff_keys(self, *others: Mapping[K, V]) -> Dict[K, V]:
    """Keep only keys present in self but not in others mappings.

    Args:
        *others (Mapping[K, V]): Other mappings to exclude keys from.

    Returns:
        Dict[K, V]: A new Dict with only the differing keys.
    ```python
    >>> import pyochain as pc
    >>> d1 = {"a": 1, "b": 2, "c": 3}
    >>> d2 = {"b": 10, "d": 40}
    >>> d3 = {"c": 30}
    >>> pc.Dict(d1).diff_keys(d2, d3)
    {'a': 1}

    ```
    """

    def _diff_keys(data: dict[K, V]) -> dict[K, V]:
        self_keys = set(data.keys())
        for other in others:
            self_keys.difference_update(other.keys())
        return {k: data[k] for k in self_keys}

    return self._new(_diff_keys)

drop

drop(*keys: K) -> Dict[K, V]

Return a new Dict with given keys removed.

Parameters:

Name Type Description Default
*keys K

keys to remove from the dictionary.

()

Returns:

Type Description
Dict[K, V]

Dict[K, V]: New Dict with specified keys removed.

New dict has d[key] deleted for each supplied key.

>>> import pyochain as pc
>>> pc.Dict({"x": 1, "y": 2}).drop("y").inner()
{'x': 1}
>>> pc.Dict({"x": 1, "y": 2}).drop("y", "x").inner()
{}
>>> pc.Dict({"x": 1}).drop("y").inner()  # Ignores missing keys
{'x': 1}
>>> pc.Dict({1: 2, 3: 4}).drop(1).inner()
{3: 4}

Source code in src/pyochain/_dict/_process.py
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
def drop(self, *keys: K) -> Dict[K, V]:
    """Return a new Dict with given keys removed.

    Args:
        *keys (K): keys to remove from the dictionary.

    Returns:
        Dict[K, V]: New Dict with specified keys removed.

    New dict has d[key] deleted for each supplied key.
    ```python
    >>> import pyochain as pc
    >>> pc.Dict({"x": 1, "y": 2}).drop("y").inner()
    {'x': 1}
    >>> pc.Dict({"x": 1, "y": 2}).drop("y", "x").inner()
    {}
    >>> pc.Dict({"x": 1}).drop("y").inner()  # Ignores missing keys
    {'x': 1}
    >>> pc.Dict({1: 2, 3: 4}).drop(1).inner()
    {3: 4}

    ```
    """

    def _drop(data: dict[K, V]) -> dict[K, V]:
        return cz.dicttoolz.dissoc(data, *keys)

    return self._new(_drop)

drop_nones

drop_nones(*, remove_empty: bool = True) -> Dict[K, V]

Recursively drop None values from the dictionary.

Options to also remove empty dicts and lists.

Parameters:

Name Type Description Default
remove_empty bool

If True (default), removes None, {} and [].

True

Returns:

Type Description
Dict[K, V]

Dict[K, V]: Dict with None values and optionally empty structures removed.

Example:

>>> import pyochain as pc
>>> data = {
...     "a": 1,
...     "b": None,
...     "c": {},
...     "d": [],
...     "e": {"f": None, "g": 2},
...     "h": [1, None, {}],
...     "i": 0,
... }
>>> p_data = pc.Dict(data)
>>>
>>> p_data.drop_nones().inner()
{'a': 1, 'e': {'g': 2}, 'h': [1], 'i': 0}
>>>
>>> p_data.drop_nones().inner()
{'a': 1, 'e': {'g': 2}, 'h': [1], 'i': 0}
>>>
>>> p_data.drop_nones(remove_empty=False).inner()
{'a': 1, 'b': None, 'c': {}, 'd': [], 'e': {'f': None, 'g': 2}, 'h': [1, None, {}], 'i': 0}

Source code in src/pyochain/_dict/_nested.py
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
def drop_nones(self, *, remove_empty: bool = True) -> Dict[K, V]:
    """Recursively drop None values from the dictionary.

    Options to also remove empty dicts and lists.

    Args:
        remove_empty (bool): If True (default), removes `None`, `{}` and `[]`.

    Returns:
        Dict[K, V]: Dict with None values and optionally empty structures removed.

    Example:
    ```python
    >>> import pyochain as pc
    >>> data = {
    ...     "a": 1,
    ...     "b": None,
    ...     "c": {},
    ...     "d": [],
    ...     "e": {"f": None, "g": 2},
    ...     "h": [1, None, {}],
    ...     "i": 0,
    ... }
    >>> p_data = pc.Dict(data)
    >>>
    >>> p_data.drop_nones().inner()
    {'a': 1, 'e': {'g': 2}, 'h': [1], 'i': 0}
    >>>
    >>> p_data.drop_nones().inner()
    {'a': 1, 'e': {'g': 2}, 'h': [1], 'i': 0}
    >>>
    >>> p_data.drop_nones(remove_empty=False).inner()
    {'a': 1, 'b': None, 'c': {}, 'd': [], 'e': {'f': None, 'g': 2}, 'h': [1, None, {}], 'i': 0}

    ```
    """

    def _apply_drop_nones(data: dict[K, V]) -> dict[Any, Any]:
        result = _drop_nones(data, remove_empty=remove_empty)
        return result if isinstance(result, dict) else {}

    return self._new(_apply_drop_nones)

eq

eq(other: Self | T) -> bool

Check if two records are equal based on their data.

Parameters:

Name Type Description Default
other Self | T

Another instance or corresponding underlying data to compare against.

required

Returns:

Name Type Description
bool bool

True if the underlying data are equal, False otherwise.

Example:

>>> import pyochain as pc
>>> d1 = pc.Dict({"a": 1, "b": 2})
>>> d2 = pc.Dict({"a": 1, "b": 2})
>>> d3 = pc.Dict({"a": 1, "b": 3})
>>> d1.eq(d2)
True
>>> d1.eq(d3)
False

Source code in src/pyochain/_core/_main.py
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
def eq(self, other: Self | T) -> bool:
    """Check if two records are equal based on their data.

    Args:
        other (Self | T): Another instance or corresponding underlying data to compare against.

    Returns:
        bool: True if the underlying data are equal, False otherwise.

    Example:
    ```python
    >>> import pyochain as pc
    >>> d1 = pc.Dict({"a": 1, "b": 2})
    >>> d2 = pc.Dict({"a": 1, "b": 2})
    >>> d3 = pc.Dict({"a": 1, "b": 3})
    >>> d1.eq(d2)
    True
    >>> d1.eq(d3)
    False

    ```
    """
    other_data = other._inner if isinstance(other, self.__class__) else other
    return self._inner == other_data

filter_items

filter_items(predicate: Callable[[tuple[K, V]], bool]) -> Dict[K, V]

Filter items by predicate applied to (key, value) tuples.

Parameters:

Name Type Description Default
predicate Callable[[tuple[K, V]], bool]

Function to determine if a (key, value) pair should be included.

required

Returns:

Type Description
Dict[K, V]

Dict[K, V]: A new Dict instance containing only the items that satisfy the predicate.

Example:

>>> import pyochain as pc
>>> def isvalid(item):
...     k, v = item
...     return k % 2 == 0 and v < 4
>>> d = pc.Dict({1: 2, 2: 3, 3: 4, 4: 5})
>>>
>>> d.filter_items(isvalid)
{2: 3}
>>> d.filter_items(lambda kv: not isvalid(kv))
{1: 2, 3: 4, 4: 5}

Source code in src/pyochain/_dict/_filters.py
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
def filter_items(self, predicate: Callable[[tuple[K, V]], bool]) -> Dict[K, V]:
    """Filter items by predicate applied to (key, value) tuples.

    Args:
        predicate (Callable[[tuple[K, V]], bool]): Function to determine if a (key, value) pair should be included.

    Returns:
        Dict[K, V]: A new Dict instance containing only the items that satisfy the predicate.

    Example:
    ```python
    >>> import pyochain as pc
    >>> def isvalid(item):
    ...     k, v = item
    ...     return k % 2 == 0 and v < 4
    >>> d = pc.Dict({1: 2, 2: 3, 3: 4, 4: 5})
    >>>
    >>> d.filter_items(isvalid)
    {2: 3}
    >>> d.filter_items(lambda kv: not isvalid(kv))
    {1: 2, 3: 4, 4: 5}

    ```
    """
    return self._new(partial(cz.dicttoolz.itemfilter, predicate))

filter_keys

filter_keys(predicate: Callable[[K], TypeIs[U]]) -> Dict[U, V]
filter_keys(predicate: Callable[[K], bool]) -> Dict[K, V]
filter_keys(predicate: Callable[[K], bool | TypeIs[U]]) -> Dict[K, V] | Dict[U, V]

Return keys that satisfy predicate.

Parameters:

Name Type Description Default
predicate Callable[[K], bool | TypeIs[U]]

Function to determine if a key should be included.

required

Returns:

Type Description
Dict[K, V] | Dict[U, V]

Dict[K, V] | Dict[U, V]: Filtered Dict with keys satisfying predicate.

Example:

>>> import pyochain as pc
>>> d = {1: 2, 2: 3, 3: 4, 4: 5}
>>> pc.Dict(d).filter_keys(lambda x: x % 2 == 0)
{2: 3, 4: 5}

Source code in src/pyochain/_dict/_filters.py
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
def filter_keys[U](
    self,
    predicate: Callable[[K], bool | TypeIs[U]],
) -> Dict[K, V] | Dict[U, V]:
    """Return keys that satisfy predicate.

    Args:
        predicate (Callable[[K], bool | TypeIs[U]]): Function to determine if a key should be included.

    Returns:
        Dict[K, V] | Dict[U, V]: Filtered Dict with keys satisfying predicate.

    Example:
    ```python
    >>> import pyochain as pc
    >>> d = {1: 2, 2: 3, 3: 4, 4: 5}
    >>> pc.Dict(d).filter_keys(lambda x: x % 2 == 0)
    {2: 3, 4: 5}

    ```
    """
    return self._new(partial(cz.dicttoolz.keyfilter, predicate))

filter_kv

filter_kv(predicate: Callable[[K, V], bool]) -> Dict[K, V]

Filter items by predicate applied to unpacked (key, value) tuples.

Parameters:

Name Type Description Default
predicate Callable[[K, V], bool]

Function to determine if a key-value pair should be included.

required

Returns:

Type Description
Dict[K, V]

Dict[K, V]: Filtered Dict with items satisfying predicate.

Example:

>>> import pyochain as pc
>>> def isvalid(key, value):
...     return key % 2 == 0 and value < 4
>>> d = pc.Dict({1: 2, 2: 3, 3: 4, 4: 5})
>>>
>>> d.filter_kv(isvalid)
{2: 3}
>>> d.filter_kv(lambda k, v: not isvalid(k, v))
{1: 2, 3: 4, 4: 5}

Source code in src/pyochain/_dict/_filters.py
 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
def filter_kv(self, predicate: Callable[[K, V], bool]) -> Dict[K, V]:
    """Filter items by predicate applied to unpacked (key, value) tuples.

    Args:
        predicate(Callable[[K, V], bool]): Function to determine if a key-value pair should be included.

    Returns:
        Dict[K, V]: Filtered Dict with items satisfying predicate.

    Example:
    ```python
    >>> import pyochain as pc
    >>> def isvalid(key, value):
    ...     return key % 2 == 0 and value < 4
    >>> d = pc.Dict({1: 2, 2: 3, 3: 4, 4: 5})
    >>>
    >>> d.filter_kv(isvalid)
    {2: 3}
    >>> d.filter_kv(lambda k, v: not isvalid(k, v))
    {1: 2, 3: 4, 4: 5}

    ```
    """

    def _filter_kv(data: dict[K, V]) -> dict[K, V]:
        def _(kv: tuple[K, V]) -> bool:
            return predicate(kv[0], kv[1])

        return cz.dicttoolz.itemfilter(_, data)

    return self._new(_filter_kv)

filter_values

filter_values(predicate: Callable[[V], TypeIs[U]]) -> Dict[K, U]
filter_values(predicate: Callable[[V], bool]) -> Dict[K, V]
filter_values(
    predicate: Callable[[V], bool] | Callable[[V], TypeIs[U]],
) -> Dict[K, V] | Dict[K, U]

Return items whose values satisfy predicate.

Parameters:

Name Type Description Default
predicate Callable[[V], bool] | Callable[[V], TypeIs[U]]

Function to determine if a value should be included.

required

Returns:

Type Description
Dict[K, V] | Dict[K, U]

Dict[K, V] | Dict[K, U]: Filtered Dict with values satisfying predicate

Example:

>>> import pyochain as pc
>>> d = {1: 2, 2: 3, 3: 4, 4: 5}
>>> pc.Dict(d).filter_values(lambda x: x % 2 == 0)
{1: 2, 3: 4}
>>> pc.Dict(d).filter_values(lambda x: not x > 3)
{1: 2, 2: 3}

Source code in src/pyochain/_dict/_filters.py
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
def filter_values[U](
    self,
    predicate: Callable[[V], bool] | Callable[[V], TypeIs[U]],
) -> Dict[K, V] | Dict[K, U]:
    """Return items whose values satisfy predicate.

    Args:
        predicate (Callable[[V], bool] | Callable[[V], TypeIs[U]]): Function to determine if a value should be included.

    Returns:
        Dict[K, V] | Dict[K, U]: Filtered Dict with values satisfying predicate

    Example:
    ```python
    >>> import pyochain as pc
    >>> d = {1: 2, 2: 3, 3: 4, 4: 5}
    >>> pc.Dict(d).filter_values(lambda x: x % 2 == 0)
    {1: 2, 3: 4}
    >>> pc.Dict(d).filter_values(lambda x: not x > 3)
    {1: 2, 2: 3}

    ```
    """
    return self._new(partial(cz.dicttoolz.valfilter, predicate))

flatten

flatten(sep: str = '.', max_depth: int | None = None) -> Dict[str, Any]

Flatten a nested dictionary, concatenating keys with the specified separator.

Parameters:

Name Type Description Default
sep str

Separator to use when concatenating keys

'.'
max_depth int | None

Maximum depth to flatten. If None, flattens completely.

None

Returns:

Type Description
Dict[str, Any]

Dict[str, Any]: Flattened Dict with concatenated keys.

>>> import pyochain as pc
>>> data = {
...     "config": {"params": {"retries": 3, "timeout": 30}, "mode": "fast"},
...     "version": 1.0,
... }
>>> pc.Dict(data).flatten().inner()
{'config.params.retries': 3, 'config.params.timeout': 30, 'config.mode': 'fast', 'version': 1.0}
>>> pc.Dict(data).flatten(sep="_").inner()
{'config_params_retries': 3, 'config_params_timeout': 30, 'config_mode': 'fast', 'version': 1.0}
>>> pc.Dict(data).flatten(max_depth=1).inner()
{'config.params': {'retries': 3, 'timeout': 30}, 'config.mode': 'fast', 'version': 1.0}
Source code in src/pyochain/_dict/_nested.py
 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
132
133
134
135
136
137
138
139
def flatten(
    self: NestedDict[str, Any],
    sep: str = ".",
    max_depth: int | None = None,
) -> Dict[str, Any]:
    """Flatten a nested dictionary, concatenating keys with the specified separator.

    Args:
        sep (str): Separator to use when concatenating keys
        max_depth (int | None): Maximum depth to flatten. If None, flattens completely.

    Returns:
        Dict[str, Any]: Flattened Dict with concatenated keys.
    ```python
    >>> import pyochain as pc
    >>> data = {
    ...     "config": {"params": {"retries": 3, "timeout": 30}, "mode": "fast"},
    ...     "version": 1.0,
    ... }
    >>> pc.Dict(data).flatten().inner()
    {'config.params.retries': 3, 'config.params.timeout': 30, 'config.mode': 'fast', 'version': 1.0}
    >>> pc.Dict(data).flatten(sep="_").inner()
    {'config_params_retries': 3, 'config_params_timeout': 30, 'config_mode': 'fast', 'version': 1.0}
    >>> pc.Dict(data).flatten(max_depth=1).inner()
    {'config.params': {'retries': 3, 'timeout': 30}, 'config.mode': 'fast', 'version': 1.0}

    ```
    """

    def _flatten(
        d: Mapping[Any, Any],
        parent_key: str = "",
        current_depth: int = 1,
    ) -> dict[str, Any]:
        def _can_recurse(v: object) -> TypeIs[Mapping[Any, Any]]:
            return isinstance(v, Mapping) and (
                max_depth is None or current_depth < max_depth + 1
            )

        items: list[tuple[str, Any]] = []
        for k, v in d.items():
            new_key = parent_key + sep + k if parent_key else k
            if _can_recurse(v):
                items.extend(_flatten(v, new_key, current_depth + 1).items())
            else:
                items.append((new_key, v))
        return dict(items)

    return self._new(_flatten)

for_each

for_each(
    func: Callable[Concatenate[K, V, P], Any], *args: P.args, **kwargs: P.kwargs
) -> Dict[K, V]

Apply a function to each key-value pair in the dict for side effects.

Parameters:

Name Type Description Default
func Callable[Concatenate[K, V, P], Any]

Function to apply to each key-value pair.

required
*args P.args

Positional arguments to pass to the function.

()
**kwargs P.kwargs

Keyword arguments to pass to the function.

{}

Returns:

Type Description
Dict[K, V]

Dict[K, V]: The original Dict unchanged.

Returns the original Dict unchanged.

>>> import pyochain as pc
>>> pc.Dict({"a": 1, "b": 2}).for_each(lambda k, v: print(f"Key: {k}, Value: {v}"))
Key: a, Value: 1
Key: b, Value: 2
{'a': 1, 'b': 2}

Source code in src/pyochain/_dict/_process.py
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
def for_each[**P](
    self,
    func: Callable[Concatenate[K, V, P], Any],
    *args: P.args,
    **kwargs: P.kwargs,
) -> Dict[K, V]:
    """Apply a function to each key-value pair in the dict for side effects.

    Args:
        func (Callable[Concatenate[K, V, P], Any]): Function to apply to each key-value pair.
        *args (P.args): Positional arguments to pass to the function.
        **kwargs (P.kwargs): Keyword arguments to pass to the function.

    Returns:
        Dict[K, V]: The original Dict unchanged.

    Returns the original Dict unchanged.
    ```python
    >>> import pyochain as pc
    >>> pc.Dict({"a": 1, "b": 2}).for_each(lambda k, v: print(f"Key: {k}, Value: {v}"))
    Key: a, Value: 1
    Key: b, Value: 2
    {'a': 1, 'b': 2}

    ```
    """

    def _for_each(data: dict[K, V]) -> dict[K, V]:
        for k, v in data.items():
            func(k, v, *args, **kwargs)
        return data

    return self._new(_for_each)

from_ staticmethod

from_(
    data: Mapping[G, I] | Iterable[tuple[G, I]] | SupportsKeysAndGetItem[G, I],
) -> Dict[G, I]

Create a Dict from a convertible value.

Parameters:

Name Type Description Default
data Mapping[G, I] | Iterable[tuple[G, I]] | SupportsKeysAndGetItem[G, I]

Object convertible into a Dict.

required

Returns:

Type Description
Dict[G, I]

Dict[G, I]: Instance containing the data from the input.

Example:

>>> import pyochain as pc
>>> class MyMapping:
...     def __init__(self):
...         self._data = {1: "a", 2: "b", 3: "c"}
...
...     def keys(self):
...         return self._data.keys()
...
...     def __getitem__(self, key):
...         return self._data[key]
>>>
>>> pc.Dict.from_(MyMapping()).inner()
{1: 'a', 2: 'b', 3: 'c'}
>>> pc.Dict.from_([("d", "e"), ("f", "g")]).inner()
{'d': 'e', 'f': 'g'}

Source code in src/pyochain/_dict/_main.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
59
60
61
62
63
64
65
66
67
@staticmethod
def from_[G, I](
    data: Mapping[G, I] | Iterable[tuple[G, I]] | SupportsKeysAndGetItem[G, I],
) -> Dict[G, I]:
    """Create a `Dict` from a convertible value.

    Args:
        data (Mapping[G, I] | Iterable[tuple[G, I]] | SupportsKeysAndGetItem[G, I]): Object convertible into a Dict.

    Returns:
        Dict[G, I]: Instance containing the data from the input.

    Example:
    ```python
    >>> import pyochain as pc
    >>> class MyMapping:
    ...     def __init__(self):
    ...         self._data = {1: "a", 2: "b", 3: "c"}
    ...
    ...     def keys(self):
    ...         return self._data.keys()
    ...
    ...     def __getitem__(self, key):
    ...         return self._data[key]
    >>>
    >>> pc.Dict.from_(MyMapping()).inner()
    {1: 'a', 2: 'b', 3: 'c'}
    >>> pc.Dict.from_([("d", "e"), ("f", "g")]).inner()
    {'d': 'e', 'f': 'g'}

    ```
    """
    return Dict(dict(data))

from_object staticmethod

from_object(obj: object) -> Dict[str, Any]

Create a Dict from an object __dict__ attribute.

We can't know in advance the values types, so we use Any.

Parameters:

Name Type Description Default
obj object

The object whose __dict__ attribute will be used to create the Dict.

required

Returns:

Type Description
Dict[str, Any]

Dict[str, Any]: A new Dict instance containing the attributes of the object.

>>> import pyochain as pc
>>> class Person:
...     def __init__(self, name: str, age: int):
...         self.name = name
...         self.age = age
>>> person = Person("Alice", 30)
>>> pc.Dict.from_object(person).inner()
{'name': 'Alice', 'age': 30}
Source code in src/pyochain/_dict/_main.py
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
@staticmethod
def from_object(obj: object) -> Dict[str, Any]:
    """Create a `Dict` from an object `__dict__` attribute.

    We can't know in advance the values types, so we use `Any`.

    Args:
        obj (object): The object whose `__dict__` attribute will be used to create the Dict.

    Returns:
        Dict[str, Any]: A new Dict instance containing the attributes of the object.

    ```python
    >>> import pyochain as pc
    >>> class Person:
    ...     def __init__(self, name: str, age: int):
    ...         self.name = name
    ...         self.age = age
    >>> person = Person("Alice", 30)
    >>> pc.Dict.from_object(person).inner()
    {'name': 'Alice', 'age': 30}

    ```
    """
    return Dict(obj.__dict__)

get_in

get_in(*keys: K) -> Option[V]

Retrieve a value from a nested dictionary structure.

Parameters:

Name Type Description Default
*keys K

keys representing the nested path to retrieve the value.

()

Returns:

Type Description
Option[V]

Option[V]: Value at the nested path or default if not found.

>>> import pyochain as pc
>>> data = {"a": {"b": {"c": 1}}}
>>> pc.Dict(data).get_in("a", "b", "c")
Some(1)
>>> pc.Dict(data).get_in("a", "x").unwrap_or('Not Found')
'Not Found'
Source code in src/pyochain/_dict/_nested.py
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
def get_in(self, *keys: K) -> Option[V]:
    """Retrieve a value from a nested dictionary structure.

    Args:
        *keys (K): keys representing the nested path to retrieve the value.

    Returns:
        Option[V]: Value at the nested path or default if not found.

    ```python
    >>> import pyochain as pc
    >>> data = {"a": {"b": {"c": 1}}}
    >>> pc.Dict(data).get_in("a", "b", "c")
    Some(1)
    >>> pc.Dict(data).get_in("a", "x").unwrap_or('Not Found')
    'Not Found'

    ```
    """

    def _get_in(data: Mapping[K, V]) -> Option[V]:
        return Option.from_(cz.dicttoolz.get_in(keys, data, None))

    return self.into(lambda d: _get_in(d.inner()))

group_by_key

group_by_key(func: Callable[[K], G]) -> Dict[G, dict[K, V]]

Group dict items into sub-dictionaries based on a function of the key.

Parameters:

Name Type Description Default
func Callable[[K], G]

Function to determine the group for each key.

required

Returns:

Type Description
Dict[G, dict[K, V]]

Dict[G, dict[K, V]]: Grouped Dict with groups as keys and sub-dicts as values.

>>> import pyochain as pc
>>> d = {"user_1": 10, "user_2": 20, "admin_1": 100}
>>> pc.Dict(d).group_by_key(lambda k: k.split("_")[0]).inner()
{'user': {'user_1': 10, 'user_2': 20}, 'admin': {'admin_1': 100}}
Source code in src/pyochain/_dict/_groups.py
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
def group_by_key[G](self, func: Callable[[K], G]) -> Dict[G, dict[K, V]]:
    """Group dict items into sub-dictionaries based on a function of the key.

    Args:
        func(Callable[[K], G]): Function to determine the group for each key.

    Returns:
        Dict[G, dict[K, V]]: Grouped Dict with groups as keys and sub-dicts as values.

    ```python
    >>> import pyochain as pc
    >>> d = {"user_1": 10, "user_2": 20, "admin_1": 100}
    >>> pc.Dict(d).group_by_key(lambda k: k.split("_")[0]).inner()
    {'user': {'user_1': 10, 'user_2': 20}, 'admin': {'admin_1': 100}}

    ```
    """

    def _group_by_key(data: dict[K, V]) -> dict[G, dict[K, V]]:
        def _(kv: tuple[K, V]) -> G:
            return func(kv[0])

        return cz.dicttoolz.valmap(dict, cz.itertoolz.groupby(_, data.items()))

    return self._new(_group_by_key)

group_by_key_agg

group_by_key_agg(
    key_func: Callable[[K], G], agg_func: Callable[[Dict[K, V]], R]
) -> Dict[G, R]

Group by key function, then apply aggregation function to each sub-dict.

Parameters:

Name Type Description Default
key_func Callable[[K], G]

Function to determine the group for each key.

required
agg_func Callable[[Dict[K, V]], R]

Function to aggregate each sub-dictionary.

required

Returns:

Type Description
Dict[G, R]

Dict[G, R]: Grouped and aggregated Dict.

This avoids materializing intermediate dict objects if you only need an aggregated result for each group.

>>> import pyochain as pc
>>>
>>> data = {"user_1": 10, "user_2": 20, "admin_1": 100}
>>> pc.Dict(data).group_by_key_agg(
...     key_func=lambda k: k.split("_")[0],
...     agg_func=lambda d: d.iter_values().sum(),
... ).inner()
{'user': 30, 'admin': 100}
>>>
>>> data_files = {
...     "file_a.txt": 100,
...     "file_b.log": 20,
...     "file_c.txt": 50,
...     "file_d.log": 5,
... }
>>>
>>> def get_stats(sub_dict: pc.Dict[str, int]) -> dict[str, Any]:
...     return {
...         "count": sub_dict.iter_keys().length(),
...         "total_size": sub_dict.iter_values().sum(),
...         "max_size": sub_dict.iter_values().max(),
...         "files": sub_dict.iter_keys().sort().into(list),
...     }
>>>
>>> pc.Dict(data_files).group_by_key_agg(
...     key_func=lambda k: k.split(".")[-1], agg_func=get_stats
... ).sort().inner()
{'log': {'count': 2, 'total_size': 25, 'max_size': 20, 'files': ['file_b.log', 'file_d.log']}, 'txt': {'count': 2, 'total_size': 150, 'max_size': 100, 'files': ['file_a.txt', 'file_c.txt']}}

Source code in src/pyochain/_dict/_groups.py
 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
def group_by_key_agg[G, R](
    self,
    key_func: Callable[[K], G],
    agg_func: Callable[[Dict[K, V]], R],
) -> Dict[G, R]:
    """Group by key function, then apply aggregation function to each sub-dict.

    Args:
        key_func(Callable[[K], G]): Function to determine the group for each key.
        agg_func(Callable[[Dict[K, V]], R]): Function to aggregate each sub-dictionary.

    Returns:
        Dict[G, R]: Grouped and aggregated Dict.

    This avoids materializing intermediate `dict` objects if you only need
    an aggregated result for each group.
    ```python
    >>> import pyochain as pc
    >>>
    >>> data = {"user_1": 10, "user_2": 20, "admin_1": 100}
    >>> pc.Dict(data).group_by_key_agg(
    ...     key_func=lambda k: k.split("_")[0],
    ...     agg_func=lambda d: d.iter_values().sum(),
    ... ).inner()
    {'user': 30, 'admin': 100}
    >>>
    >>> data_files = {
    ...     "file_a.txt": 100,
    ...     "file_b.log": 20,
    ...     "file_c.txt": 50,
    ...     "file_d.log": 5,
    ... }
    >>>
    >>> def get_stats(sub_dict: pc.Dict[str, int]) -> dict[str, Any]:
    ...     return {
    ...         "count": sub_dict.iter_keys().length(),
    ...         "total_size": sub_dict.iter_values().sum(),
    ...         "max_size": sub_dict.iter_values().max(),
    ...         "files": sub_dict.iter_keys().sort().into(list),
    ...     }
    >>>
    >>> pc.Dict(data_files).group_by_key_agg(
    ...     key_func=lambda k: k.split(".")[-1], agg_func=get_stats
    ... ).sort().inner()
    {'log': {'count': 2, 'total_size': 25, 'max_size': 20, 'files': ['file_b.log', 'file_d.log']}, 'txt': {'count': 2, 'total_size': 150, 'max_size': 100, 'files': ['file_a.txt', 'file_c.txt']}}

    ```
    """
    from ._main import Dict

    def _group_by_key_agg(data: dict[K, V]) -> dict[G, R]:
        def _key_func(kv: tuple[K, V]) -> G:
            return key_func(kv[0])

        def _agg_func(items: list[tuple[K, V]]) -> R:
            return agg_func(Dict(dict(items)))

        groups = cz.itertoolz.groupby(_key_func, data.items())
        return cz.dicttoolz.valmap(_agg_func, groups)

    return self._new(_group_by_key_agg)

group_by_value

group_by_value(func: Callable[[V], G]) -> Dict[G, dict[K, V]]

Group dict items into sub-dictionaries based on a function of the value.

Parameters:

Name Type Description Default
func Callable[[V], G]

Function to determine the group for each value.

required

Returns:

Type Description
Dict[G, dict[K, V]]

Dict[G, dict[K, V]]: Grouped Dict with groups as keys and sub-dicts as values.

>>> import pyochain as pc
>>> d = {"a": 1, "b": 2, "c": 3, "d": 2}
>>> pc.Dict(d).group_by_value(lambda v: v % 2).inner()
{1: {'a': 1, 'c': 3}, 0: {'b': 2, 'd': 2}}
Source code in src/pyochain/_dict/_groups.py
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
def group_by_value[G](self, func: Callable[[V], G]) -> Dict[G, dict[K, V]]:
    """Group dict items into sub-dictionaries based on a function of the value.

    Args:
        func(Callable[[V], G]): Function to determine the group for each value.

    Returns:
        Dict[G, dict[K, V]]: Grouped Dict with groups as keys and sub-dicts as values.

    ```python
    >>> import pyochain as pc
    >>> d = {"a": 1, "b": 2, "c": 3, "d": 2}
    >>> pc.Dict(d).group_by_value(lambda v: v % 2).inner()
    {1: {'a': 1, 'c': 3}, 0: {'b': 2, 'd': 2}}

    ```
    """

    def _group_by_value(data: dict[K, V]) -> dict[G, dict[K, V]]:
        def _(kv: tuple[K, V]) -> G:
            return func(kv[1])

        return cz.dicttoolz.valmap(dict, cz.itertoolz.groupby(_, data.items()))

    return self._new(_group_by_value)

group_by_value_agg

group_by_value_agg(
    value_func: Callable[[V], G], agg_func: Callable[[Dict[K, V]], R]
) -> Dict[G, R]

Group by value function, then apply aggregation function to each sub-dict.

Parameters:

Name Type Description Default
value_func Callable[[V], G]

Function to determine the group for each value.

required
agg_func Callable[[Dict[K, V]], R]

Function to aggregate each sub-dictionary.

required

Returns:

Type Description
Dict[G, R]

Dict[G, R]: Grouped and aggregated Dict.

This avoids materializing intermediate dict objects if you only need an aggregated result for each group.

>>> import pyochain as pc
>>>
>>> data = {"math": "A", "physics": "B", "english": "A"}
>>> pc.Dict(data).group_by_value_agg(
...     value_func=lambda grade: grade,
...     agg_func=lambda d: d.iter_keys().length(),
... ).inner()
{'A': 2, 'B': 1}
>>> # Second example
>>> sales_data = {
...     "store_1": "Electronics",
...     "store_2": "Groceries",
...     "store_3": "Electronics",
...     "store_4": "Clothing",
... }
>>>
>>> # Obtain the first store for each category (after sorting store names)
>>> pc.Dict(sales_data).group_by_value_agg(
...     value_func=lambda category: category,
...     agg_func=lambda d: d.iter_keys().sort().first(),
... ).sort().inner()
{'Clothing': 'store_4', 'Electronics': 'store_1', 'Groceries': 'store_2'}

Source code in src/pyochain/_dict/_groups.py
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
def group_by_value_agg[G, R](
    self,
    value_func: Callable[[V], G],
    agg_func: Callable[[Dict[K, V]], R],
) -> Dict[G, R]:
    """Group by value function, then apply aggregation function to each sub-dict.

    Args:
        value_func(Callable[[V], G]): Function to determine the group for each value.
        agg_func(Callable[[Dict[K, V]], R]): Function to aggregate each sub-dictionary.

    Returns:
        Dict[G, R]: Grouped and aggregated Dict.

    This avoids materializing intermediate `dict` objects if you only need
    an aggregated result for each group.
    ```python
    >>> import pyochain as pc
    >>>
    >>> data = {"math": "A", "physics": "B", "english": "A"}
    >>> pc.Dict(data).group_by_value_agg(
    ...     value_func=lambda grade: grade,
    ...     agg_func=lambda d: d.iter_keys().length(),
    ... ).inner()
    {'A': 2, 'B': 1}
    >>> # Second example
    >>> sales_data = {
    ...     "store_1": "Electronics",
    ...     "store_2": "Groceries",
    ...     "store_3": "Electronics",
    ...     "store_4": "Clothing",
    ... }
    >>>
    >>> # Obtain the first store for each category (after sorting store names)
    >>> pc.Dict(sales_data).group_by_value_agg(
    ...     value_func=lambda category: category,
    ...     agg_func=lambda d: d.iter_keys().sort().first(),
    ... ).sort().inner()
    {'Clothing': 'store_4', 'Electronics': 'store_1', 'Groceries': 'store_2'}

    ```
    """
    from ._main import Dict

    def _group_by_value_agg(data: dict[K, V]) -> dict[G, R]:
        def _key_func(kv: tuple[K, V]) -> G:
            return value_func(kv[1])

        def _agg_func(items: list[tuple[K, V]]) -> R:
            return agg_func(Dict(dict(items)))

        groups = cz.itertoolz.groupby(_key_func, data.items())
        return cz.dicttoolz.valmap(_agg_func, groups)

    return self._new(_group_by_value_agg)

implode

implode() -> Dict[K, list[V]]

Nest all the values in lists.

Returns:

Type Description
Dict[K, list[V]]

Dict[K, list[V]]: Dict with all values nested in lists.

syntactic sugar for map_values(lambda v: [v])

>>> import pyochain as pc
>>> pc.Dict({1: 2, 3: 4}).implode().inner()
{1: [2], 3: [4]}

Source code in src/pyochain/_dict/_maps.py
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
def implode(self) -> Dict[K, list[V]]:
    """Nest all the values in lists.

    Returns:
        Dict[K, list[V]]: Dict with all values nested in lists.

    syntactic sugar for map_values(lambda v: [v])
    ```python
    >>> import pyochain as pc
    >>> pc.Dict({1: 2, 3: 4}).implode().inner()
    {1: [2], 3: [4]}

    ```
    """

    def _implode(data: dict[K, V]) -> dict[K, list[V]]:
        def _(v: V) -> list[V]:
            return [v]

        return cz.dicttoolz.valmap(_, data)

    return self._new(_implode)

inner

inner() -> T

Get the underlying data.

This is a terminal operation that ends the chain.

Returns:

Name Type Description
T T

The underlying data.

Source code in src/pyochain/_core/_main.py
105
106
107
108
109
110
111
112
113
def inner(self) -> T:
    """Get the underlying data.

    This is a terminal operation that ends the chain.

    Returns:
        T: The underlying data.
    """
    return self._inner

inner_join

inner_join(other: Mapping[K, W]) -> Dict[K, tuple[V, W]]

Performs an inner join with another mapping based on keys.

Parameters:

Name Type Description Default
other Mapping[K, W]

The mapping to join with.

required

Returns:

Type Description
Dict[K, tuple[V, W]]

Dict[K, tuple[V, W]]: Joined Dict with tuples of values from both mappings.

Only keys present in both mappings are kept.

>>> import pyochain as pc
>>> d1 = {"a": 1, "b": 2}
>>> d2 = {"b": 10, "c": 20}
>>> pc.Dict(d1).inner_join(d2).inner()
{'b': (2, 10)}

Source code in src/pyochain/_dict/_joins.py
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
def inner_join[W](self, other: Mapping[K, W]) -> Dict[K, tuple[V, W]]:
    """Performs an inner join with another mapping based on keys.

    Args:
        other(Mapping[K, W]): The mapping to join with.

    Returns:
        Dict[K, tuple[V, W]]: Joined Dict with tuples of values from both mappings.

    Only keys present in both mappings are kept.
    ```python
    >>> import pyochain as pc
    >>> d1 = {"a": 1, "b": 2}
    >>> d2 = {"b": 10, "c": 20}
    >>> pc.Dict(d1).inner_join(d2).inner()
    {'b': (2, 10)}

    ```
    """

    def _inner_join(data: Mapping[K, V]) -> dict[K, tuple[V, W]]:
        return {k: (v, other[k]) for k, v in data.items() if k in other}

    return self._new(_inner_join)

intersect_keys

intersect_keys(*others: Mapping[K, V]) -> Dict[K, V]

Keep only keys present in self and all others mappings.

Parameters:

Name Type Description Default
*others Mapping[K, V]

Other mappings to intersect keys with.

()

Returns:

Type Description
Dict[K, V]

Dict[K, V]: A new Dict with only the intersected keys.

>>> import pyochain as pc
>>> d1 = {"a": 1, "b": 2, "c": 3}
>>> d2 = {"b": 10, "c": 20}
>>> d3 = {"c": 30}
>>> pc.Dict(d1).intersect_keys(d2, d3)
{'c': 3}
Source code in src/pyochain/_dict/_filters.py
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
def intersect_keys(self, *others: Mapping[K, V]) -> Dict[K, V]:
    """Keep only keys present in self and all others mappings.

    Args:
        *others (Mapping[K, V]): Other mappings to intersect keys with.

    Returns:
        Dict[K, V]: A new Dict with only the intersected keys.

    ```python
    >>> import pyochain as pc
    >>> d1 = {"a": 1, "b": 2, "c": 3}
    >>> d2 = {"b": 10, "c": 20}
    >>> d3 = {"c": 30}
    >>> pc.Dict(d1).intersect_keys(d2, d3)
    {'c': 3}

    ```
    """

    def _intersect_keys(data: dict[K, V]) -> dict[K, V]:
        self_keys = set(data.keys())
        for other in others:
            self_keys.intersection_update(other.keys())
        return {k: data[k] for k in self_keys}

    return self._new(_intersect_keys)

into

into(func: Callable[Concatenate[Self, P], R], *args: P.args, **kwargs: P.kwargs) -> R

Convert Self to R.

This method allows to pipe the instance into an object or function that can convert Self into another type.

Conceptually, this allow to do x.into(f) instead of f(x), hence keeping a functional chaining style.

This is a core method, shared by all pyochain wrappers, that allows chaining operations in a functional style.

Parameters:

Name Type Description Default
func Callable[Concatenate[Self, P], R]

Function for conversion.

required
*args P.args

Positional arguments to pass to the function.

()
**kwargs P.kwargs

Keyword arguments to pass to the function.

{}

Returns:

Name Type Description
R R

The converted value.

Example:

>>> import pyochain as pc
>>> def maybe_sum(data: pc.Seq[int]) -> pc.Option[int]:
...     match data.length():
...         case 0:
...             return pc.NONE
...         case _:
...             return pc.Some(data.sum())
>>>
>>> pc.Seq(range(5)).into(maybe_sum).unwrap()
10

Source code in src/pyochain/_core/_main.py
20
21
22
23
24
25
26
27
28
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
def into[**P, R](
    self,
    func: Callable[Concatenate[Self, P], R],
    *args: P.args,
    **kwargs: P.kwargs,
) -> R:
    """Convert `Self` to `R`.

    This method allows to pipe the instance into an object or function that can convert `Self` into another type.

    Conceptually, this allow to do x.into(f) instead of f(x), hence keeping a functional chaining style.

    This is a core method, shared by all pyochain wrappers, that allows chaining operations in a functional style.

    Args:
        func (Callable[Concatenate[Self, P], R]): Function for conversion.
        *args (P.args): Positional arguments to pass to the function.
        **kwargs (P.kwargs): Keyword arguments to pass to the function.

    Returns:
        R: The converted value.

    Example:
    ```python
    >>> import pyochain as pc
    >>> def maybe_sum(data: pc.Seq[int]) -> pc.Option[int]:
    ...     match data.length():
    ...         case 0:
    ...             return pc.NONE
    ...         case _:
    ...             return pc.Some(data.sum())
    >>>
    >>> pc.Seq(range(5)).into(maybe_sum).unwrap()
    10

    ```
    """
    return func(self, *args, **kwargs)

invert

invert() -> Dict[V, list[K]]

Invert the dictionary, grouping keys by common (and hashable) values.

Returns:

Type Description
Dict[V, list[K]]

Dict[V, list[K]]: Inverted Dict with values as keys and lists of original keys as values.

>>> import pyochain as pc
>>> d = {"a": 1, "b": 2, "c": 1}
>>> pc.Dict(d).invert().inner()
{1: ['a', 'c'], 2: ['b']}
Source code in src/pyochain/_dict/_maps.py
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
def invert(self) -> Dict[V, list[K]]:
    """Invert the dictionary, grouping keys by common (and hashable) values.

    Returns:
        Dict[V, list[K]]: Inverted Dict with values as keys and lists of original keys as values.

    ```python
    >>> import pyochain as pc
    >>> d = {"a": 1, "b": 2, "c": 1}
    >>> pc.Dict(d).invert().inner()
    {1: ['a', 'c'], 2: ['b']}

    ```
    """

    def _invert(data: dict[K, V]) -> dict[V, list[K]]:
        inverted: dict[V, list[K]] = defaultdict(list)
        for k, v in data.items():
            inverted[v].append(k)
        return dict(inverted)

    return self._new(_invert)

iter_items

iter_items() -> Iter[tuple[K, V]]

Return an Iter of the dict's items.

Returns:

Type Description
Iter[tuple[K, V]]

Iter[tuple[K, V]]: An Iter wrapping the dictionary's (key, value) pairs.

>>> import pyochain as pc
>>> pc.Dict({"a": 1, "b": 2}).iter_items().collect()
Seq(('a', 1), ('b', 2))
Source code in src/pyochain/_dict/_iter.py
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
def iter_items(self) -> Iter[tuple[K, V]]:
    """Return an Iter of the dict's items.

    Returns:
        Iter[tuple[K, V]]: An Iter wrapping the dictionary's (key, value) pairs.

    ```python
    >>> import pyochain as pc
    >>> pc.Dict({"a": 1, "b": 2}).iter_items().collect()
    Seq(('a', 1), ('b', 2))

    ```
    """
    from .._iter import Iter

    return self.into(lambda d: Iter(d.inner().items()))

iter_keys

iter_keys() -> Iter[K]

Return an Iter of the dict's keys.

Returns:

Type Description
Iter[K]

Iter[K]: An Iter wrapping the dictionary's keys.

>>> import pyochain as pc
>>> pc.Dict({1: 2}).iter_keys().collect()
Seq(1,)
Source code in src/pyochain/_dict/_iter.py
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
def iter_keys(self) -> Iter[K]:
    """Return an Iter of the dict's keys.

    Returns:
        Iter[K]: An Iter wrapping the dictionary's keys.

    ```python
    >>> import pyochain as pc
    >>> pc.Dict({1: 2}).iter_keys().collect()
    Seq(1,)

    ```
    """
    from .._iter import Iter

    return self.into(lambda d: Iter(d._inner.keys()))

iter_values

iter_values() -> Iter[V]

Return an Iter of the dict's values.

Returns:

Type Description
Iter[V]

Iter[V]: An Iter wrapping the dictionary's values.

>>> import pyochain as pc
>>> pc.Dict({1: 2}).iter_values().collect()
Seq(2,)
Source code in src/pyochain/_dict/_iter.py
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
def iter_values(self) -> Iter[V]:
    """Return an Iter of the dict's values.

    Returns:
        Iter[V]: An Iter wrapping the dictionary's values.

    ```python
    >>> import pyochain as pc
    >>> pc.Dict({1: 2}).iter_values().collect()
    Seq(2,)

    ```
    """
    from .._iter import Iter

    return self.into(lambda d: Iter(d.inner().values()))

itr

itr(
    func: Callable[Concatenate[Iter[U], P], R], *args: P.args, **kwargs: P.kwargs
) -> Dict[K, R]

Apply a function to each value after wrapping it in an Iter.

Parameters:

Name Type Description Default
func Callable[Concatenate[Iter[U], P], R]

Function to apply to each value after wrapping it in an Iter.

required
*args P.args

Positional arguments to pass to the function.

()
**kwargs P.kwargs

Keyword arguments to pass to the function.

{}

Returns:

Type Description
Dict[K, R]

Dict[K, R]: Dict with function results as values.

Syntactic sugar for map_values(lambda data: func(Iter(data), *args, **kwargs))

>>> import pyochain as pc
>>> data = {
...     "numbers1": [1, 2, 3],
...     "numbers2": [4, 5, 6],
... }
>>> pc.Dict(data).itr(lambda v: v.repeat(5).flatten().sum())
{'numbers1': 30, 'numbers2': 75}

Source code in src/pyochain/_dict/_iter.py
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
def itr[**P, R, U](
    self: MappingWrapper[K, Iterable[U]],
    func: Callable[Concatenate[Iter[U], P], R],
    *args: P.args,
    **kwargs: P.kwargs,
) -> Dict[K, R]:
    """Apply a function to each value after wrapping it in an Iter.

    Args:
        func(Callable[Concatenate[Iter[U], P], R]): Function to apply to each value after wrapping it in an Iter.
        *args(P.args): Positional arguments to pass to the function.
        **kwargs(P.kwargs): Keyword arguments to pass to the function.

    Returns:
        Dict[K, R]: Dict with function results as values.

    Syntactic sugar for `map_values(lambda data: func(Iter(data), *args, **kwargs))`
    ```python
    >>> import pyochain as pc
    >>> data = {
    ...     "numbers1": [1, 2, 3],
    ...     "numbers2": [4, 5, 6],
    ... }
    >>> pc.Dict(data).itr(lambda v: v.repeat(5).flatten().sum())
    {'numbers1': 30, 'numbers2': 75}

    ```
    """
    from .._iter import Iter

    def _itr(data: Mapping[K, Iterable[U]]) -> dict[K, R]:
        def _(v: Iterable[U]) -> R:
            return func(Iter(v), *args, **kwargs)

        return cz.dicttoolz.valmap(_, data)

    return self._new(_itr)

left_join

left_join(other: Mapping[K, W]) -> Dict[K, tuple[V, W | None]]

Performs a left join with another mapping based on keys.

Parameters:

Name Type Description Default
other Mapping[K, W]

The mapping to join with.

required

Returns:

Type Description
Dict[K, tuple[V, W | None]]

Dict[K, tuple[V, W | None]]: Joined Dict with tuples of values, right side can be None.

All keys from the left dictionary (self) are kept.

>>> import pyochain as pc
>>> d1 = {"a": 1, "b": 2}
>>> d2 = {"b": 10, "c": 20}
>>> pc.Dict(d1).left_join(d2).inner()
{'a': (1, None), 'b': (2, 10)}

Source code in src/pyochain/_dict/_joins.py
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
def left_join[W](self, other: Mapping[K, W]) -> Dict[K, tuple[V, W | None]]:
    """Performs a left join with another mapping based on keys.

    Args:
        other(Mapping[K, W]): The mapping to join with.

    Returns:
        Dict[K, tuple[V, W | None]]: Joined Dict with tuples of values, right side can be None.

    All keys from the left dictionary (self) are kept.
    ```python
    >>> import pyochain as pc
    >>> d1 = {"a": 1, "b": 2}
    >>> d2 = {"b": 10, "c": 20}
    >>> pc.Dict(d1).left_join(d2).inner()
    {'a': (1, None), 'b': (2, 10)}

    ```
    """

    def _left_join(data: Mapping[K, V]) -> dict[K, tuple[V, W | None]]:
        return {k: (v, other.get(k)) for k, v in data.items()}

    return self._new(_left_join)

map_items

map_items(func: Callable[[tuple[K, V]], tuple[KR, VR]]) -> Dict[KR, VR]

Transform (key, value) pairs using a function that takes a (key, value) tuple.

Parameters:

Name Type Description Default
func Callable[[tuple[K, V]], tuple[KR, VR]]

Function to transform each (key, value) pair into a new (key, value) tuple.

required

Returns:

Type Description
Dict[KR, VR]

Dict[KR, VR]: Dict with transformed items.

>>> import pyochain as pc
>>> pc.Dict({"Alice": 10, "Bob": 20}).map_items(
...     lambda kv: (kv[0].upper(), kv[1] * 2)
... ).inner()
{'ALICE': 20, 'BOB': 40}
Source code in src/pyochain/_dict/_maps.py
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
def map_items[KR, VR](
    self,
    func: Callable[[tuple[K, V]], tuple[KR, VR]],
) -> Dict[KR, VR]:
    """Transform (key, value) pairs using a function that takes a (key, value) tuple.

    Args:
        func (Callable[[tuple[K, V]], tuple[KR, VR]]): Function to transform each (key, value) pair into a new (key, value) tuple.

    Returns:
        Dict[KR, VR]: Dict with transformed items.

    ```python
    >>> import pyochain as pc
    >>> pc.Dict({"Alice": 10, "Bob": 20}).map_items(
    ...     lambda kv: (kv[0].upper(), kv[1] * 2)
    ... ).inner()
    {'ALICE': 20, 'BOB': 40}

    ```
    """
    return self._new(partial(cz.dicttoolz.itemmap, func))

map_keys

map_keys(func: Callable[[K], T]) -> Dict[T, V]

Return keys transformed by func.

Parameters:

Name Type Description Default
func Callable[[K], T]

Function to apply to each key in the dictionary.

required

Returns:

Type Description
Dict[T, V]

Dict[T, V]: Dict with transformed keys.

>>> import pyochain as pc
>>> pc.Dict({"Alice": [20, 15, 30], "Bob": [10, 35]}).map_keys(
...     str.lower
... ).inner()
{'alice': [20, 15, 30], 'bob': [10, 35]}
>>>
>>> pc.Dict({1: "a"}).map_keys(str).inner()
{'1': 'a'}
Source code in src/pyochain/_dict/_maps.py
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
def map_keys[T](self, func: Callable[[K], T]) -> Dict[T, V]:
    """Return keys transformed by func.

    Args:
        func (Callable[[K], T]): Function to apply to each key in the dictionary.

    Returns:
        Dict[T, V]: Dict with transformed keys.

    ```python
    >>> import pyochain as pc
    >>> pc.Dict({"Alice": [20, 15, 30], "Bob": [10, 35]}).map_keys(
    ...     str.lower
    ... ).inner()
    {'alice': [20, 15, 30], 'bob': [10, 35]}
    >>>
    >>> pc.Dict({1: "a"}).map_keys(str).inner()
    {'1': 'a'}

    ```
    """
    return self._new(partial(cz.dicttoolz.keymap, func))

map_kv

map_kv(func: Callable[[K, V], tuple[KR, VR]]) -> Dict[KR, VR]

Transform (key, value) pairs using a function that takes key and value as separate arguments.

Parameters:

Name Type Description Default
func Callable[[K, V], tuple[KR, VR]]

Function to transform each key and value into a new (key, value) tuple.

required

Returns:

Type Description
Dict[KR, VR]

Dict[KR, VR]: Dict with transformed items.

>>> import pyochain as pc
>>> pc.Dict({1: 2}).map_kv(lambda k, v: (k + 1, v * 10)).inner()
{2: 20}
Source code in src/pyochain/_dict/_maps.py
 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
def map_kv[KR, VR](
    self,
    func: Callable[[K, V], tuple[KR, VR]],
) -> Dict[KR, VR]:
    """Transform (key, value) pairs using a function that takes key and value as separate arguments.

    Args:
        func (Callable[[K, V], tuple[KR, VR]]): Function to transform each key and value into a new (key, value) tuple.

    Returns:
        Dict[KR, VR]: Dict with transformed items.

    ```python
    >>> import pyochain as pc
    >>> pc.Dict({1: 2}).map_kv(lambda k, v: (k + 1, v * 10)).inner()
    {2: 20}

    ```
    """

    def _map_kv(data: dict[K, V]) -> dict[KR, VR]:
        def _(kv: tuple[K, V]) -> tuple[KR, VR]:
            return func(kv[0], kv[1])

        return cz.dicttoolz.itemmap(_, data)

    return self._new(_map_kv)

map_values

map_values(func: Callable[[V], T]) -> Dict[K, T]

Return values transformed by func.

Parameters:

Name Type Description Default
func Callable[[V], T]

Function to apply to each value in the dictionary.

required

Returns:

Type Description
Dict[K, T]

Dict[K, T]: Dict with transformed values.

>>> import pyochain as pc
>>> pc.Dict({"Alice": [20, 15, 30], "Bob": [10, 35]}).map_values(sum).inner()
{'Alice': 65, 'Bob': 45}
>>>
>>> pc.Dict({1: 1}).map_values(lambda v: v + 1).inner()
{1: 2}
Source code in src/pyochain/_dict/_maps.py
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
def map_values[T](self, func: Callable[[V], T]) -> Dict[K, T]:
    """Return values transformed by func.

    Args:
        func (Callable[[V], T]): Function to apply to each value in the dictionary.

    Returns:
        Dict[K, T]: Dict with transformed values.

    ```python
    >>> import pyochain as pc
    >>> pc.Dict({"Alice": [20, 15, 30], "Bob": [10, 35]}).map_values(sum).inner()
    {'Alice': 65, 'Bob': 45}
    >>>
    >>> pc.Dict({1: 1}).map_values(lambda v: v + 1).inner()
    {1: 2}

    ```
    """
    return self._new(partial(cz.dicttoolz.valmap, func))

merge

merge(*others: Mapping[K, V]) -> Dict[K, V]

Merge other dicts into this one.

Parameters:

Name Type Description Default
*others Mapping[K, V]

One or more mappings to merge into the current dictionary.

()

Returns:

Type Description
Dict[K, V]

Dict[K, V]: Merged Dict with values from all dicts.

>>> import pyochain as pc
>>> pc.Dict({1: "one"}).merge({2: "two"}).inner()
{1: 'one', 2: 'two'}
>>> # Later dictionaries have precedence
>>> pc.Dict({1: 2, 3: 4}).merge({3: 3, 4: 4}).inner()
{1: 2, 3: 3, 4: 4}
Source code in src/pyochain/_dict/_joins.py
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
def merge(self, *others: Mapping[K, V]) -> Dict[K, V]:
    """Merge other dicts into this one.

    Args:
        *others(Mapping[K, V]): One or more mappings to merge into the current dictionary.

    Returns:
        Dict[K, V]: Merged Dict with values from all dicts.

    ```python
    >>> import pyochain as pc
    >>> pc.Dict({1: "one"}).merge({2: "two"}).inner()
    {1: 'one', 2: 'two'}
    >>> # Later dictionaries have precedence
    >>> pc.Dict({1: 2, 3: 4}).merge({3: 3, 4: 4}).inner()
    {1: 2, 3: 3, 4: 4}

    ```
    """

    def _merge(data: Mapping[K, V]) -> dict[K, V]:
        return cz.dicttoolz.merge(data, *others)

    return self._new(_merge)

merge_with

merge_with(*others: Mapping[K, V], func: Callable[[Iterable[V]], V]) -> Dict[K, V]

Merge dicts using a function to combine values for duplicate keys.

Parameters:

Name Type Description Default
*others Mapping[K, V]

One or more mappings to merge into the current dictionary.

()
func Callable[[Iterable[V]], V]

Function to combine values for duplicate keys.

required

Returns:

Type Description
Dict[K, V]

Dict[K, V]: Merged Dict with combined values.

A key may occur in more than one dict, and all values mapped from the key will be passed to the function as a list, such as func([val1, val2, ...]).

>>> import pyochain as pc
>>> pc.Dict({1: 1, 2: 2}).merge_with({1: 10, 2: 20}, func=sum).inner()
{1: 11, 2: 22}
>>> pc.Dict({1: 1, 2: 2}).merge_with({2: 20, 3: 30}, func=max).inner()
{1: 1, 2: 20, 3: 30}

Source code in src/pyochain/_dict/_joins.py
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
def merge_with(
    self,
    *others: Mapping[K, V],
    func: Callable[[Iterable[V]], V],
) -> Dict[K, V]:
    """Merge dicts using a function to combine values for duplicate keys.

    Args:
        *others(Mapping[K, V]): One or more mappings to merge into the current dictionary.
        func(Callable[[Iterable[V]], V]): Function to combine values for duplicate keys.

    Returns:
        Dict[K, V]: Merged Dict with combined values.

    A key may occur in more than one dict, and all values mapped from the key will be passed to the function as a list, such as func([val1, val2, ...]).
    ```python
    >>> import pyochain as pc
    >>> pc.Dict({1: 1, 2: 2}).merge_with({1: 10, 2: 20}, func=sum).inner()
    {1: 11, 2: 22}
    >>> pc.Dict({1: 1, 2: 2}).merge_with({2: 20, 3: 30}, func=max).inner()
    {1: 1, 2: 20, 3: 30}

    ```
    """

    def _merge_with(data: Mapping[K, V]) -> dict[K, V]:
        return cz.dicttoolz.merge_with(func, data, *others)

    return self._new(_merge_with)

pivot

pivot(*indices: int) -> Dict[Any, Any]

Pivot a nested dictionary by rearranging the key levels according to order.

Syntactic sugar for Dict.to_arrays().rearrange(*indices).to_records()

Parameters:

Name Type Description Default
*indices int

Indices specifying the new order of key levels

()

Returns:

Type Description
Dict[Any, Any]

Dict[Any, Any]: Pivoted dictionary with keys rearranged

Example: ```python

import pyochain as pc d = {"A": {"X": 1, "Y": 2}, "B": {"X": 3, "Y": 4}} pc.Dict(d).pivot(1, 0).inner() {'X': {'A': 1, 'B': 3}, 'Y': {'A': 2, 'B': 4}}

Source code in src/pyochain/_dict/_main.py
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
def pivot(self, *indices: int) -> Dict[Any, Any]:
    """Pivot a nested dictionary by rearranging the key levels according to order.

    Syntactic sugar for `Dict.to_arrays().rearrange(*indices).to_records()`

    Args:
        *indices (int): Indices specifying the new order of key levels

    Returns:
        Dict[Any, Any]: Pivoted dictionary with keys rearranged

    Example:
    ```python
    >>> import pyochain as pc
    >>> d = {"A": {"X": 1, "Y": 2}, "B": {"X": 3, "Y": 4}}
    >>> pc.Dict(d).pivot(1, 0).inner()
    {'X': {'A': 1, 'B': 3}, 'Y': {'A': 2, 'B': 4}}
    """
    return self.to_arrays().rearrange(*indices).to_records()

pluck

pluck(*keys: str) -> Dict[U, Any]

Extract values from nested dictionaries using a sequence of keys.

Parameters:

Name Type Description Default
*keys str

keys to extract values from the nested dictionaries.

()

Returns:

Type Description
Dict[U, Any]

Dict[U, Any]: Dict with extracted values from nested dictionaries.

>>> import pyochain as pc
>>> data = {
...     "person1": {"name": "Alice", "age": 30},
...     "person2": {"name": "Bob", "age": 25},
... }
>>> pc.Dict(data).pluck("name").inner()
{'person1': 'Alice', 'person2': 'Bob'}
Source code in src/pyochain/_dict/_nested.py
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
def pluck[U: str | int](self: NestedDict[U, Any], *keys: str) -> Dict[U, Any]:
    """Extract values from nested dictionaries using a sequence of keys.

    Args:
        *keys (str): keys to extract values from the nested dictionaries.

    Returns:
        Dict[U, Any]: Dict with extracted values from nested dictionaries.
    ```python
    >>> import pyochain as pc
    >>> data = {
    ...     "person1": {"name": "Alice", "age": 30},
    ...     "person2": {"name": "Bob", "age": 25},
    ... }
    >>> pc.Dict(data).pluck("name").inner()
    {'person1': 'Alice', 'person2': 'Bob'}

    ```
    """
    getter = partial(cz.dicttoolz.get_in, keys)

    def _pluck(data: Mapping[U, Any]) -> dict[U, Any]:
        return cz.dicttoolz.valmap(getter, data)

    return self._new(_pluck)

rename

rename(mapping: Mapping[K, K]) -> Dict[K, V]

Return a new Dict with keys renamed according to the mapping.

Parameters:

Name Type Description Default
mapping Mapping[K, K]

A dictionary mapping old keys to new keys.

required

Returns:

Type Description
Dict[K, V]

Dict[K, V]: Dict with keys renamed according to the mapping.

Keys not in the mapping are kept as is.

>>> import pyochain as pc
>>> d = {"a": 1, "b": 2, "c": 3}
>>> mapping = {"b": "beta", "c": "gamma"}
>>> pc.Dict(d).rename(mapping).inner()
{'a': 1, 'beta': 2, 'gamma': 3}

Source code in src/pyochain/_dict/_process.py
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
def rename(self, mapping: Mapping[K, K]) -> Dict[K, V]:
    """Return a new Dict with keys renamed according to the mapping.

    Args:
        mapping (Mapping[K, K]): A dictionary mapping old keys to new keys.

    Returns:
        Dict[K, V]: Dict with keys renamed according to the mapping.

    Keys not in the mapping are kept as is.
    ```python
    >>> import pyochain as pc
    >>> d = {"a": 1, "b": 2, "c": 3}
    >>> mapping = {"b": "beta", "c": "gamma"}
    >>> pc.Dict(d).rename(mapping).inner()
    {'a': 1, 'beta': 2, 'gamma': 3}

    ```
    """

    def _rename(data: dict[K, V]) -> dict[K, V]:
        return {mapping.get(k, k): v for k, v in data.items()}

    return self._new(_rename)

sort

sort(*, reverse: bool = False) -> Dict[K, V]

Sort the dictionary by its keys and return a new Dict.

Parameters:

Name Type Description Default
reverse bool

Whether to sort in descending order. Defaults to False.

False

Returns:

Type Description
Dict[K, V]

Dict[K, V]: Sorted Dict by keys.

>>> import pyochain as pc
>>> pc.Dict({"b": 2, "a": 1}).sort().inner()
{'a': 1, 'b': 2}
Source code in src/pyochain/_dict/_process.py
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
def sort(self, *, reverse: bool = False) -> Dict[K, V]:
    """Sort the dictionary by its keys and return a new Dict.

    Args:
        reverse (bool): Whether to sort in descending order. Defaults to False.

    Returns:
        Dict[K, V]: Sorted Dict by keys.

    ```python
    >>> import pyochain as pc
    >>> pc.Dict({"b": 2, "a": 1}).sort().inner()
    {'a': 1, 'b': 2}

    ```
    """

    def _sort(data: dict[K, V]) -> dict[K, V]:
        return dict(sorted(data.items(), reverse=reverse))

    return self._new(_sort)

sort_values

sort_values(*, reverse: bool = False) -> Dict[K, U]

Sort the dictionary by its values and return a new Dict.

Parameters:

Name Type Description Default
reverse bool

Whether to sort in descending order. Defaults to False.

False

Returns:

Type Description
Dict[K, U]

Dict[K, U]: Sorted Dict by values.

>>> import pyochain as pc
>>> pc.Dict({"a": 2, "b": 1}).sort_values().inner()
{'b': 1, 'a': 2}
Source code in src/pyochain/_dict/_process.py
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
def sort_values[U: SupportsRichComparison[Any]](
    self: ProcessDict[K, U],
    *,
    reverse: bool = False,
) -> Dict[K, U]:
    """Sort the dictionary by its values and return a new Dict.

    Args:
        reverse (bool): Whether to sort in descending order. Defaults to False.

    Returns:
        Dict[K, U]: Sorted Dict by values.
    ```python
    >>> import pyochain as pc
    >>> pc.Dict({"a": 2, "b": 1}).sort_values().inner()
    {'b': 1, 'a': 2}

    ```
    """

    def _sort_values(data: dict[K, U]) -> dict[K, U]:
        return dict(sorted(data.items(), key=lambda item: item[1], reverse=reverse))

    return self._new(_sort_values)

struct

struct(
    func: Callable[Concatenate[Dict[K, U], P], R], *args: P.args, **kwargs: P.kwargs
) -> Dict[K, R]

Apply a function to each value after wrapping it in a Dict.

Parameters:

Name Type Description Default
func Callable[Concatenate[Dict[K, U], P], R]

Function to apply to each value after wrapping it in a Dict.

required
*args P.args

Positional arguments to pass to the function.

()
**kwargs P.kwargs

Keyword arguments to pass to the function.

{}

Returns:

Type Description
Dict[K, R]

Dict[K, R]: Dict with function results as values.

Syntactic sugar for map_values(lambda data: func(pc.Dict(data), *args, **kwargs))

>>> import pyochain as pc
>>> data = {
...     "person1": {"name": "Alice", "age": 30, "city": "New York"},
...     "person2": {"name": "Bob", "age": 25, "city": "Los Angeles"},
... }
>>> pc.Dict(data).struct(lambda d: d.map_keys(str.upper).drop("AGE").inner())
... # doctest: +NORMALIZE_WHITESPACE
{'person1': {'CITY': 'New York', 'NAME': 'Alice'},
'person2': {'CITY': 'Los Angeles', 'NAME': 'Bob'}}

Source code in src/pyochain/_dict/_nested.py
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
def struct[**P, R, U: dict[Any, Any]](
    self: NestedDict[K, U],
    func: Callable[Concatenate[Dict[K, U], P], R],
    *args: P.args,
    **kwargs: P.kwargs,
) -> Dict[K, R]:
    """Apply a function to each value after wrapping it in a Dict.

    Args:
        func (Callable[Concatenate[Dict[K, U], P], R]): Function to apply to each value after wrapping it in a Dict.
        *args (P.args): Positional arguments to pass to the function.
        **kwargs (P.kwargs): Keyword arguments to pass to the function.

    Returns:
        Dict[K, R]: Dict with function results as values.

    Syntactic sugar for `map_values(lambda data: func(pc.Dict(data), *args, **kwargs))`
    ```python
    >>> import pyochain as pc
    >>> data = {
    ...     "person1": {"name": "Alice", "age": 30, "city": "New York"},
    ...     "person2": {"name": "Bob", "age": 25, "city": "Los Angeles"},
    ... }
    >>> pc.Dict(data).struct(lambda d: d.map_keys(str.upper).drop("AGE").inner())
    ... # doctest: +NORMALIZE_WHITESPACE
    {'person1': {'CITY': 'New York', 'NAME': 'Alice'},
    'person2': {'CITY': 'Los Angeles', 'NAME': 'Bob'}}

    ```
    """
    from ._main import Dict

    def _struct(data: Mapping[K, U]) -> dict[K, R]:
        def _(v: dict[Any, Any]) -> R:
            return func(Dict(v), *args, **kwargs)

        return cz.dicttoolz.valmap(_, data)

    return self._new(_struct)

tap

tap(
    func: Callable[Concatenate[Self, P], Any], *args: P.args, **kwargs: P.kwargs
) -> Self

Tap into the chain to perform side effects without altering the data.

Parameters:

Name Type Description Default
func Callable[Concatenate[Self, P], Any]

Function to apply to the instance for side effects.

required
*args P.args

Positional arguments to pass to the function.

()
**kwargs P.kwargs

Keyword arguments to pass to the function.

{}

Returns:

Name Type Description
Self Self

The instance itself for chaining.

Example:

>>> import pyochain as pc
>>> pc.Seq([1, 2, 3, 4]).tap(print).last()
Seq(1, 2, 3, 4)
4

Source code in src/pyochain/_core/_main.py
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
def tap[**P](
    self,
    func: Callable[Concatenate[Self, P], Any],
    *args: P.args,
    **kwargs: P.kwargs,
) -> Self:
    """Tap into the chain to perform side effects without altering the data.

    Args:
        func (Callable[Concatenate[Self, P], Any]): Function to apply to the instance for side effects.
        *args (P.args): Positional arguments to pass to the function.
        **kwargs (P.kwargs): Keyword arguments to pass to the function.

    Returns:
        Self: The instance itself for chaining.

    Example:
    ```python
    >>> import pyochain as pc
    >>> pc.Seq([1, 2, 3, 4]).tap(print).last()
    Seq(1, 2, 3, 4)
    4

    ```
    """
    func(self, *args, **kwargs)
    return self

to_arrays

to_arrays() -> Vec[list[Any]]

Convert the nested dictionary into a Vec of arrays.

The sequence represents all paths from root to leaves.

Returns:

Type Description
Vec[list[Any]]

Vec[list[Any]]: A Vec of arrays representing paths from root to leaves.

>>> import pyochain as pc
>>> data = {
...     "a": {"b": 1, "c": 2},
...     "d": {"e": {"f": 3}},
... }
>>> pc.Dict(data).to_arrays().inner()
[['a', 'b', 1], ['a', 'c', 2], ['d', 'e', 'f', 3]]
Source code in src/pyochain/_dict/_iter.py
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
132
133
134
135
136
137
def to_arrays(self) -> Vec[list[Any]]:
    """Convert the nested dictionary into a `Vec` of arrays.

    The sequence represents all paths from root to leaves.

    Returns:
        Vec[list[Any]]: A `Vec` of arrays representing paths from root to leaves.

    ```python
    >>> import pyochain as pc
    >>> data = {
    ...     "a": {"b": 1, "c": 2},
    ...     "d": {"e": {"f": 3}},
    ... }
    >>> pc.Dict(data).to_arrays().inner()
    [['a', 'b', 1], ['a', 'c', 2], ['d', 'e', 'f', 3]]

    ```
    """
    from .._iter import Vec

    def _to_arrays(d: Mapping[Any, Any]) -> list[list[Any]]:
        match d:
            case Mapping():
                arr: list[Any] = []
                for k, v in d.items():
                    arr.extend([[k, *el] for el in _to_arrays(v)])
                return arr

            case _:
                return [[d]]

    return self.into(lambda d: Vec(_to_arrays(d.inner())))

unpivot

unpivot() -> Dict[str, dict[str, Any]]

Unpivot a nested dictionary by swapping rows and columns.

Returns:

Type Description
Dict[str, dict[str, Any]]

Dict[str, dict[str, Any]]: Unpivoted Dict with columns as keys and rows as sub-dicts.

Example: ```python

import pyochain as pc data = { ... "row1": {"col1": "A", "col2": "B"}, ... "row2": {"col1": "C", "col2": "D"}, ... } pc.Dict(data).unpivot() ... # doctest: +NORMALIZE_WHITESPACE {'col1': {'row1': 'A', 'row2': 'C'}, 'col2': {'row1': 'B', 'row2': 'D'}}

Source code in src/pyochain/_dict/_nested.py
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
def unpivot(
    self: NestedDict[str, Mapping[str, Any]],
) -> Dict[str, dict[str, Any]]:
    """Unpivot a nested dictionary by swapping rows and columns.

    Returns:
        Dict[str, dict[str, Any]]: Unpivoted Dict with columns as keys and rows as sub-dicts.

    Example:
    ```python
    >>> import pyochain as pc
    >>> data = {
    ...     "row1": {"col1": "A", "col2": "B"},
    ...     "row2": {"col1": "C", "col2": "D"},
    ... }
    >>> pc.Dict(data).unpivot()
    ... # doctest: +NORMALIZE_WHITESPACE
    {'col1': {'row1': 'A', 'row2': 'C'}, 'col2': {'row1': 'B', 'row2': 'D'}}
    """

    def _unpivot(
        data: Mapping[str, Mapping[str, Any]],
    ) -> dict[str, dict[str, Any]]:
        out: dict[str, dict[str, Any]] = {}
        for rkey, inner in data.items():
            for ckey, val in inner.items():
                out.setdefault(ckey, {})[rkey] = val
        return out

    return self._new(_unpivot)

update_in

update_in(*keys: K, func: Callable[[V], V], default: V | None = None) -> Dict[K, V]

Update value in a (potentially) nested dictionary.

Parameters:

Name Type Description Default
*keys K

keys representing the nested path to update.

()
func Callable[[V], V]

Function to apply to the value at the specified path.

required
default V | None

Default value to use if the path does not exist, by default None

None

Returns:

Type Description
Dict[K, V]

Dict[K, V]: Dict with the updated value at the nested path.

Applies the func to the value at the path specified by keys, returning a new Dict with the updated value.

If the path does not exist, it will be created with the default value (if provided) before applying func.

>>> import pyochain as pc
>>> inc = lambda x: x + 1
>>> pc.Dict({"a": 0}).update_in("a", func=inc).inner()
{'a': 1}
>>> transaction = {
...     "name": "Alice",
...     "purchase": {"items": ["Apple", "Orange"], "costs": [0.50, 1.25]},
...     "credit card": "5555-1234-1234-1234",
... }
>>> pc.Dict(transaction).update_in("purchase", "costs", func=sum).inner()
{'name': 'Alice', 'purchase': {'items': ['Apple', 'Orange'], 'costs': 1.75}, 'credit card': '5555-1234-1234-1234'}
>>> # updating a value when k0 is not in d
>>> pc.Dict({}).update_in(1, 2, 3, func=str, default="bar").inner()
{1: {2: {3: 'bar'}}}
>>> pc.Dict({1: "foo"}).update_in(2, 3, 4, func=inc, default=0).inner()
{1: 'foo', 2: {3: {4: 1}}}

Source code in src/pyochain/_dict/_process.py
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
def update_in(
    self,
    *keys: K,
    func: Callable[[V], V],
    default: V | None = None,
) -> Dict[K, V]:
    """Update value in a (potentially) nested dictionary.

    Args:
        *keys (K): keys representing the nested path to update.
        func (Callable[[V], V]): Function to apply to the value at the specified path.
        default (V | None): Default value to use if the path does not exist, by default None

    Returns:
        Dict[K, V]: Dict with the updated value at the nested path.

    Applies the func to the value at the path specified by keys, returning a new Dict with the updated value.

    If the path does not exist, it will be created with the default value (if provided) before applying func.
    ```python
    >>> import pyochain as pc
    >>> inc = lambda x: x + 1
    >>> pc.Dict({"a": 0}).update_in("a", func=inc).inner()
    {'a': 1}
    >>> transaction = {
    ...     "name": "Alice",
    ...     "purchase": {"items": ["Apple", "Orange"], "costs": [0.50, 1.25]},
    ...     "credit card": "5555-1234-1234-1234",
    ... }
    >>> pc.Dict(transaction).update_in("purchase", "costs", func=sum).inner()
    {'name': 'Alice', 'purchase': {'items': ['Apple', 'Orange'], 'costs': 1.75}, 'credit card': '5555-1234-1234-1234'}
    >>> # updating a value when k0 is not in d
    >>> pc.Dict({}).update_in(1, 2, 3, func=str, default="bar").inner()
    {1: {2: {3: 'bar'}}}
    >>> pc.Dict({1: "foo"}).update_in(2, 3, 4, func=inc, default=0).inner()
    {1: 'foo', 2: {3: {4: 1}}}

    ```
    """

    def _update_in(data: dict[K, V]) -> dict[K, V]:
        return cz.dicttoolz.update_in(data, keys, func, default=default)

    return self._new(_update_in)

with_key

with_key(key: K, value: V) -> Dict[K, V]

Return a new Dict with key set to value.

Parameters:

Name Type Description Default
key K

Key to set in the dictionary.

required
value V

Value to associate with the specified key.

required

Returns:

Type Description
Dict[K, V]

Dict[K, V]: New Dict with the key-value pair set.

Does not modify the initial dictionary.

>>> import pyochain as pc
>>> pc.Dict({"x": 1}).with_key("x", 2).inner()
{'x': 2}
>>> pc.Dict({"x": 1}).with_key("y", 3).inner()
{'x': 1, 'y': 3}
>>> pc.Dict({}).with_key("x", 1).inner()
{'x': 1}

Source code in src/pyochain/_dict/_process.py
 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
def with_key(self, key: K, value: V) -> Dict[K, V]:
    """Return a new Dict with key set to value.

    Args:
        key (K): Key to set in the dictionary.
        value (V): Value to associate with the specified key.

    Returns:
        Dict[K, V]: New Dict with the key-value pair set.

    Does not modify the initial dictionary.
    ```python
    >>> import pyochain as pc
    >>> pc.Dict({"x": 1}).with_key("x", 2).inner()
    {'x': 2}
    >>> pc.Dict({"x": 1}).with_key("y", 3).inner()
    {'x': 1, 'y': 3}
    >>> pc.Dict({}).with_key("x", 1).inner()
    {'x': 1}

    ```
    """

    def _with_key(data: dict[K, V]) -> dict[K, V]:
        return cz.dicttoolz.assoc(data, key, value)

    return self._new(_with_key)

with_nested_key

with_nested_key(*keys: K, value: V) -> Dict[K, V]

Set a nested key path and return a new Dict with new, potentially nested, key value pair.

Parameters:

Name Type Description Default
*keys K

keys representing the nested path.

()
value V

Value to set at the specified nested path.

required

Returns:

Type Description
Dict[K, V]

Dict[K, V]: Dict with the new nested key-value pair.

>>> import pyochain as pc
>>> purchase = {
...     "name": "Alice",
...     "order": {"items": ["Apple", "Orange"], "costs": [0.50, 1.25]},
...     "credit card": "5555-1234-1234-1234",
... }
>>> pc.Dict(purchase).with_nested_key(
...     "order", "costs", value=[0.25, 1.00]
... ).inner()
{'name': 'Alice', 'order': {'items': ['Apple', 'Orange'], 'costs': [0.25, 1.0]}, 'credit card': '5555-1234-1234-1234'}
Source code in src/pyochain/_dict/_nested.py
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
def with_nested_key(self, *keys: K, value: V) -> Dict[K, V]:
    """Set a nested key path and return a new Dict with new, potentially nested, key value pair.

    Args:
        *keys (K): keys representing the nested path.
        value (V): Value to set at the specified nested path.

    Returns:
        Dict[K, V]: Dict with the new nested key-value pair.
    ```python
    >>> import pyochain as pc
    >>> purchase = {
    ...     "name": "Alice",
    ...     "order": {"items": ["Apple", "Orange"], "costs": [0.50, 1.25]},
    ...     "credit card": "5555-1234-1234-1234",
    ... }
    >>> pc.Dict(purchase).with_nested_key(
    ...     "order", "costs", value=[0.25, 1.00]
    ... ).inner()
    {'name': 'Alice', 'order': {'items': ['Apple', 'Orange'], 'costs': [0.25, 1.0]}, 'credit card': '5555-1234-1234-1234'}

    ```
    """

    def _with_nested_key(data: dict[K, V]) -> dict[K, V]:
        return cz.dicttoolz.assoc_in(data, keys, value=value)

    return self._new(_with_nested_key)