Iterator

coola.iterator ¶

Contain code to iterate over nested data.

coola.iterator.bfs_iterate ¶

bfs_iterate(
    data: Any, registry: ChildFinderRegistry | None = None
) -> Iterator[Any]

Perform Depth-First Search (DFS) iteration over nested data structures (lists, dicts, tuples, sets, etc.).

This function yields elements from the data structure in a DFS manner, recursively traversing all levels of nested structures. It uses the appropriate iterators registered for the data types (e.g., lists, dictionaries, etc.).

Parameters:

Name	Type	Description	Default
`data`	`Any`	The nested data structure to traverse. Can be a list, tuple, dict, set, or any other registered type.	required
`registry`	`ChildFinderRegistry \| None`	The registry to resolve iterators for nested data. If `None`, the default registry is used.	`None`

Yields:

Type	Description
`Any`	Atomic leaf values in BFS order (excludes containers even if empty)

Example

>>> from coola.iterator import bfs_iterate
>>> list(bfs_iterate({"a": 1, "b": "abc"}))
[1, 'abc']
>>> list(bfs_iterate([1, [2, 3], {"x": 4}]))
[1, 2, 3, 4]

coola.iterator.dfs_iterate ¶

dfs_iterate(
    data: Any, registry: IteratorRegistry | None = None
) -> Iterator[Any]

Perform Depth-First Search (DFS) iteration over nested data structures (lists, dicts, tuples, sets, etc.).

This function yields elements from the data structure in a DFS manner, recursively traversing all levels of nested structures. It uses the appropriate iterators registered for the data types (e.g., lists, dictionaries, etc.).

Parameters:

Name	Type	Description	Default
`data`	`Any`	The nested data structure to traverse. Can be a list, tuple, dict, set, or any other registered type.	required
`registry`	`IteratorRegistry \| None`	The registry to resolve iterators for nested data. If `None`, the default registry is used.	`None`

Yields:

Type	Description
`Any`	The elements from the nested data structure in DFS order.

Example

>>> from coola.iterator import dfs_iterate
>>> list(dfs_iterate({"a": 1, "b": "abc"}))
[1, 'abc']
>>> list(dfs_iterate([1, [2, 3], {"x": 4}]))
[1, 2, 3, 4]

coola.iterator.filter_by_type ¶

filter_by_type(
    iterator: Iterable[Any],
    types: type[T] | tuple[type, ...],
) -> Iterator[T]

Filter an iterator to yield only values of specified types.

This function acts as a type-safe filter that passes through only values matching the specified type(s). It's particularly useful when working with heterogeneous collections where you need to extract elements of specific types.

Parameters:

Name	Type	Description	Default
`iterator`	`Iterable[Any]`	An iterator or iterable to filter. Can contain values of any type.	required
`types`	`type[T] \| tuple[type, ...]`	A single type or tuple of types to filter for. Only values that are instances of these types will be yielded. Follows the same semantics as the built-in isinstance() function.	required

Yields:

Type	Description
`T`	Values from the input iterator that are instances of any of the specified
`T`	types, preserving the original order of elements.

Example

Filter mixed-type list to get only integers:

>>> from coola.iterator import filter_by_type
>>> list(filter_by_type([1, "hello", 2, 3.14, "world", 4], int))
[1, 2, 4]

Filter for multiple types:

>>> from coola.iterator import filter_by_type
>>> # Note: bool is a subclass of int
>>> list(filter_by_type([1, "hello", 2.5, True, None, [1, 2]], (int, float)))
[1, 2.5, True]

Notes

This function uses isinstance() internally, so subclass relationships are respected (e.g., bool values will match int type).
The input iterator is consumed as items are yielded.
For empty iterators or when no items match, the generator yields nothing.

coola.iterator.bfs ¶

Contain code to iterate over nested data with a Breath-First Search (BFS) strategy.

coola.iterator.bfs.BaseChildFinder ¶

Bases: ABC, Generic[T]

Abstract base class for child finders used in breadth-first search.

