Skip to content

Recursive

coola.recursive

Utilities to recursively transform nested data structures.

Example
>>> from coola.recursive import recursive_apply
>>> recursive_apply({"a": 1, "b": 2}, lambda x: x * 10)
{'a': 10, 'b': 20}

coola.recursive.BaseTransformer

Bases: ABC, Generic[T]

Base class for type-specific transformers.

This abstract base class defines the interface for transformers that recursively apply functions to nested data structures. Each concrete transformer implementation handles a specific data type and knows how to reconstruct that type after transforming its nested elements.

Notes

Subclasses must implement the transform method to define how their specific type should be traversed and reconstructed.

Example
>>> from coola.recursive import DefaultTransformer, TransformerRegistry
>>> registry = TransformerRegistry()
>>> transformer = DefaultTransformer()
>>> transformer
DefaultTransformer()
>>> transformer.transform([1, 2, 3], func=str, registry=registry)
'[1, 2, 3]'

coola.recursive.BaseTransformer.transform abstractmethod

transform(
    data: T,
    func: Callable[[Any], Any],
    registry: TransformerRegistry,
) -> Any

Transform data by recursively applying a function.

This method traverses the data structure, applies the given function to leaf values, and reconstructs the original structure with the transformed values. The registry is used to resolve appropriate transformers for nested data types encountered during traversal.

Parameters:

Name Type Description Default
data T

The data structure to transform. Must be of type T that this transformer handles.

required
func Callable[[Any], Any]

A function to apply to leaf values (non-container elements). Should accept a single argument and return the transformed value.

required
registry TransformerRegistry

The transformer registry used to look up transformers for nested data structures of different types.

required

Returns:

Type Description
Any

The transformed data structure, maintaining the original type

Any

and structure but with leaf values transformed by func.

Example
>>> from coola.recursive import DefaultTransformer, SequenceTransformer, TransformerRegistry
>>> registry = TransformerRegistry()
>>> transformer = DefaultTransformer()
>>> # Convert numeric values to strings
>>> transformer.transform([1, 2, 3], func=str, registry=registry)
'[1, 2, 3]'
>>> # Apply a mathematical operation
>>> transformer.transform([1, 2, 3], func=lambda x: x * 2, registry=registry)
[1, 2, 3, 1, 2, 3]

coola.recursive.ConditionalTransformer

Bases: BaseTransformer[T]

Wrapper transformer that conditionally applies transformations.

This transformer wraps another transformer and only applies it when a given condition evaluates to True. If the condition is False, the data is returned unchanged. This allows for selective transformation based on runtime checks without modifying the underlying transformer or core architecture.

Parameters:

Name Type Description Default
transformer BaseTransformer[T]

The underlying transformer to apply when the condition is met. This can be any BaseTransformer implementation.

required
condition Callable[[Any], bool]

A predicate function that determines whether to apply the transformation. Should accept the data as input and return True to transform or False to pass through unchanged.

required
Example
>>> from coola.recursive import (
...     DefaultTransformer,
...     ConditionalTransformer,
...     TransformerRegistry,
... )
>>> registry = TransformerRegistry()
>>> # Create a transformer that only processes positive numbers
>>> transformer = ConditionalTransformer(
...     transformer=DefaultTransformer(),
...     condition=lambda x: isinstance(x, (int, float)) and x > 0,
... )
>>> transformer
ConditionalTransformer(
  (transformer): DefaultTransformer()
  (condition): <function <lambda> at 0x...>
)
>>> # Positive number: condition passes, transformation applied
>>> transformer.transform(5, func=lambda x: x * 2, registry=registry)
10
>>> # Negative number: condition fails, returned unchanged
>>> transformer.transform(-5, func=lambda x: x * 2, registry=registry)
-5
>>> # Non-numeric: condition fails, returned unchanged
>>> transformer.transform("text", func=lambda x: x * 2, registry=registry)
'text'

coola.recursive.DefaultTransformer

Bases: BaseTransformer[Any]

Transformer for leaf nodes that directly applies the function.

This is the default transformer used for values that don't require recursive traversal. It simply applies the given function directly to the data without any structural transformation or nested processing. This transformer is typically used as the terminal case in recursive transformations when a leaf value (non-container) is encountered.

Notes

Unlike container-specific transformers (e.g., list, dict), this transformer does not traverse nested structures. It treats all input as atomic values and applies the function directly.

