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 | |
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 | |
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 | |
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 | |
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 |
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 | |
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 | |
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 | |
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 | |
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 | |
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 | |
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 | |
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 | |
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 | |
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 |
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 | |
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 | |
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 | |
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 | |
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 | |
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 | |
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 | |
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 | |
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 | |
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 | |
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 | |
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 | |
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 | |
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 | |
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 | |
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 | |
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 | |
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 | |
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 | |
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 | |
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 | |
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 | |
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 | |
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 | |
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 | |
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 | |
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 | |
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 | |
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 | |
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 | |
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 |
>>> 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 | |
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 | |
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 | |
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 | |
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 | |