This class defines the interface that all child finders must implement. Child finders are responsible for finding and yielding the immediate children of a given data structure. Custom child finders can be registered with a ChildFinderRegistry to handle specific data types during BFS traversal.

The generic type parameter T indicates the type of data this child finder is designed to handle.

Notes

Subclasses must implement the find_children method.
For leaf types (types with no children), simply return without yielding.

Example

>>> from coola.iterator.bfs import DefaultChildFinder
>>> child_finder = DefaultChildFinder()
>>> list(child_finder.find_children(42))
[]
>>> list(child_finder.find_children("hello"))
[]

coola.iterator.bfs.BaseChildFinder.find_children `abstractmethod` ¶

find_children(data: T) -> Iterator[Any]

Find and yield the immediate children of the given data structure.

This method defines how to extract children from the data structure. For container types, this typically means yielding the contained elements. For leaf types, this method should return without yielding anything.

Parameters:

Name	Type	Description	Default
`data`	`T`	The data structure whose children should be found.	required

Yields:

Type	Description
`Any`	The immediate children of the data structure. The type of yielded
`Any`	elements depends on the specific data structure being processed.

Notes

This method should only yield direct children, not recurse deeply.
The BFS traversal logic handles visiting children recursively.

Example

>>> from coola.iterator.bfs import DefaultChildFinder
>>> child_finder = DefaultChildFinder()
>>> list(child_finder.find_children(42))
[]
>>> list(child_finder.find_children("hello"))
[]

coola.iterator.bfs.ChildFinderRegistry ¶

Registry that manages child finders for breadth-first traversal of nested data structures.

This registry maps Python data types to BaseChildFinder instances. During traversal, the registry selects the most specific child finder for a given object using Method Resolution Order (MRO). If no match is found, a default child finder is used.

The registry also caches resolved child finders to speed up repeated lookups.

Parameters:

Name	Type	Description	Default
`registry`	`dict[type, BaseChildFinder[Any]] \| None`	An optional dictionary mapping Python types to `BaseChildFinder` instances. If provided, the registry is initialized with this mapping.	`None`

Attributes:

Name	Type	Description
`_registry`	`dict[type, BaseChildFinder[Any]]`	Mapping of registered data types to child finders.
`_default_child_finder`	`BaseChildFinder[Any]`	Fallback child finder used when no match is found in the registry.
`_child_finder_cache`	`dict[type, BaseChildFinder[Any]]`	Cache mapping data types to resolved child finders (after MRO lookup).

Example

Basic usage with a flat iterable:

>>> from coola.iterator.bfs import ChildFinderRegistry, IterableChildFinder
>>> registry = ChildFinderRegistry({list: IterableChildFinder()})
>>> list(registry.iterate([1, 2, 3]))
[1, 2, 3]

Working with nested structures using the default registry:

>>> from coola.iterator.bfs import get_default_registry
>>> registry = get_default_registry()
>>> data = {"a": [1, 2], "b": [3, 4]}
>>> list(registry.iterate(data))
[1, 2, 3, 4]

Breadth-first traversal over mixed nested data:

>>> from coola.iterator.bfs import (
...     ChildFinderRegistry,
...     IterableChildFinder,
...     MappingChildFinder,
... )
>>> registry = ChildFinderRegistry(
...     {list: IterableChildFinder(), dict: MappingChildFinder()}
... )
>>> data = {"a": [1, 2], "b": [3, 4], "c": 5, "d": {"e": 6}}
>>> list(registry.iterate(data))
[5, 1, 2, 3, 4, 6]

coola.iterator.bfs.ChildFinderRegistry.find_child_finder ¶

find_child_finder(data_type: type) -> BaseChildFinder[Any]

Find the appropriate child finder for a given data type.

This method resolves the child finder using MRO lookup and caches the result for faster subsequent access.

Parameters:

Name	Type	Description	Default
`data_type`	`type`	The data type for which to find a child finder.	required

Returns:

Type	Description
`BaseChildFinder[Any]`	The resolved child finder instance.

Example

>>> from coola.iterator.bfs import ChildFinderRegistry, IterableChildFinder
>>> registry = ChildFinderRegistry({list: IterableChildFinder()})
>>> registry.find_child_finder(list)
IterableChildFinder()
>>> registry.find_child_finder(tuple)
DefaultChildFinder()

