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