​
prefect.utilities.collections

Utilities for extensions of and operations on Python collections.

​
Functions

​
dict_to_flatdict 

dict_to_flatdict(dct: NestedDict[KT, VT]) -> dict[tuple[KT, ...], VT]

Converts a (nested) dictionary to a flattened representation.

Each key of the flat dict will be a CompoundKey tuple containing the “chain of keys” for the corresponding value.

Args:

  • dct: The dictionary to flatten

Returns:

  • A flattened dict of the same type as dct

​
flatdict_to_dict 

flatdict_to_dict(dct: dict[tuple[KT, ...], VT]) -> NestedDict[KT, VT]

Converts a flattened dictionary back to a nested dictionary.

Args:

  • dct: The dictionary to be nested. Each key should be a tuple of keys as generated by dict_to_flatdict

Returns A nested dict of the same type as dct

​
isiterable 

isiterable(obj: Any) -> bool

Return a boolean indicating if an object is iterable.

Excludes types that are iterable but typically used as singletons:

  • str
  • bytes
  • IO objects

​
ensure_iterable 

ensure_iterable(obj: Union[T, Iterable[T]]) -> Collection[T]

​
listrepr 

listrepr(objs: Iterable[Any], sep: str = ' ') -> str

​
extract_instances 

extract_instances(objects: Iterable[Any], types: Union[type[T], tuple[type[T], ...]] = object) -> Union[list[T], dict[type[T], list[T]]]

Extract objects from a file and returns a dict of type -> instances

Args:

  • objects: An iterable of objects
  • types: A type or tuple of types to extract, defaults to all objects

Returns:

  • If a single type is given: a list of instances of that type
  • If a tuple of types is given: a mapping of type to a list of instances

​
batched_iterable 

batched_iterable(iterable: Iterable[T], size: int) -> Generator[tuple[T, ...], None, None]

Yield batches of a certain size from an iterable

Args:

  • iterable: An iterable
  • size: The batch size to return

​
visit_collection 

visit_collection(expr: Any, visit_fn: Union[Callable[[Any, dict[str, VT]], Any], Callable[[Any], Any]]) -> Optional[Any]

Visits and potentially transforms every element of an arbitrary Python collection.

If an element is a Python collection, it will be visited recursively. If an element is not a collection, visit_fn will be called with the element. The return value of visit_fn can be used to alter the element if return_data is set to True.

Note:

  • When return_data is True, a copy of each collection is created only if visit_fn modifies an element within that collection. This approach minimizes performance penalties by avoiding unnecessary copying.
  • When return_data is False, no copies are created, and only side effects from visit_fn are applied. This mode is faster and should be used when no transformation of the collection is required, because it never has to copy any data.

Supported types:

  • List (including iterators)
  • Tuple
  • Set
  • Dict (note: keys are also visited recursively)
  • Dataclass
  • Pydantic model
  • Prefect annotations

Note that visit_collection will not consume generators or async generators, as it would prevent the caller from iterating over them.

Args:

  • expr: A Python object or expression.
  • visit_fn: A function that will be applied to every non-collection element of expr. The function can accept one or two arguments. If two arguments are accepted, the second argument will be the context dictionary.
  • return_data: If True, a copy of expr containing data modified by visit_fn will be returned. This is slower than return_data=False (the default).
  • max_depth: Controls the depth of recursive visitation. If set to zero, no recursion will occur. If set to a positive integer N, visitation will only descend to N layers deep. If set to any negative integer, no limit will be enforced and recursion will continue until terminal items are reached. By default, recursion is unlimited.
  • context: An optional dictionary. If passed, the context will be sent to each call to the visit_fn. The context can be mutated by each visitor and will be available for later visits to expressions at the given depth. Values will not be available “up” a level from a given expression. The context will be automatically populated with an ‘annotation’ key when visiting collections within a BaseAnnotation type. This requires the caller to pass context={} and will not be activated by default.
  • remove_annotations: If set, annotations will be replaced by their contents. By default, annotations are preserved but their contents are visited.
  • _seen: A set of object ids that have already been visited. This prevents infinite recursion when visiting recursive data structures.