coola.iterator.bfs.ChildFinderRegistry.find_children ¶

find_children(data: Any) -> Iterator[Any]

Return the immediate children of an object using its child finder.

This method does not perform traversal by itself. It delegates to the appropriate child finder for the object's type.

Parameters:

Name	Type	Description	Default
`data`	`Any`	The object whose children should be extracted.	required

Yields:

Type	Description
`Any`	Child objects as defined by the resolved child finder.

Example

>>> from coola.iterator.bfs import ChildFinderRegistry, IterableChildFinder
>>> registry = ChildFinderRegistry({list: IterableChildFinder()})
>>> list(registry.find_children([1, 2, 3]))
[1, 2, 3]

coola.iterator.bfs.ChildFinderRegistry.has_child_finder ¶

has_child_finder(data_type: type) -> bool

Check if a child finder is directly registered for a data type.

This method only checks for an exact type match in the registry. Even if this returns False, a suitable child finder may still be resolved via MRO lookup.

Parameters:

Name	Type	Description	Default
`data_type`	`type`	The type to check.	required

Returns:

Type	Description
`bool`	`True` if a child finder is directly registered for the type, `False` otherwise.

Example

>>> from coola.iterator.bfs import ChildFinderRegistry, IterableChildFinder
>>> registry = ChildFinderRegistry({list: IterableChildFinder()})
>>> registry.has_child_finder(list)
True
>>> registry.has_child_finder(tuple)
False

coola.iterator.bfs.ChildFinderRegistry.iterate ¶

iterate(data: Any) -> Iterator[Any]

Perform a breadth-first traversal over a nested data structure.

This method traverses the input data using breadth-first search (BFS). Container objects (mappings and iterables, excluding strings and bytes) are expanded using registered child finders. Only non-container (leaf) values are yielded.

Containers themselves are never yielded, even if they are empty.

Parameters:

Name	Type	Description	Default
`data`	`Any`	The data structure to traverse.	required

Yields:

Type	Description
`Any`	Atomic (non-container) values in breadth-first order.

Example

>>> from coola.iterator.bfs import (
...     ChildFinderRegistry,
...     IterableChildFinder,
...     MappingChildFinder,
... )
>>> registry = ChildFinderRegistry(
...     {list: IterableChildFinder(), dict: MappingChildFinder()}
... )
>>> list(registry.iterate({"a": [1, 2], "b": [3, 4], "c": 5, "d": {"e": 6}}))
[5, 1, 2, 3, 4, 6]

coola.iterator.bfs.ChildFinderRegistry.register ¶

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

Register a child finder for a given data type.

This method associates a specific BaseChildFinder with a Python type. When an object of this type (or a subclass) is encountered during traversal, the registered child finder will be used.

The internal cache is cleared after registration to ensure consistency.

Parameters:

Name	Type	Description	Default
`data_type`	`type`	The Python type to register (e.g., `list`, `dict`, or a custom class).	required
`child_finder`	`BaseChildFinder[Any]`	The child finder instance responsible for extracting children from objects of this type.	required
`exist_ok`	`bool`	If `True`, allows overwriting an existing registration. If `False`, raises an error.	`False`

Raises:

Type	Description
`RuntimeError`	If the type is already registered and `exist_ok` is `False`.

Example

>>> from coola.iterator.bfs import ChildFinderRegistry, IterableChildFinder
>>> registry = ChildFinderRegistry()
>>> registry.register(list, IterableChildFinder())
>>> registry.has_child_finder(list)
True

coola.iterator.bfs.ChildFinderRegistry.register_many ¶

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

Register multiple child finders at once.

Parameters:

Name	Type	Description	Default
`mapping`	`Mapping[type, BaseChildFinder[Any]]`	A mapping from Python types to their corresponding child finders.	required
`exist_ok`	`bool`	If `True`, allows overwriting existing registrations.	`False`

Raises:

Type	Description
`RuntimeError`	If any type is already registered and `exist_ok` is `False`.