Example
>>> from coola.recursive import DefaultTransformer, TransformerRegistry
>>> registry = TransformerRegistry()
>>> transformer = DefaultTransformer()
>>> transformer
DefaultTransformer()
>>> # Transform a simple value directly
>>> transformer.transform(42, func=str, registry=registry)
'42'
>>> # Transform a string
>>> transformer.transform("hello", func=str.upper, registry=registry)
'HELLO'
>>> # Apply a mathematical operation
>>> transformer.transform(10, func=lambda x: x * 2, registry=registry)
20
>>> # Even container types are treated as atomic values
>>> transformer.transform([1, 2, 3], func=str, registry=registry)
'[1, 2, 3]'

coola.recursive.IdentityTransformer

Bases: BaseTransformer[Any]

Transformer for leaf nodes that passes the value through unchanged, ignoring func.

This transformer never applies the transformation function. It is useful to register for types that must be preserved as-is while traversing a nested data structure (e.g. to skip certain leaf types instead of transforming them).

Example
>>> from coola.recursive import IdentityTransformer, TransformerRegistry
>>> registry = TransformerRegistry()
>>> transformer = IdentityTransformer()
>>> transformer.transform(42, func=str, registry=registry)
42

coola.recursive.IdentityTransformer.transform

transform(
    data: Any,
    func: Callable[[Any], bool],
    registry: TransformerRegistry,
) -> Any

Return the data unchanged, ignoring func.

Parameters:

Name Type Description Default
data Any

The data to return unchanged.

required
func Callable[[Any], bool]

Unused. Accepted only to satisfy the BaseTransformer interface.

required
registry TransformerRegistry

Unused. Accepted only to satisfy the BaseTransformer interface.

required

Returns:

Type Description
Any

The input data, unchanged.

coola.recursive.KeyFilterTransformer

Bases: BaseTransformer[Mapping[Any, Any]]

Transformer for mappings that drops entries whose key matches a predicate and recurses into the remaining values.

func plays a dual role: it is called directly on each key to decide whether the entry should be dropped, and it is also passed down as the transformation function applied recursively (via the registry) to the values of the entries that are kept. After filtering and transforming, the mapping is reconstructed using its original type.

Notes
  • Entries whose key satisfies func (i.e. func(key) is truthy) are removed from the output.
  • Keys of the entries that are kept are preserved unchanged.
  • If the value of a kept entry is itself a nested mapping that is dispatched to a KeyFilterTransformer, keys matching func are removed at every level of nesting.
Example
>>> from coola.recursive import (
...     KeyFilterTransformer,
...     TransformerRegistry,
...     DefaultTransformer,
... )
>>> registry = TransformerRegistry({object: DefaultTransformer()})
>>> transformer = KeyFilterTransformer()
>>> transformer.transform(
...     {"keep": 1, "secret": 2}, func=lambda x: x == "secret", registry=registry
... )
{'keep': False}

coola.recursive.KeyFilterTransformer.transform

transform(
    data: Mapping[Any, Any],
    func: Callable[[Any], bool],
    registry: TransformerRegistry,
) -> Mapping[Any, Any]

Drop entries whose key matches func and transform the remaining values recursively.

Parameters:

Name Type Description Default
data Mapping[Any, Any]

The mapping to filter and transform.

required
func Callable[[Any], bool]

A predicate applied to each key: entries whose key satisfies it are dropped. It is also passed as the transformation function used recursively (via registry) on the values of the entries that are kept.

required
registry TransformerRegistry

The transformer registry used to recursively transform the values of the kept entries.

required

Returns:

Type Description
Mapping[Any, Any]

A new mapping of the same type as data, containing

Mapping[Any, Any]

only the entries whose key did not match func, with

Mapping[Any, Any]

their values transformed recursively.

coola.recursive.MappingTransformer

Bases: BaseTransformer[Mapping[Any, Any]]

Transformer for mapping types that recursively transforms values.

This transformer handles dict-like mapping structures by recursively transforming all values while preserving keys unchanged. After transformation, it reconstructs the mapping using its original type (dict, OrderedDict, defaultdict, etc.), maintaining the mapping's specific characteristics and behavior.

Notes
  • Keys are never transformed, only values are processed recursively
  • The original mapping type is preserved in the output
  • Nested mappings and other containers in values are handled recursively
  • Empty mappings are preserved as empty mappings of the same type
