Skip to content

Recursive

coola.recursive

Recursive data transformation using DFS pattern.

This design is inspired by the DFS array iterator pattern and provides: 1. Memory-efficient generator-based traversal 2. Clean separation between transformation logic and type dispatch 3. No state object threading through recursion 4. Easy extensibility via registry pattern

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.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
>>> registry = TransformerRegistry()
>>> 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
>>> transformer = SequenceTransformer()
>>> transformer
SequenceTransformer()
>>> registry = TransformerRegistry({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
>>> registry = TransformerRegistry()
>>> 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.

The registry includes an LRU cache for type lookups to optimize performance in applications that repeatedly transform similar data structures.

Parameters:

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

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

None

Attributes:

Name Type Description
_registry dict[type, BaseTransformer[Any]]

Internal mapping of registered types to transformers

_default_transformer BaseTransformer[Any]

Fallback transformer for unregistered types

_find_transformer_cached BaseTransformer[Any]

Cached version of transformer lookup

Example

Basic usage with a sequence transformer:

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

Registering custom transformers:

>>> from coola.recursive import TransformerRegistry, SequenceTransformer
>>> registry = TransformerRegistry()
>>> 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 you register a transformer for Sequence but not for list, lists will use the Sequence transformer.

Results are cached using an LRU cache (256 entries) for performance, as transformer lookup is a hot path in recursive transformations.

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, a parent

BaseTransformer[Any]

type's transformer via MRO, or the default transformer

Example
>>> from collections.abc import Sequence
>>> from coola.recursive import TransformerRegistry, SequenceTransformer
>>> registry = TransformerRegistry()
>>> 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,

bool

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.

This method associates a transformer instance with a specific Python type. When data of this type is transformed, the registered transformer will be used. The 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(
  (<class 'list'>): SequenceTransformer()
  (<class 'dict'>): MappingTransformer()
)

coola.recursive.TransformerRegistry.transform

transform(data: Any, 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 3. Delegates to that transformer's transform method 4. The transformer recursively processes 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 Any

The data to transform (can be nested: lists, dicts, tuples, etc.)

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

Transformed data with the same structure as the input but with

Any

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 TransformerRegistry instance with transformers registered for: - Scalar types (int, float, complex, bool, str) - Sequences (list, tuple, Sequence ABC) - Sets (set, frozenset) - Mappings (dict, Mapping ABC)

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: Any,
    func: Callable[[Any], Any],
    registry: TransformerRegistry | None = None,
) -> Any

Recursively apply a function to all items in nested data.

This is the main public interface that maintains compatibility with the original implementation.

Parameters:

Name Type Description Default
data Any

Input data (can be nested)

required
func Callable[[Any], Any]

Function to apply to each leaf value

required
registry TransformerRegistry | None

Registry to resolve transformers for nested data.

None

Returns:

Type Description
Any

Transformed data with same structure as input

Example
>>> from coola.recursive import recursive_apply
>>> recursive_apply({"a": 1, "b": "abc"}, str)
{'a': '1', 'b': 'abc'}
>>> recursive_apply([1, [2, 3], {"x": 4}], lambda x: x * 2)
[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 allows users to add support for custom types without modifying global state directly.

Parameters:

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

Dictionary mapping types to transformer instances

required
exist_ok bool

If False, raises error if any type already registered

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()})