Example

>>> from coola.iterator.bfs import (
...     ChildFinderRegistry,
...     IterableChildFinder,
...     MappingChildFinder,
... )
>>> registry = ChildFinderRegistry()
>>> registry.register_many({list: IterableChildFinder(), dict: MappingChildFinder()})
>>> registry.has_child_finder(list), registry.has_child_finder(dict)
(True, True)

coola.iterator.bfs.DefaultChildFinder ¶

Bases: BaseChildFinder[Any]

Default child finder for breadth-first search traversal.

This child finder serves as the fallback handler for objects that don't have
a specialized child finder registered. It treats the input data as a leaf node
with no children, so it yields nothing during traversal.

The DefaultChildFinder is typically used for:
- Primitive types (int, float, str, bool, None)
- Objects without internal structure to traverse
- Terminal nodes in a data structure

Example

>>> from coola.iterator.bfs import DefaultChildFinder
>>> child_finder = DefaultChildFinder()
>>> list(child_finder.find_children(42))
[]
>>> list(child_finder.find_children("hello"))
[]

coola.iterator.bfs.IterableChildFinder ¶

Bases: BaseChildFinder[Iterable[Any]]

Child finder for iterable objects.

This child finder handles iterable objects by yielding each element of the iterable. It works with lists, tuples, sets, strings, and any object implementing the Iterable protocol.

Example

>>> from coola.iterator.bfs import IterableChildFinder
>>> child_finder = IterableChildFinder()
>>> list(child_finder.find_children((4, 2, 1)))
[4, 2, 1]
>>> list(child_finder.find_children("hello"))
['h', 'e', 'l', 'l', 'o']

coola.iterator.bfs.MappingChildFinder ¶

Bases: BaseChildFinder[Mapping[Any, Any]]

Child finder for iterable objects.

This child finder handles iterable objects by yielding each element of the iterable. It works with lists, tuples, sets, strings, and any object implementing the Mapping protocol.

Example

>>> from coola.iterator.bfs import MappingChildFinder
>>> child_finder = MappingChildFinder()
>>> list(child_finder.find_children({"a": 1, "b": 2}))
[1, 2]
>>> list(child_finder.find_children({"a": {"b": 1, "c": 2}, "d": 3}))
[{'b': 1, 'c': 2}, 3]

coola.iterator.bfs.bfs_iterate ¶

bfs_iterate(
    data: Any, registry: ChildFinderRegistry | None = None
) -> Iterator[Any]

Perform Depth-First Search (DFS) iteration over nested data structures (lists, dicts, tuples, sets, etc.).

This function yields elements from the data structure in a DFS manner, recursively traversing all levels of nested structures. It uses the appropriate iterators registered for the data types (e.g., lists, dictionaries, etc.).

Parameters:

Name	Type	Description	Default
`data`	`Any`	The nested data structure to traverse. Can be a list, tuple, dict, set, or any other registered type.	required
`registry`	`ChildFinderRegistry \| None`	The registry to resolve iterators for nested data. If `None`, the default registry is used.	`None`

Yields:

Type	Description
`Any`	Atomic leaf values in BFS order (excludes containers even if empty)

Example

>>> from coola.iterator import bfs_iterate
>>> list(bfs_iterate({"a": 1, "b": "abc"}))
[1, 'abc']
>>> list(bfs_iterate([1, [2, 3], {"x": 4}]))
[1, 2, 3, 4]

coola.iterator.bfs.get_default_registry ¶

get_default_registry() -> ChildFinderRegistry

Get or create the default global registry for iterators.

This function returns a singleton instance of the ChildFinderRegistry, which is pre-configured with iterators for common Python types, including iterables (lists, tuples), mappings (dicts), sets, and scalars (int, float, str, bool). The registry is used to look up the appropriate iterator for a given data structure during iteration.

Returns:

Type	Description
`ChildFinderRegistry`	An `ChildFinderRegistry` instance with iterators registered for common Python types.

Notes

The singleton pattern means any changes to the returned registry affect all future calls to this function. If an isolated registry is needed, create a new ChildFinderRegistry instance directly.