Example
>>> from coola.recursive import MappingTransformer, TransformerRegistry, DefaultTransformer
>>> registry = TransformerRegistry({object: DefaultTransformer()})
>>> transformer = MappingTransformer()
>>> transformer
MappingTransformer()
>>> # Transform simple dict values
>>> transformer.transform({"a": 1, "b": 2}, func=str, registry=registry)
{'a': '1', 'b': '2'}
>>> # Keys remain unchanged
>>> transformer.transform({1: "x", 2: "y"}, func=str.upper, registry=registry)
{1: 'X', 2: 'Y'}
>>> # Nested structures in values are handled recursively
>>> transformer.transform({"nums": [1, 2, 3], "text": "hello"}, func=str, registry=registry)
{'nums': '[1, 2, 3]', 'text': 'hello'}
>>> # Empty mappings are preserved
>>> transformer.transform({}, func=str, registry=registry)
{}

coola.recursive.SequenceTransformer

Bases: BaseTransformer[Sequence[Any]]

Transformer for sequence types that recursively transforms elements.

This transformer handles sequence structures (list, tuple, namedtuple, etc.) by recursively transforming all elements while preserving the original sequence type and structure. After transformation, it reconstructs the sequence using its original type, with special handling for named tuples to preserve their field structure.

Notes
  • All elements are transformed recursively through the registry
  • The original sequence type is preserved (list remains list, tuple remains tuple)
  • Named tuples receive special handling to preserve field names
  • Nested sequences and other containers are handled recursively
  • Empty sequences are preserved as empty sequences of the same type
  • String sequences (str) should typically use a different transformer as they are often treated as atomic values
Example
>>> from coola.recursive import SequenceTransformer, TransformerRegistry, DefaultTransformer
>>> transformer = SequenceTransformer()
>>> transformer
SequenceTransformer()
>>> registry = TransformerRegistry({object: DefaultTransformer(), list: transformer})
>>> # Transform list elements
>>> transformer.transform([1, 2, 3], func=str, registry=registry)
['1', '2', '3']
>>> # Tuple type is preserved
>>> transformer.transform((1, 2, 3), func=lambda x: x * 2, registry=registry)
(2, 4, 6)
>>> # Nested sequences are handled recursively
>>> transformer.transform([[1, 2], [3, 4]], func=str, registry=registry)
[['1', '2'], ['3', '4']]
>>> # Empty sequences are preserved
>>> transformer.transform([], func=str, registry=registry)
[]
>>> # Named tuples preserve their structure
>>> from collections import namedtuple
>>> Point = namedtuple("Point", ["x", "y"])
>>> transformer.transform(Point(1, 2), func=str, registry=registry)
Point(x='1', y='2')

coola.recursive.SetTransformer

Bases: BaseTransformer[Set[Any]]

Transformer for set types that recursively transforms elements.

This transformer handles set structures (set, frozenset) by recursively transforming all elements while preserving the original set type. After transformation, it reconstructs the set using its original type. Sets maintain their unordered, unique-element properties.

Important

HASHABILITY REQUIREMENT: All transformed values MUST remain hashable (i.e., immutable and hashable) since sets can only contain hashable elements. Transforming to unhashable types (like lists or dicts) will raise a TypeError.

Notes
  • All elements are transformed recursively through the registry
  • The original set type is preserved (set remains set, frozenset remains frozenset)
  • Element order is not guaranteed (sets are unordered)
  • Duplicate transformed values will be automatically deduplicated
  • Empty sets are preserved as empty sets of the same type
  • If transformation produces unhashable values, a TypeError will be raised