Returns:

  • The modified collection if return_data is True, otherwise None.

​
remove_nested_keys 

remove_nested_keys(keys_to_remove: list[HashableT], obj: Union[NestedDict[HashableT, VT], Any]) -> Union[NestedDict[HashableT, VT], Any]

Recurses a dictionary returns a copy without all keys that match an entry in key_to_remove. Return obj unchanged if not a dictionary.

Args:

  • keys_to_remove: A list of keys to remove from obj obj: The object to remove keys from.

Returns:

  • obj without keys matching an entry in keys_to_remove if obj is a dictionary. obj if obj is not a dictionary.

​
distinct 

distinct(iterable: Iterable[Union[T, HashableT]], key: Optional[Callable[[T], Hashable]] = None) -> Iterator[Union[T, HashableT]]

​
get_from_dict 

get_from_dict(dct: NestedDict[str, VT], keys: Union[str, list[str]], default: Optional[R] = None) -> Union[VT, R, None]

Fetch a value from a nested dictionary or list using a sequence of keys.

This function allows to fetch a value from a deeply nested structure of dictionaries and lists using either a dot-separated string or a list of keys. If a requested key does not exist, the function returns the provided default value.

Args:

  • dct: The nested dictionary or list from which to fetch the value.
  • keys: The sequence of keys to use for access. Can be a dot-separated string or a list of keys. List indices can be included in the sequence as either integer keys or as string indices in square brackets.
  • default: The default value to return if the requested key path does not exist. Defaults to None.

Returns:

  • The fetched value if the key exists, or the default value if it does not.

Examples:

get_from_dict({'a': {'b': {'c': [1, 2, 3, 4]}}}, 'a.b.c[1]') # 2
get_from_dict({'a': {'b': [0, {'c': [1, 2]}]}}, ['a', 'b', 1, 'c', 1]) # 2
get_from_dict({'a': {'b': [0, {'c': [1, 2]}]}}, 'a.b.1.c.2', 'default') # 'default'

​
set_in_dict 

set_in_dict(dct: NestedDict[str, VT], keys: Union[str, list[str]], value: VT) -> None

Sets a value in a nested dictionary using a sequence of keys.

This function allows to set a value in a deeply nested structure of dictionaries and lists using either a dot-separated string or a list of keys. If a requested key does not exist, the function will create it as a new dictionary.

Args:

  • dct: The dictionary to set the value in.
  • keys: The sequence of keys to use for access. Can be a dot-separated string or a list of keys.
  • value: The value to set in the dictionary.

Returns:

  • The modified dictionary with the value set at the specified key path.

Raises:

  • KeyError: If the key path exists and is not a dictionary.

​
deep_merge 

deep_merge(dct: NestedDict[str, VT1], merge: NestedDict[str, VT2]) -> NestedDict[str, Union[VT1, VT2]]

Recursively merges merge into dct.

Args:

  • dct: The dictionary to merge into.
  • merge: The dictionary to merge from.

Returns:

  • A new dictionary with the merged contents.

​
deep_merge_dicts 

deep_merge_dicts(*dicts: NestedDict[str, Any]) -> NestedDict[str, Any]

Recursively merges multiple dictionaries.

Args:

  • dicts: The dictionaries to merge.

Returns:

  • A new dictionary with the merged contents.

​
Classes

​
AutoEnum

An enum class that automatically generates value from variable names.

This guards against common errors where variable names are updated but values are not.

In addition, because AutoEnums inherit from str, they are automatically JSON-serializable.

See https://docs.python.org/3/library/enum.html#using-automatic-values

Methods:

​
auto 

auto() -> str

Exposes enum.auto() to avoid requiring a second import to use AutoEnum

​
StopVisiting

A special exception used to stop recursive visits in visit_collection.

When raised, the expression is returned without modification and recursive visits in that path will end.