Example

>>> from coola.iterator.bfs import get_default_registry
>>> reg = get_default_registry()
>>> list(reg.iterate([1, 2, 3]))
[1, 2, 3]

coola.iterator.bfs.register_child_finders ¶

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

Register custom iterators 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, BaseChildFinder[Any]]`	A dictionary mapping Python types to their corresponding iterator instances.	required
`exist_ok`	`bool`	If `True`, existing registrations for types will be overwritten. If `False`, an error is raised when a type is already registered.	`False`

Example

>>> from coola.iterator.bfs import (
...     register_child_finders,
...     IterableChildFinder,
...     ChildFinderRegistry,
... )
>>> register_child_finders({list: IterableChildFinder()}, exist_ok=True)
>>> registry = get_default_registry()
>>> list(registry.iterate([1, 2, 3]))
[1, 2, 3]

coola.iterator.dfs ¶

Contain code to iterate over nested data with a Depth-First Search (DFS) strategy.

coola.iterator.dfs.BaseIterator ¶

Bases: ABC, Generic[T]

Abstract base class for depth-first search iterators.

This class defines the interface that all DFS iterators must implement. Iterators are responsible for traversing specific data types and yielding their elements during depth-first search traversal. Custom iterators can be registered with an IteratorRegistry to handle specific data types.

The generic type parameter T indicates the type of data this iterator is designed to handle, though the iterate method accepts Any for flexibility.

Notes

Subclasses must implement the iterate method.
For container types, use registry.iterate() to recursively traverse nested structures.
For leaf types, simply yield the data directly.

Example

>>> from coola.iterator.dfs import IteratorRegistry, DefaultIterator
>>> iterator = DefaultIterator()
>>> registry = IteratorRegistry()
>>> list(iterator.iterate(42, registry))
[42]
>>> list(iterator.iterate("hello", registry))
['hello']

coola.iterator.dfs.BaseIterator.iterate `abstractmethod` ¶

iterate(
    data: T, registry: IteratorRegistry
) -> Iterator[Any]

Traverse the data structure and yield elements depth-first.

This method defines how the iterator traverses its associated data type. For container or composite types, it should recursively traverse nested elements using registry.iterate() to delegate to appropriate iterators. For leaf types, it should yield the data directly.

Parameters:

Name	Type	Description	Default
`data`	`T`	The data structure to traverse. While typed as `T` for flexibility, implementations typically expect a specific type corresponding to the iterator's purpose.	required
`registry`	`IteratorRegistry`	The iterator registry used to resolve and dispatch iterators for nested data structures. Use `registry.iterate()` to recursively traverse nested elements.	required

Yields:

Type	Description
`Any`	Elements found during depth-first traversal. The exact type and
`Any`	nature of yielded elements depends on the specific iterator
`Any`	implementation and traversal strategy.

Notes

The registry parameter should be used to maintain consistent traversal behavior across different data types.
Implementations should handle the specific structure of their target data type appropriately.

Example

>>> from coola.iterator.dfs import IteratorRegistry, DefaultIterator
>>> iterator = DefaultIterator()
>>> registry = IteratorRegistry()
>>> list(iterator.iterate(42, registry))
[42]
>>> list(iterator.iterate("hello", registry))
['hello']

coola.iterator.dfs.DefaultIterator ¶

Bases: BaseIterator[Any]

Default iterator for depth-first search traversal of leaf nodes.

This iterator serves as the fallback handler for objects that don't have a specialized iterator registered. It treats the input data as a leaf node and yields it directly without further traversal.

The DefaultIterator is typically used for: - Primitive types (int, float, str, bool, None) - Objects without internal structure to traverse - Terminal nodes in a data structure

Example

>>> from coola.iterator.dfs import IteratorRegistry, DefaultIterator
>>> iterator = DefaultIterator()
>>> registry = IteratorRegistry()
>>> list(iterator.iterate(42, registry))
[42]
>>> list(iterator.iterate("hello", registry))
['hello']

coola.iterator.dfs.IterableIterator ¶

Bases: BaseIterator[Iterable[Any]]

Iterator for performing a depth-first traversal over iterable data structures.

This iterator recursively traverses through iterable structures such as lists, tuples, or other collections that implement the Iterable interface, yielding elements one by one.

Example

>>> from coola.iterator.dfs import IteratorRegistry, IterableIterator
>>> iterator = IterableIterator()
>>> registry = IteratorRegistry({list: iterator})
>>> # Iterating over a simple list
>>> list(iterator.iterate([1, 2, 3], registry))
[1, 2, 3]
>>> # Iterating over a string (iterable of characters)
>>> list(iterator.iterate("hello", registry))
['h', 'e', 'l', 'l', 'o']
>>> # Iterating over nested iterables (lists within lists)
>>> list(iterator.iterate([[1, 2, 3], [4, 5, 6]], registry))
[1, 2, 3, 4, 5, 6]

coola.iterator.dfs.IteratorRegistry ¶

Registry that manages iterators for different data types.

This registry stores iterators for various data types and handles the dispatching of the appropriate iterator based on the data type during iteration. It uses Method Resolution Order (MRO) to resolve the most specific iterator for a given data type. It also supports caching of iterators for performance optimization in repetitive iteration tasks.

Parameters:

Name	Type	Description	Default
`registry`	`dict[type, BaseIterator[Any]] \| None`	An optional dictionary mapping types to iterators. If provided, the registry is initialized with this mapping.	`None`

Attributes:

Name	Type	Description
`_registry`	`dict[type, BaseIterator[Any]]`	Internal mapping of registered data types to iterators.
`_default_iterator`	`BaseIterator[Any]`	The fallback iterator used for types not explicitly registered.
`_iterator_cache`	`dict[type, BaseIterator[Any]]`	Cache to speed up iterator lookups.

Example

Basic usage:

>>> from coola.iterator.dfs import IteratorRegistry, IterableIterator
>>> registry = IteratorRegistry({list: IterableIterator()})
>>> registry
IteratorRegistry(
  (<class 'list'>): IterableIterator()
)
>>> list(registry.iterate([1, 2, 3]))
[1, 2, 3]

Working with nested structures:

>>> from coola.iterator.dfs import get_default_registry
>>> registry = get_default_registry()
>>> data = {"a": [1, 2], "b": [3, 4]}
>>> list(registry.iterate(data))
[1, 2, 3, 4]

coola.iterator.dfs.IteratorRegistry.find_iterator ¶

find_iterator(data_type: type) -> BaseIterator[Any]

Find the appropriate iterator for a given type.

This method uses the MRO to find the most specific iterator. It caches the result for performance, so subsequent lookups are faster.

Parameters:

Name	Type	Description	Default
`data_type`	`type`	The data type for which to find an iterator.	required

Returns:

Type	Description
`BaseIterator[Any]`	The appropriate iterator for the data type.

Example

>>> from coola.iterator.dfs import IteratorRegistry, IterableIterator
>>> registry = IteratorRegistry({list: IterableIterator()})
>>> registry.find_iterator(list)
IterableIterator()
>>> registry.find_iterator(tuple)
DefaultIterator()

coola.iterator.dfs.IteratorRegistry.has_iterator ¶

has_iterator(data_type: type) -> bool

Check if an iterator is registered for a given data type.

This method checks for direct registration. Even if this method returns False, a suitable iterator might still be found using the MRO lookup.

Parameters:

Name	Type	Description	Default
`data_type`	`type`	The type to check.	required

Returns:

Type	Description
`bool`	`True` if an iterator is registered for the type, `False` otherwise.

Example

>>> from coola.iterator.dfs import IteratorRegistry, IterableIterator
>>> registry = IteratorRegistry({list: IterableIterator()})
>>> registry.has_iterator(list)
True
>>> registry.has_iterator(tuple)
False

coola.iterator.dfs.IteratorRegistry.iterate ¶

iterate(data: Any) -> Iterator[Any]

Perform depth-first iteration over a data structure.

This method uses the appropriate iterator for the data type, which may be retrieved via the registry. The iterator will recursively traverse the data structure, yielding elements based on its specific implementation.

Parameters:

Name	Type	Description	Default
`data`	`Any`	The data structure to iterate over.	required

Yields:

Type	Description
`Any`	The elements of the data structure according to the appropriate iterator's traversal logic.

Example

>>> from coola.iterator.dfs import IteratorRegistry, IterableIterator, MappingIterator
>>> registry = IteratorRegistry({list: IterableIterator(), dict: MappingIterator()})
>>> list(registry.iterate({"a": [1, 2], "b": [3, 4]}))
[1, 2, 3, 4]

coola.iterator.dfs.IteratorRegistry.register ¶

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

Register an iterator for a given data type.

This method associates a specific iterator with a type. If data of this type is iterated, the registered iterator will be used. The cache is cleared after a registration to ensure consistency.

Parameters:

Name	Type	Description	Default
`data_type`	`type`	The Python type to register (e.g., `list`, `dict`, custom types).	required
`iterator`	`BaseIterator[Any]`	The iterator instance that handles this type.	required
`exist_ok`	`bool`	If `True`, allows overwriting an existing registration. If `False`, raises an error.	`False`

Raises:

Type	Description
`RuntimeError`	If the type is already registered and `exist_ok` is `False`.

Example

>>> from coola.iterator.dfs import IteratorRegistry, IterableIterator
>>> registry = IteratorRegistry()
>>> registry.register(list, IterableIterator())
>>> registry.has_iterator(list)
True

coola.iterator.dfs.IteratorRegistry.register_many ¶

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

Register multiple iterators at once.

This method allows for bulk registration of iterators for multiple data types.

Parameters:

Name	Type	Description	Default
`mapping`	`Mapping[type, BaseIterator[Any]]`	A dictionary mapping Python types to their respective iterators.	required
`exist_ok`	`bool`	If `True`, allows overwriting existing registrations.	`False`

Raises:

Type	Description
`RuntimeError`	If any type is already registered and `exist_ok` is `False`.

Example

>>> from coola.iterator.dfs import IteratorRegistry, IterableIterator, MappingIterator
>>> registry = IteratorRegistry()
>>> registry.register_many({list: IterableIterator(), dict: MappingIterator()})
>>> registry
IteratorRegistry(
  (<class 'list'>): IterableIterator()
  (<class 'dict'>): MappingIterator()
)

coola.iterator.dfs.MappingIterator ¶

Bases: BaseIterator[Mapping[Any, Any]]

Iterator for depth-first traversal of mapping data structures.

This iterator handles dict-like objects by recursively iterating over their values. Keys are not yielded during iteration - only the values are traversed. If values contain nested structures (lists, dicts, etc.), those are recursively iterated as well.

Example

>>> from coola.iterator.dfs import IteratorRegistry, MappingIterator, IterableIterator
>>> iterator = MappingIterator()
>>> registry = IteratorRegistry({dict: iterator, list: IterableIterator()})
>>> # Simple dictionary with scalar values
>>> list(iterator.iterate({"a": 1, "b": 2}, registry))
[1, 2]
>>> # Nested dictionary with scalar values
>>> list(iterator.iterate({"a": {"b": 1, "c": 2}, "d": 3}, registry))
[1, 2, 3]
>>> # Dictionary with list values
>>> list(iterator.iterate({"x": [1, 2], "y": [3, 4]}, registry))
[1, 2, 3, 4]

coola.iterator.dfs.dfs_iterate ¶

dfs_iterate(
    data: Any, registry: IteratorRegistry | None = None
) -> Iterator[Any]

Perform Depth-First Search (DFS) iteration over nested data structures (lists, dicts, tuples, sets, etc.).

This function yields elements from the data structure in a DFS manner, recursively traversing all levels of nested structures. It uses the appropriate iterators registered for the data types (e.g., lists, dictionaries, etc.).

Parameters:

Name	Type	Description	Default
`data`	`Any`	The nested data structure to traverse. Can be a list, tuple, dict, set, or any other registered type.	required
`registry`	`IteratorRegistry \| None`	The registry to resolve iterators for nested data. If `None`, the default registry is used.	`None`

Yields:

Type	Description
`Any`	The elements from the nested data structure in DFS order.

Example

>>> from coola.iterator import dfs_iterate
>>> list(dfs_iterate({"a": 1, "b": "abc"}))
[1, 'abc']
>>> list(dfs_iterate([1, [2, 3], {"x": 4}]))
[1, 2, 3, 4]

coola.iterator.dfs.get_default_registry ¶

get_default_registry() -> IteratorRegistry

Get or create the default global registry for iterators.

This function returns a singleton instance of the IteratorRegistry, which is pre-configured with iterators for common Python types, including iterables (lists, tuples), mappings (dicts), sets, and scalars (int, float, str, bool). The registry is used to look up the appropriate iterator for a given data structure during iteration.

Returns:

Type	Description
`IteratorRegistry`	An `IteratorRegistry` instance with iterators registered for common Python types.

Notes

The singleton pattern means any changes to the returned registry affect all future calls to this function. If an isolated registry is needed, create a new IteratorRegistry instance directly.

Example

>>> from coola.iterator.dfs import get_default_registry
>>> reg = get_default_registry()
>>> list(reg.iterate([1, 2, 3]))
[1, 2, 3]

coola.iterator.dfs.register_iterators ¶

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

Register custom iterators 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, BaseIterator[Any]]`	A dictionary mapping Python types to their corresponding iterator instances.	required
`exist_ok`	`bool`	If `True`, existing registrations for types will be overwritten. If `False`, an error is raised when a type is already registered.	`False`