Example
>>> from coola.recursive import SetTransformer, TransformerRegistry, DefaultTransformer
>>> registry = TransformerRegistry({object: DefaultTransformer()})
>>> transformer = SetTransformer()
>>> transformer
SetTransformer()
>>> # Transform set elements (order may vary in output)
>>> transformer.transform({1}, func=str, registry=registry)
{'1'}
>>> # Frozenset type is preserved
>>> transformer.transform(frozenset([1, 2, 3]), func=lambda x: x * 2, registry=registry)
frozenset({2, 4, 6})
>>> # Duplicate values after transformation are automatically deduplicated
>>> transformer.transform({1, 2, 3}, func=lambda x: x // 4, registry=registry)
{0}

coola.recursive.TransformerRegistry

Registry that manages and dispatches transformers based on data type.

This registry maintains a mapping from Python types to transformer instances and uses the Method Resolution Order (MRO) for type lookup. When transforming data, it automatically selects the most specific registered transformer for the data's type, falling back to parent types or a default transformer if needed.

Type lookups are cached internally for performance. The cache is cleared automatically whenever the registry is modified (e.g. via register or register_many).

Parameters:

Name Type Description Default
initial_state dict[type, BaseTransformer[Any]] | None

Optional initial mapping of types to transformers. If provided, the state is copied to prevent external mutations.

None
Example

Basic usage with a sequence transformer:

>>> from coola.recursive import TransformerRegistry, SequenceTransformer, DefaultTransformer
>>> registry = TransformerRegistry(
...     {object: DefaultTransformer(), list: SequenceTransformer()}
... )
>>> registry
TransformerRegistry(
  (state): TypeRegistry(
      (<class 'object'>): DefaultTransformer()
      (<class 'list'>): SequenceTransformer()
    )
)
>>> registry.transform([1, 2, 3], str)
['1', '2', '3']

Registering custom transformers:

>>> from coola.recursive import TransformerRegistry, SequenceTransformer
>>> registry = TransformerRegistry({object: DefaultTransformer()})
>>> registry.register(list, SequenceTransformer())
>>> registry.transform([1, 2, 3], lambda x: x * 2)
[2, 4, 6]

Working with nested structures:

>>> from coola.recursive import get_default_registry
>>> registry = get_default_registry()
>>> data = {"a": [1, 2], "b": [3, 4]}
>>> registry.transform(data, str)
{'a': ['1', '2'], 'b': ['3', '4']}

coola.recursive.TransformerRegistry.find_transformer

find_transformer(data_type: type) -> BaseTransformer[Any]

Find the appropriate transformer for a given type.

Uses the Method Resolution Order (MRO) to find the most specific registered transformer. For example, if a transformer is registered for Sequence but not for list, lists will use the Sequence transformer.

Parameters:

Name Type Description Default
data_type type

The Python type to find a transformer for.

required

Returns:

Type Description
BaseTransformer[Any]

The most specific registered transformer for this type, resolved

BaseTransformer[Any]

via MRO, or the default transformer if no match is found.

Example
>>> from collections.abc import Sequence
>>> from coola.recursive import TransformerRegistry, SequenceTransformer, DefaultTransformer
>>> registry = TransformerRegistry({object: DefaultTransformer()})
>>> registry.register(Sequence, SequenceTransformer())
>>> # list does not inherit from Sequence, so it uses DefaultTransformer
>>> transformer = registry.find_transformer(list)
>>> transformer
DefaultTransformer()

coola.recursive.TransformerRegistry.has_transformer

has_transformer(data_type: type) -> bool

Check if a transformer is explicitly registered for the given type.

Note that this only checks for direct registration. Even if this returns False, find_transformer may still return a transformer via MRO lookup or the default transformer.

Parameters:

Name Type Description Default
data_type type

The type to check.

required

Returns:

Type Description
bool

True if a transformer is explicitly registered for this type, False otherwise.

Example
>>> from coola.recursive import TransformerRegistry, SequenceTransformer
>>> registry = TransformerRegistry()
>>> registry.register(list, SequenceTransformer())
>>> registry.has_transformer(list)
True
>>> registry.has_transformer(tuple)
False

coola.recursive.TransformerRegistry.register

register(
    data_type: type,
    transformer: BaseTransformer[Any],
    exist_ok: bool = False,
) -> None

Register a transformer for a given data type.

The internal type-lookup cache is automatically cleared after registration to ensure consistency.

Parameters:

Name Type Description Default
data_type type

The Python type to register (e.g., list, dict, custom classes).

required
transformer BaseTransformer[Any]

The transformer instance that handles this type.

required
exist_ok bool

If False (default), raises an error if the type is already registered. If True, overwrites the existing registration silently.

False

Raises:

Type Description
RuntimeError

If the type is already registered and exist_ok is False.

Example
>>> from coola.recursive import TransformerRegistry, SequenceTransformer
>>> registry = TransformerRegistry()
>>> registry.register(list, SequenceTransformer())
>>> registry.has_transformer(list)
True

coola.recursive.TransformerRegistry.register_many

register_many(
    mapping: Mapping[type, BaseTransformer[Any]],
    exist_ok: bool = False,
) -> None

Register multiple transformers at once.

This is a convenience method for bulk registration that internally calls register for each type-transformer pair.

Parameters:

Name Type Description Default
mapping Mapping[type, BaseTransformer[Any]]

Dictionary mapping Python types to transformer instances.

required
exist_ok bool

If False (default), raises an error if any type is already registered. If True, overwrites existing registrations silently.

False

Raises:

Type Description
RuntimeError

If any type is already registered and exist_ok is False.

Example
>>> from coola.recursive import TransformerRegistry, SequenceTransformer, MappingTransformer
>>> registry = TransformerRegistry()
>>> registry.register_many(
...     {
...         list: SequenceTransformer(),
...         dict: MappingTransformer(),
...     }
... )
>>> registry
TransformerRegistry(
  (state): TypeRegistry(
      (<class 'list'>): SequenceTransformer()
      (<class 'dict'>): MappingTransformer()
    )
)

coola.recursive.TransformerRegistry.transform

transform(data: object, func: Callable[[Any], Any]) -> Any

Transform data by applying a function recursively through the structure.

This is the main entry point for transformation. It automatically:

  1. Determines the data's type.
  2. Finds the appropriate transformer via find_transformer.
  3. Delegates to that transformer's transform method, which recursively processes any nested structures.

The original structure of the data is preserved - only the leaf values are transformed by the provided function.

Parameters:

Name Type Description Default
data object

The data to transform. Can be a nested structure such as a list, dict, or tuple.

required
func Callable[[Any], Any]

Function to apply to leaf values. Should accept one argument and return a transformed value.

required

Returns:

Type Description
Any

The transformed data, with the same structure as the input but

Any

with leaf values transformed by func.

Example

Converting all numbers to strings in a nested structure:

>>> from coola.recursive import get_default_registry
>>> registry = get_default_registry()
>>> registry.transform({"scores": [95, 87, 92], "name": "test"}, str)
{'scores': ['95', '87', '92'], 'name': 'test'}

Doubling all numeric values:

>>> from coola.recursive import get_default_registry
>>> registry = get_default_registry()
>>> registry.transform(
...     [1, [2, 3], {"a": 4}], lambda x: x * 2 if isinstance(x, (int, float)) else x
... )
[2, [4, 6], {'a': 8}]

coola.recursive.get_default_registry

get_default_registry() -> TransformerRegistry

Get or create the default global registry with common Python types.

Returns a singleton registry instance that is pre-configured with transformers for Python's built-in types including sequences (list, tuple), mappings (dict), sets, and scalar types (int, float, str, bool).

This function uses a singleton pattern to ensure the same registry instance is returned on subsequent calls, which is efficient and maintains consistency across an application.

Returns:

Type Description
TransformerRegistry

A singleton TransformerRegistry configured for common built-in

TransformerRegistry

scalar and container types.

Notes

The singleton pattern means modifications to the returned registry affect all future calls to this function. If you need an isolated registry, create a new TransformerRegistry instance directly.

Example
>>> from coola.recursive import get_default_registry
>>> registry = get_default_registry()
>>> # Registry is ready to use with common Python types
>>> registry.transform([1, 2, 3], str)
['1', '2', '3']
>>> registry.transform({"a": 1, "b": 2}, lambda x: x * 10)
{'a': 10, 'b': 20}

coola.recursive.recursive_apply

recursive_apply(
    data: object,
    func: Callable[[Any], Any],
    registry: TransformerRegistry | None = None,
) -> Any

Recursively transform nested data with a type-aware registry.

Parameters:

Name Type Description Default
data object

Input data, potentially containing nested containers.

required
func Callable[[Any], Any]

Function applied to leaf values.

required
registry TransformerRegistry | None

Registry to resolve transformers for nested data. If None, the default registry is used.

None

Returns:

Type Description
Any

Transformed data with the same structure as the input.

Notes

Mapping keys are preserved. By default, values of str are treated as leaves and are not iterated character by character.

Example
>>> from coola.recursive import recursive_apply
>>> recursive_apply({"a": 1, "b": "abc"}, str)
{'a': '1', 'b': 'abc'}
>>> # Guard the operation so non-numeric leaves are preserved.
>>> recursive_apply([1, [2, 3], {"x": 4}], lambda x: x * 2 if isinstance(x, int) else x)
[2, [4, 6], {'x': 8}]

coola.recursive.register_transformers

register_transformers(
    mapping: Mapping[type, BaseTransformer[Any]],
    exist_ok: bool = False,
) -> None

Register custom transformers to the default global registry.

This function extends the singleton registry returned by get_default_registry.

Parameters:

Name Type Description Default
mapping Mapping[type, BaseTransformer[Any]]

Mapping of Python types to transformer instances.

required
exist_ok bool

If False (default), raise an error when a type is already registered. If True, overwrites existing registrations silently.

False

Raises:

Type Description
RuntimeError

If any type is already registered and exist_ok is False.

Example
>>> from coola.recursive import register_transformers, BaseTransformer
>>> class MyType:
...     def __init__(self, value):
...         self.value = value
...
>>> class MyTransformer(BaseTransformer):
...     def transform(self, data, func, registry):
...         return MyType(func(data.value))
...
>>> register_transformers({MyType: MyTransformer()})