Example

>>> from coola.iterator.dfs import register_iterators, IterableIterator, IteratorRegistry
>>> register_iterators({list: IterableIterator()}, exist_ok=True)
>>> registry = get_default_registry()
>>> list(registry.iterate([1, 2, 3]))
[1, 2, 3]

Iterator

coola.iterator ¶

coola.iterator.bfs_iterate ¶

coola.iterator.dfs_iterate ¶

coola.iterator.filter_by_type ¶

coola.iterator.bfs ¶

coola.iterator.bfs.BaseChildFinder ¶

coola.iterator.bfs.BaseChildFinder.find_children abstractmethod ¶

coola.iterator.bfs.ChildFinderRegistry ¶

coola.iterator.bfs.ChildFinderRegistry.find_child_finder ¶

coola.iterator.bfs.ChildFinderRegistry.find_children ¶

coola.iterator.bfs.ChildFinderRegistry.has_child_finder ¶

coola.iterator.bfs.ChildFinderRegistry.iterate ¶

coola.iterator.bfs.ChildFinderRegistry.register ¶

coola.iterator.bfs.ChildFinderRegistry.register_many ¶

coola.iterator.bfs.DefaultChildFinder ¶

coola.iterator.bfs.IterableChildFinder ¶

coola.iterator.bfs.MappingChildFinder ¶

coola.iterator.bfs.bfs_iterate ¶

coola.iterator.bfs.get_default_registry ¶

coola.iterator.bfs.register_child_finders ¶

coola.iterator.dfs ¶

coola.iterator.dfs.BaseIterator ¶

coola.iterator.dfs.BaseIterator.iterate abstractmethod ¶

coola.iterator.dfs.DefaultIterator ¶

coola.iterator.dfs.IterableIterator ¶

coola.iterator.dfs.IteratorRegistry ¶

coola.iterator.dfs.IteratorRegistry.find_iterator ¶

coola.iterator.dfs.IteratorRegistry.has_iterator ¶

coola.iterator.dfs.IteratorRegistry.iterate ¶

coola.iterator.dfs.IteratorRegistry.register ¶

coola.iterator.dfs.IteratorRegistry.register_many ¶

coola.iterator.dfs.MappingIterator ¶

coola.iterator.dfs.dfs_iterate ¶

coola.iterator.dfs.get_default_registry ¶

coola.iterator.dfs.register_iterators ¶

coola.iterator.bfs.BaseChildFinder.find_children `abstractmethod` ¶

coola.iterator.dfs.BaseIterator.iterate `abstractmethod` ¶