Utils

coola.utils ¶

Shared helper utilities used by public and internal coola APIs.

The package exports nothing: import a function from its module. The core helpers, used by the other coola packages, are imports (optional dependencies), fallback, fork, integer, introspection, mapping, path, safe, selection, singleton (e.g. LazySingleton), stats and version. The standalone helpers are array, tensor, batching, bloom_filter, date, env_vars, file_size, git, logging, lru, password, recursion and timing. Functions that turn values into text are in coola.formatting.

coola.utils.array ¶

Implement some utility functions for numpy.ndarrays.

coola.utils.array.to_array ¶

to_array(
    data: Sequence[int | float] | Tensor | ndarray,
) -> ndarray

Convert the input to a numpy.ndarray.

Parameters:

Name	Type	Description	Default
`data`	`Sequence[int \| float] \| Tensor \| ndarray`	The data to convert to a NumPy array.	required

Returns:

Type	Description
`ndarray`	A NumPy array. A NumPy array is returned as is, and the array of a tensor or of a NumPy-compatible input may share its memory with the input; a list is copied. A `bfloat16` tensor (not supported by NumPy) is converted to `float32`.

Example

>>> from coola.utils.array import to_array
>>> x = to_array([1, 2, 3, 4, 5])
>>> x
array([1, 2, 3, 4, 5])
>>> import torch
>>> x = to_array(torch.tensor([1, 2, 3, 4, 5]))
>>> x
array([1, 2, 3, 4, 5])

coola.utils.batching ¶

Contain utility functions for batching.

coola.utils.batching.batchify ¶

batchify(
    items: Iterable[T], *, size: int
) -> Iterator[tuple[T, ...]]

Generate batches of items from an iterable.

Splits items into consecutive non-overlapping tuples of length size. The last batch may be shorter than size if the number of items is not evenly divisible.

Parameters:

Name	Type	Description	Default
`items`	`Iterable[T]`	The iterable of items to batch.	required
`size`	`int`	The maximum number of items per batch. Must be >= 1.	required

Yields:

Type	Description
`tuple[T, ...]`	Tuples of at most `size` items each. Nothing is yielded when `items` is empty.

Raises:

Type	Description
`ValueError`	If `size` is less than 1.

Example

>>> from coola.utils.batching import batchify
>>> list(batchify([1, 2, 3, 4, 5], size=2))
[(1, 2), (3, 4), (5,)]
>>> list(batchify([], size=3))
[]
>>> list(batchify(iter(range(6)), size=3))
[(0, 1, 2), (3, 4, 5)]

coola.utils.bloom_filter ¶

Contain the implementation of a Bloom filter.

coola.utils.bloom_filter.BloomFilter ¶

Fixed-memory approximate set-membership structure.

A Bloom filter answers "have I probably seen this item before?" using a fixed amount of memory. It guarantees no false negatives (an item that was added is always reported as present), at the cost of a tunable false-positive rate (an item that was never added may occasionally be reported as present). Memory usage is fixed up front based on the expected number of items and the desired false-positive rate, regardless of how many items are actually added. It is a standalone utility, typically used for approximate duplicate detection over large streams of items.

Warning

The filter is not thread-safe: concurrent add or add_and_check calls can lose bits. Protect it with a lock if it is shared between threads.

Parameters:

Name	Type	Description	Default
`expected_items`	`int`	Approximate number of unique items the filter is expected to hold. Used to size the bit array for the requested false-positive rate; overestimating is safe and simply uses more memory, while underestimating raises the effective false-positive rate as more items are added than planned for.	`1000000`
`fp_rate`	`float`	Target false-positive probability once approximately `expected_items` unique items have been added, expressed as a value in `(0, 1)`.	`0.01`

Raises:

Type	Description
`ValueError`	if `expected_items` is not positive or `fp_rate` is not in `(0, 1)`.

Example

>>> from coola.utils.bloom_filter import BloomFilter
>>> bf = BloomFilter(expected_items=1000, fp_rate=0.01)
>>> bf.add_and_check(b"hello")
False
>>> bf.add_and_check(b"hello")
True
>>> b"hello" in bf
True
>>> b"world" in bf
False

coola.utils.bloom_filter.BloomFilter.contains ¶

__contains__(item: object) -> bool

Indicate if an item is probably in the filter, without adding it.

Parameters:

Name	Type	Description	Default
`item`	`object`	The raw bytes to check.	required

Returns:

Type	Description
`bool`	`False` if `item` is definitely not in the filter,
`bool`	`True` if it probably is (subject to the false-positive
`bool`	rate).

Example

>>> from coola.utils.bloom_filter import BloomFilter
>>> bf = BloomFilter(expected_items=1000)
>>> b"hello" in bf
False

coola.utils.bloom_filter.BloomFilter.add ¶

add(item: bytes) -> None

Add an item to the filter.

Parameters:

Name	Type	Description	Default
`item`	`bytes`	The raw bytes to add.	required

Example

>>> from coola.utils.bloom_filter import BloomFilter
>>> bf = BloomFilter(expected_items=1000)
>>> bf.add(b"hello")
>>> b"hello" in bf
True

coola.utils.bloom_filter.BloomFilter.add_and_check ¶

add_and_check(item: bytes) -> bool

Add an item to the filter and report whether it was probably already present.

Parameters:

Name	Type	Description	Default
`item`	`bytes`	The raw bytes to add/check.	required

Returns:

Type	Description
`bool`	`True` if `item` was probably already in the filter
`bool`	before this call (i.e. it is probably a duplicate, subject
`bool`	to the filter's false-positive rate). `False` if it was
`bool`	definitely not present before this call. Either way, the
`bool`	item's bits are set, so subsequent calls with the same
`bool`	item will return `True`.

coola.utils.bloom_filter.BloomFilter.fill_ratio ¶

fill_ratio() -> float

Return the fraction of the bits that are set.

A ratio close to 1 means that the false-positive rate is much higher than the requested one.

Returns:

Type	Description
`float`	The fraction of set bits, in `[0, 1]`.

Example

>>> from coola.utils.bloom_filter import BloomFilter
>>> bf = BloomFilter(expected_items=1000)
>>> bf.fill_ratio()
0.0

coola.utils.date ¶

Contain utility functions to manipulate dates.

coola.utils.date.get_today_date ¶

get_today_date(
    timezone: str = "UTC",
    date_format: str = DEFAULT_DATE_FORMAT,
) -> str

Return the current date for the specified timezone.

Parameters:

Name	Type	Description	Default
`timezone`	`str`	The IANA timezone name to use when determining the current date. Defaults to `"UTC"`.	`'UTC'`
`date_format`	`str`	A `datetime.datetime.strftime` format string used to format the returned date. Defaults to `"%Y-%m-%d"`.	`DEFAULT_DATE_FORMAT`

Returns:

Type	Description
`str`	A string representing the current date in the given timezone, formatted according to `date_format`.

Example

>>> from coola.utils.date import get_today_date
>>> date = get_today_date()
>>> isinstance(date, str)
True
>>> date = get_today_date(timezone="America/New_York", date_format="%d/%m/%Y")
>>> isinstance(date, str)
True

coola.utils.date.to_date ¶

to_date(
    d: date | str, date_format: str = DEFAULT_DATE_FORMAT
) -> date

Convert the input to a date object.

Parameters:

Name	Type	Description	Default
`d`	`date \| str`	The date to convert. If a string is given, it is parsed using `date_format`. If a `datetime` is given, its date part is returned.	required
`date_format`	`str`	A `datetime.datetime.strptime` format string used to parse `d` when it is a string. Defaults to `"%Y-%m-%d"`.	`DEFAULT_DATE_FORMAT`

Returns:

Type	Description
`date`	The input as a `date` object.

Example

>>> from datetime import date
>>> from coola.utils.date import to_date
>>> to_date("2026-03-19")
datetime.date(2026, 3, 19)
>>> to_date(date(2026, 3, 19))
datetime.date(2026, 3, 19)

coola.utils.date.to_datetime ¶

to_datetime(d: datetime | str) -> datetime

Convert the input to a datetime object.

Parameters:

Name	Type	Description	Default
`d`	`datetime \| str`	The datetime to convert. If a string is given, it is parsed using `datetime.datetime.fromisoformat`.	required

Returns:

Type	Description
`datetime`	The input as a `datetime` object.

Example

>>> from datetime import datetime
>>> from coola.utils.date import to_datetime
>>> to_datetime("2026-03-19T12:00:00")
datetime.datetime(2026, 3, 19, 12, 0)
>>> to_datetime(datetime(2026, 3, 19, 12, 0))
datetime.datetime(2026, 3, 19, 12, 0)

coola.utils.env_vars ¶

Implement some utility functions to manage environment variables.

coola.utils.env_vars.check_env_vars ¶

check_env_vars(
    var_names: Sequence[str],
    raise_on_missing: bool = False,
    missing_level: int = WARNING,
) -> dict[str, bool]

Check whether each environment variable in var_names is defined.

Logs the status of each variable to the terminal. A variable set to an empty (or blank) string is reported as not defined, as in get_required_env_var.

Parameters:

Name	Type	Description	Default
`var_names`	`Sequence[str]`	The sequence of environment variable names to check.	required
`raise_on_missing`	`bool`	If `True`, raises `OSError` when any variable is missing. Defaults to `False`.	`False`
`missing_level`	`int`	The logging level used to report a missing variable. Use `logging.INFO` when the variables are optional. Defaults to `logging.WARNING`.	`WARNING`

Returns:

Type	Description
`dict[str, bool]`	A mapping of variable name to `True` if defined, `False` otherwise.

Raises:

Type	Description
`OSError`	If `raise_on_missing` is `True` and one or more variables are missing.

Example

>>> import os
>>> from coola.utils.env_vars import check_env_vars
>>> os.environ["MY_VAR"] = "abc"
>>> check_env_vars(["MY_VAR", "MISSING_VAR"])
{'MY_VAR': True, 'MISSING_VAR': False}

coola.utils.env_vars.get_required_env_var ¶

get_required_env_var(name: str) -> str

Retrieve a required environment variable with validation.

This function fetches an environment variable and ensures it exists and contains a non-empty value after stripping whitespace. If the variable is missing or empty, a ValueError is raised with a descriptive message.

Parameters:

Name	Type	Description	Default
`name`	`str`	The name of the environment variable to retrieve.	required

Returns:

Type	Description
`str`	The value of the environment variable with leading/trailing whitespace removed.

Raises:

Type	Description
`ValueError`	If the environment variable is not set or contains only whitespace.

Example

>>> import os
>>> from coola.utils.env_vars import get_required_env_var
>>> os.environ["API_KEY"] = "my-secret-key"
>>> get_required_env_var("API_KEY")
'my-secret-key'
>>> get_required_env_var("MISSING_VAR")
Traceback (most recent call last):
    ...
ValueError: Environment variable 'MISSING_VAR' is required but not set or is empty

coola.utils.env_vars.temp_env_vars ¶

temp_env_vars(env_vars: dict[str, Any]) -> Generator[None]

Context manager to temporarily set or modify environment variables.

Changes are automatically reverted when exiting the context, even if an exception occurs. If a variable already exists, its original value is saved and restored on exit. If a variable doesn't exist, it is created temporarily and removed on exit. A value of None unsets the variable for the block (it is restored on exit if it existed). All operations are guaranteed to execute via try/finally, even if an exception is raised inside the with block.

Note

This context manager mutates os.environ, which is process-wide: it is not thread-safe.

Parameters:

Name	Type	Description	Default
`env_vars`	`dict[str, Any]`	The environment variables to set, as a mapping of variable name to value. Values are automatically converted to strings, except `None`, which unsets the variable.	required

Yields:

Type	Description
`Generator[None]`	`None`. The environment variables are restored when the block exits.

Example

>>> import os
>>> from coola.utils.env_vars import temp_env_vars
>>> # Temporarily override an existing variable
>>> os.environ["MY_HOME"] = "/original/home"
>>> with temp_env_vars({"MY_HOME": "/tmp/home"}):
...     print(os.environ["MY_HOME"])
...
/tmp/home
>>> print(os.environ["MY_HOME"])
/original/home
>>> # Temporarily create new variables
>>> with temp_env_vars({"TEMP_API_KEY": "secret123", "TEMP_DEBUG": "true"}):
...     print(os.environ["TEMP_API_KEY"])
...     print(os.environ["TEMP_DEBUG"])
...
secret123
true
>>> print(os.environ.get("TEMP_API_KEY"))
None
>>> # Temporarily unset a variable
>>> with temp_env_vars({"MY_HOME": None}):
...     print(os.environ.get("MY_HOME"))
...
None
>>> print(os.environ["MY_HOME"])
/original/home

Note

Changes to os.environ affect the entire Python process, not just the current thread. Use with caution in multi-threaded applications.

coola.utils.fallback ¶

Fallback helpers used when optional dependencies are unavailable.

coola.utils.file_size ¶

Contain file size utility functions.

coola.utils.file_size.get_file_size ¶

get_file_size(path: Path) -> int

Return the size of a file or directory, in bytes.

If path is a directory, returns the total size of all files it contains (recursively). Symlinks inside the directory are neither followed nor counted, and files deleted while the directory is being scanned are skipped. If path itself is a symlink to a file, the size of the file is returned (the link is followed).

Parameters:

Name	Type	Description	Default
`path`	`Path`	The path to the file or directory.	required

Returns:

Type	Description
`int`	The size in bytes.

Raises:

Type	Description
`FileNotFoundError`	if `path` does not exist.

Example

>>> from pathlib import Path
>>> from coola.utils.file_size import get_file_size
>>> get_file_size(Path(__file__))  # doctest: +SKIP
512

coola.utils.fork ¶

Reset the state of the objects in a forked process.

A process created with os.fork (e.g. a worker of a pre-fork server, a torch.utils.data.DataLoader worker or multiprocessing with the fork start method) starts with a copy of the memory of its parent, but only with the thread that forked. Two kinds of state need a reset in the child process:

the state of the identifier generators: without a reset, the parent and its children would produce the same random values and the same counters, so they could produce the same identifiers;
the locks: a lock that another thread of the parent held at the time of the fork stays locked forever in the child, because the owner does not exist there.

The objects that register with track_fork are reset in the child process right after the fork.

coola.utils.fork.ForkAware ¶

Bases: Protocol

Protocol for the objects that need to reset their state in a forked child process.

coola.utils.fork.track_fork ¶

track_fork(obj: ForkAware) -> None

Reset an object in the child process of every future fork.

The object is only weakly referenced: it is not kept alive by this function.

Parameters:

Name	Type	Description	Default
`obj`	`ForkAware`	The object to track. Its `_reset_after_fork` method is called in the child process after a fork.	required

Example

>>> from coola.utils.fork import track_fork
>>> class Counter:
...     def __init__(self):
...         self.value = 0
...         track_fork(self)
...     def _reset_after_fork(self):
...         self.value = 0
...
>>> counter = Counter()

coola.utils.git ¶

Implement some utility functions to retrieve git repository information.

coola.utils.git.get_current_branch ¶

get_current_branch(cwd: Path | str | None = None) -> str

Get the name of the current git branch.

Parameters:

Name	Type	Description	Default
`cwd`	`Path \| str \| None`	The working directory where the command is executed. If `None`, the current working directory is used.	`None`

Returns:

Type	Description
`str`	The name of the current branch, or the literal `"HEAD"` if the checkout is detached.

Raises:

Type	Description
`OSError`	If the current directory is not part of a git repository or the branch name cannot be resolved.

Example

>>> from coola.utils.git import get_current_branch
>>> get_current_branch()  # doctest: +SKIP

coola.utils.git.get_current_commit_hash ¶

get_current_commit_hash(
    cwd: Path | str | None = None, short: bool = False
) -> str

Get the hash of the current git commit.

Parameters:

Name	Type	Description	Default
`cwd`	`Path \| str \| None`	The working directory where the command is executed. If `None`, the current working directory is used.	`None`
`short`	`bool`	If `True`, returns the abbreviated commit hash. Defaults to `False`.	`False`

Returns:

Type	Description
`str`	The hash of the current commit.

Raises:

Type	Description
`OSError`	If the current directory is not part of a git repository or the commit hash cannot be resolved.

Example

>>> from coola.utils.git import get_current_commit_hash
>>> get_current_commit_hash()  # doctest: +SKIP

coola.utils.git.is_git_repo ¶

is_git_repo(cwd: Path | str | None = None) -> bool

Indicate if the given directory is part of a git repository.

Parameters:

Name	Type	Description	Default
`cwd`	`Path \| str \| None`	The working directory to check. If `None`, the current working directory is used.	`None`

Returns:

Type	Description
`bool`	`True` if the directory is part of a git repository, otherwise `False`.

Example

>>> from coola.utils.git import is_git_repo
>>> is_git_repo()  # doctest: +SKIP

coola.utils.git.run_git_command ¶

run_git_command(
    args: list[str], cwd: Path | str | None = None
) -> str

Run a git command and return its stripped stdout.

Parameters:

Name	Type	Description	Default
`args`	`list[str]`	The git command arguments, excluding the `git` executable itself.	required
`cwd`	`Path \| str \| None`	The working directory where the command is executed. If `None`, the current working directory is used.	`None`

Returns:

Type	Description
`str`	The stripped standard output of the command.

Raises:

Type	Description
`OSError`	If the git command fails or `git` is not available.

Example

>>> from coola.utils.git import run_git_command
>>> run_git_command(["rev-parse", "HEAD"])  # doctest: +SKIP

Warning

args is passed to git without going through a shell, so it is not vulnerable to shell injection. However, the caller is still responsible for not passing untrusted input as args, since arbitrary git subcommands and options (e.g. ones that read or write files) could be executed.

coola.utils.imports ¶

Helpers to detect, lazily import, and validate optional dependencies.

coola.utils.imports.MissingPackageError ¶

Bases: ModuleNotFoundError, RuntimeError

Raised when an optional package is required but not installed.

It is a ModuleNotFoundError (so except ImportError works as expected). It also derives from RuntimeError, which this error used to be, so existing except RuntimeError code keeps working.

Example

>>> from coola.utils.imports import MissingPackageError, check_package
>>> try:
...     check_package("not_an_installed_package")
... except ImportError as error:
...     print(type(error).__name__, error.name)
...
MissingPackageError not_an_installed_package

coola.utils.imports.availability_cache ¶

availability_cache(fn: F) -> F

Cache the result of an availability check and register it, so clear_availability_caches can reset it.

Use it instead of functools.cache on the functions that check if a package is available.

Parameters:

Name	Type	Description	Default
`fn`	`F`	The availability check to cache.	required

Returns:

Type	Description
`F`	The cached function, which has a `cache_clear` method.

Example

>>> from coola.utils.imports import availability_cache
>>> @availability_cache
... def is_my_package_available() -> bool:
...     return True
...
>>> is_my_package_available()
True

coola.utils.imports.check_colorlog ¶

check_colorlog() -> None

Check if the colorlog package is installed.

Raises:

Type	Description
`MissingPackageError`	if the `colorlog` package is not installed.

Example

>>> from coola.utils.imports import check_colorlog
>>> check_colorlog()

coola.utils.imports.check_jax ¶

check_jax() -> None

Check if the jax package is installed.

Raises:

Type	Description
`MissingPackageError`	if the `jax` package is not installed.

Example

>>> from coola.utils.imports import check_jax
>>> check_jax()

coola.utils.imports.check_numpy ¶

check_numpy() -> None

Check if the numpy package is installed.

Raises:

Type	Description
`MissingPackageError`	if the `numpy` package is not installed.

Example

>>> from coola.utils.imports import check_numpy
>>> check_numpy()

coola.utils.imports.check_package ¶

check_package(
    package: str, *, install: str | None = None
) -> None

Check if the given package is installed and can be imported.

Parameters:

Name	Type	Description	Default
`package`	`str`	The package name.	required
`install`	`str \| None`	The name to give to `pip install` when it differs from `package`. By default, it is `package`.	`None`

Raises:

Type	Description
`MissingPackageError`	if the package is not available.

Example

>>> from coola.utils.imports import check_package
>>> check_package("os")

coola.utils.imports.check_packaging ¶

check_packaging() -> None

Check if the packaging package is installed.

Raises:

Type	Description
`MissingPackageError`	if the `packaging` package is not installed.

Example

>>> from coola.utils.imports import check_packaging
>>> check_packaging()

coola.utils.imports.check_pandas ¶

check_pandas() -> None

Check if the pandas package is installed.

Raises:

Type	Description
`MissingPackageError`	if the `pandas` package is not installed.

Example

>>> from coola.utils.imports import check_pandas
>>> check_pandas()

coola.utils.imports.check_polars ¶

check_polars() -> None

Check if the polars package is installed.

Raises:

Type	Description
`MissingPackageError`	if the `polars` package is not installed.

Example

>>> from coola.utils.imports import check_polars
>>> check_polars()

coola.utils.imports.check_pyarrow ¶

check_pyarrow() -> None

Check if the pyarrow package is installed.

Raises:

Type	Description
`MissingPackageError`	if the `pyarrow` package is not installed.

Example

>>> from coola.utils.imports import check_pyarrow
>>> check_pyarrow()

coola.utils.imports.check_pydantic ¶

check_pydantic() -> None

Check if the pydantic package is installed.

Raises:

Type	Description
`MissingPackageError`	if the `pydantic` package is not installed.

Example

>>> from coola.utils.imports import check_pydantic
>>> check_pydantic()

coola.utils.imports.check_rich ¶

check_rich() -> None

Check if the rich package is installed.

Raises:

Type	Description
`MissingPackageError`	if the `rich` package is not installed.

Example

>>> from coola.utils.imports import check_rich
>>> check_rich()

coola.utils.imports.check_scipy ¶

check_scipy() -> None

Check if the scipy package is installed.

Raises:

Type	Description
`MissingPackageError`	if the `scipy` package is not installed.

Example

>>> from coola.utils.imports import check_scipy
>>> check_scipy()

coola.utils.imports.check_torch ¶

check_torch() -> None

Check if the torch package is installed.

Raises:

Type	Description
`MissingPackageError`	if the `torch` package is not installed.

Example

>>> from coola.utils.imports import check_torch
>>> check_torch()

coola.utils.imports.check_torch_numpy ¶

check_torch_numpy() -> None

Check if the torch and numpy packages are installed and are compatible.

Unlike the other check_<package> functions, this function checks a pair of packages (see is_torch_numpy_available).

Raises:

Type	Description
`MissingPackageError`	if one of the packages is not installed or if they are not compatible.

Example

>>> from coola.utils.imports import check_torch_numpy
>>> check_torch_numpy()

coola.utils.imports.check_xarray ¶

check_xarray() -> None

Check if the xarray package is installed.

Raises:

Type	Description
`MissingPackageError`	if the `xarray` package is not installed.

Example

>>> from coola.utils.imports import check_xarray
>>> check_xarray()

coola.utils.imports.clear_availability_caches ¶

clear_availability_caches() -> None

Clear the cached results of the availability checks.

package_available, module_available and the is_<package>_available functions cache their result, so the availability of a package cannot change during the life of the process. Call this function to force the checks to run again, for example in a test that changes sys.modules or sys.path.

The functions to clear are the ones decorated with availability_cache, so there is no list to maintain.

Example

>>> from coola.utils.imports import clear_availability_caches
>>> clear_availability_caches()

coola.utils.imports.is_colorlog_available ¶

is_colorlog_available() -> bool

Indicate if the colorlog package is installed and can be imported.

Returns:

Type	Description
`bool`	`True` if `colorlog` is available otherwise `False`.

Example

>>> from coola.utils.imports import is_colorlog_available
>>> is_colorlog_available()

coola.utils.imports.is_jax_available ¶

is_jax_available() -> bool

Indicate if the jax package is installed and can be imported.

Returns:

Type	Description
`bool`	`True` if `jax` is available otherwise `False`.

Example

>>> from coola.utils.imports import is_jax_available
>>> is_jax_available()

coola.utils.imports.is_module_imported ¶

is_module_imported(target: object) -> bool

Indicate if the module of a target is already imported.

The module is not imported. A target that is not a string is reported as imported, so coola.factory.import_object raises its usual error.

Parameters:

Name	Type	Description	Default
`target`	`object`	The fully qualified name of the object (for example `"collections.Counter"`), or a module name.	required

Returns:

Type	Description
`bool`	`True` if the module is in `sys.modules` or if `target` is not a string, otherwise `False`.

Example

>>> from coola.utils.imports import is_module_imported
>>> is_module_imported("collections.Counter")
True
>>> is_module_imported("missing_module.Name")
False

coola.utils.imports.is_numpy_available ¶

is_numpy_available() -> bool

Indicate if the numpy package is installed and can be imported.

Returns:

Type	Description
`bool`	`True` if `numpy` is available otherwise `False`.

Example

>>> from coola.utils.imports import is_numpy_available
>>> is_numpy_available()

coola.utils.imports.is_packaging_available ¶

is_packaging_available() -> bool

Indicate if the packaging package is installed and can be imported.

Returns:

Type	Description
`bool`	`True` if `packaging` is available otherwise `False`.

Example

>>> from coola.utils.imports import is_packaging_available
>>> is_packaging_available()

coola.utils.imports.is_pandas_available ¶

is_pandas_available() -> bool

Indicate if the pandas package is installed and can be imported.

Returns:

Type	Description
`bool`	`True` if `pandas` is available otherwise `False`.

Example

>>> from coola.utils.imports import is_pandas_available
>>> is_pandas_available()

coola.utils.imports.is_polars_available ¶

is_polars_available() -> bool

Indicate if the polars package is installed and can be imported.

Returns:

Type	Description
`bool`	`True` if `polars` is available otherwise `False`.

Example

>>> from coola.utils.imports import is_polars_available
>>> is_polars_available()

coola.utils.imports.is_pyarrow_available ¶

is_pyarrow_available() -> bool

Indicate if the pyarrow package is installed and can be imported.

Returns:

Type	Description
`bool`	`True` if `pyarrow` is available otherwise `False`.

Example

>>> from coola.utils.imports import is_pyarrow_available
>>> is_pyarrow_available()

coola.utils.imports.is_pydantic_available ¶

is_pydantic_available() -> bool

Indicate if the pydantic package is installed and can be imported.

Returns:

Type	Description
`bool`	`True` if `pydantic` is available otherwise `False`.

Example

>>> from coola.utils.imports import is_pydantic_available
>>> is_pydantic_available()

coola.utils.imports.is_rich_available ¶

is_rich_available() -> bool

Indicate if the rich package is installed and can be imported.

Returns:

Type	Description
`bool`	`True` if `rich` is available otherwise `False`.

Example

>>> from coola.utils.imports import is_rich_available
>>> is_rich_available()

coola.utils.imports.is_scipy_available ¶

is_scipy_available() -> bool

Indicate if the scipy package is installed and can be imported.

Returns:

Type	Description
`bool`	`True` if `scipy` is available otherwise `False`.

Example

>>> from coola.utils.imports import is_scipy_available
>>> is_scipy_available()

coola.utils.imports.is_torch_available ¶

is_torch_available() -> bool

Indicate if the torch package is installed and can be imported.

Returns:

Type	Description
`bool`	`True` if `torch` is available otherwise `False`.

Example

>>> from coola.utils.imports import is_torch_available
>>> is_torch_available()

coola.utils.imports.is_torch_numpy_available ¶

is_torch_numpy_available() -> bool

Indicate if the torch and numpy packages are installed and are compatible.

The packages are compatible if a torch tensor can be converted to a numpy array. Unlike the other is_<package>_available functions, this function checks a pair of packages. Use it for code that converts between tensors and arrays; if the code only needs one of the two, use is_torch_available or is_numpy_available.

Returns:

Type	Description
`bool`	`True` if both packages are available and compatible, otherwise `False`.

Example

>>> from coola.utils.imports import is_torch_numpy_available
>>> is_torch_numpy_available()

coola.utils.imports.is_xarray_available ¶

is_xarray_available() -> bool

Indicate if the xarray package is installed and can be imported.

Returns:

Type	Description
`bool`	`True` if `xarray` is available otherwise `False`.

Example

>>> from coola.utils.imports import is_xarray_available
>>> is_xarray_available()

coola.utils.imports.lazy_exports ¶

lazy_exports(
    package: str,
    exports: Mapping[str, str],
    namespace: dict[str, Any],
) -> tuple[Callable[[str], Any], Callable[[], list[str]]]

Create the __getattr__ and __dir__ of a package that imports some of its names when they are used.

The names are in the __all__ of the package, so from package import name works as usual, but the module that defines a name is imported the first time the name is used. This avoids importing the optional libraries that a module needs when the package is imported.

Parameters:

Name	Type	Description	Default
`package`	`str`	The name of the package, usually `__name__`.	required
`exports`	`Mapping[str, str]`	The module that defines each lazy name.	required
`namespace`	`dict[str, Any]`	The namespace of the package, usually `globals()`. A name is added to it when it is imported, so the next access does not call `__getattr__`.	required

Returns:

Type	Description
`Callable[[str], Any]`	The `__getattr__` and `__dir__` functions of the package.
`Callable[[], list[str]]`	The returned `__getattr__` raises an `AttributeError` if the
`tuple[Callable[[str], Any], Callable[[], list[str]]]`	name is not in `exports`.

Example

>>> namespace = {}
>>> getattr_, dir_ = lazy_exports("example", {"OrderedDict": "collections"}, namespace)
>>> getattr_("OrderedDict")
<class 'collections.OrderedDict'>
>>> "OrderedDict" in namespace
True

coola.utils.imports.module_available ¶

module_available(name: str) -> bool

Indicate if a module is available or not.

The module is imported to check that it works, so a package that is installed but broken is reported as not available. The result is cached; use clear_availability_caches to reset it.

Parameters:

Name	Type	Description	Default
`name`	`str`	The module name to check.	required

Returns:

Type	Description
`bool`	`True` if the module is available, otherwise `False`.

Example

>>> from coola.utils.imports import module_available
>>> module_available("os")
True
>>> module_available("os.missing")
False
>>> module_available("missing.module")
False

coola.utils.imports.package_available ¶

package_available(name: str) -> bool

Indicate if a package is installed or not.

The package is only looked up (it is not imported), so a package that is installed but broken is reported as available. Use module_available to check that a module can be imported.

Parameters:

Name	Type	Description	Default
`name`	`str`	The package name to check.	required

Returns:

Type	Description
`bool`	`True` if the package is available, otherwise `False`.

Example

>>> from coola.utils.imports import package_available
>>> package_available("os")
True
>>> package_available("missing_package")
False

coola.utils.imports.raise_package_missing_error ¶

raise_package_missing_error(
    package: str, *, install: str | None = None
) -> Never

Raise a MissingPackageError for a missing package.

Parameters:

Name	Type	Description	Default
`package`	`str`	The name of the missing package.	required
`install`	`str \| None`	The name to give to `pip install` when it differs from `package`. By default, it is `package`.	`None`

Raises:

Type	Description
`MissingPackageError`	Always raised to indicate the package is missing.

Example

>>> from coola.utils.imports import raise_package_missing_error
>>> raise_package_missing_error("numpy")  # doctest: +SKIP

coola.utils.integer ¶

Helpers to convert integers to strings.

coola.utils.integer.int_repr ¶

int_repr(value: int) -> str

Return the repr of an integer, in hexadecimal if it has more than 4300 decimal digits.

The limit is fixed, so the result does not depend on sys.set_int_max_str_digits or PYTHONINTMAXSTRDIGITS.

Parameters:

Name	Type	Description	Default
`value`	`int`	The integer to convert.	required

Returns:

Type	Description
`str`	The decimal `repr` of `value`, or `0x...` (`-0x...` for a negative integer) when it has more than 4300 digits.

Example

>>> from coola.utils.integer import int_repr
>>> int_repr(42)
'42'
>>> int_repr(10**5000)[:6]
'0x31e2'

coola.utils.introspection ¶

Utilities for introspecting Python objects and retrieving their fully qualified names.

This module provides functions to retrieve information about Python objects, including their fully qualified names and checking if objects are lambda functions.

coola.utils.introspection.check_not_lambda_function ¶

check_not_lambda_function(obj: Any) -> None

Raise an exception if the object is a lambda function.

Lambda functions cannot be registered because they cannot be reliably serialized or referenced by name.

Parameters:

Name	Type	Description	Default
`obj`	`Any`	The object to check. Can be any Python object.	required

Raises:

Type	Description
`TypeError`	if the object is a lambda function.

Example

>>> from coola.utils.introspection import check_not_lambda_function
>>> check_not_lambda_function(1)

coola.utils.introspection.get_all_child_classes ¶

get_all_child_classes(cls: type) -> set[type]

Get all the child classes (or subclasses) of a given class.

Based on: https://stackoverflow.com/a/3862957

Parameters:

Name	Type	Description	Default
`cls`	`type`	The class whose child classes are to be retrieved.	required

Returns:

Type	Description
`set[type]`	The set of all the child classes of the given class.

Example

>>> from coola.utils.introspection import get_all_child_classes
>>> class Foo:
...     pass
...
>>> get_all_child_classes(Foo)
set()
>>> class Bar(Foo):
...     pass
...
>>> get_all_child_classes(Foo)
{<class '....Bar'>}

coola.utils.introspection.get_qualified_class_name ¶

get_qualified_class_name(cls: type) -> str

Return the qualified name of a class, with its module unless it is a builtin.

The module is also kept for __main__, so the name identifies the class in a namespace shared with the names of the builtins and of other modules.

Parameters:

Name	Type	Description	Default
`cls`	`type`	The class whose name is to be computed.	required

Returns:

Type	Description
`str`	The name in the format `"module.qualname"`, or `"qualname"`
`str`	for a builtin class.

Example

>>> from collections import Counter
>>> from coola.utils.introspection import get_qualified_class_name
>>> get_qualified_class_name(Counter)
'collections.Counter'
>>> get_qualified_class_name(int)
'int'

coola.utils.introspection.get_type_constructor ¶

get_type_constructor(obj: object) -> Callable[..., Any]

Get the type of an object as a generic callable.

Type checkers do not allow calling type(obj) with arbitrary arguments. This helper centralizes the cast needed to rebuild an object of the same type from new data.

Parameters:

Name	Type	Description	Default
`obj`	`object`	The object whose type is returned.	required

Returns:

Type	Description
`Callable[..., Any]`	The type of `obj`, typed as a callable.

Example

>>> from coola.utils.introspection import get_type_constructor
>>> get_type_constructor([1, 2])((3, 4))
[3, 4]

coola.utils.introspection.has_generated_dataclass_eq ¶

has_generated_dataclass_eq(data_type: type) -> bool

Indicate if a type is a dataclass that uses the __eq__ generated by @dataclass.

@dataclass does not overwrite an __eq__ defined in the class body, and the generated one is built with exec so its code object has no source file. A dataclass with eq=False or with its own __eq__ is not matched: its equality is the choice of the user.

Parameters:

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

Returns:

Type	Description
`bool`	`True` if `data_type` is a dataclass with the generated `__eq__`, otherwise `False`.

Example

>>> from dataclasses import dataclass
>>> from coola.utils.introspection import has_generated_dataclass_eq
>>> @dataclass
... class Point:
...     x: int
...
>>> has_generated_dataclass_eq(Point)
True
>>> has_generated_dataclass_eq(int)
False

coola.utils.introspection.is_lambda_function ¶

is_lambda_function(obj: Any) -> bool

Indicate if the object is a lambda function or not.

This function checks whether a given object is a lambda function by examining its type and name. Lambda functions are not allowed in factories because they cannot be reliably serialized or referenced by name. Adapted from https://stackoverflow.com/a/23852434

Parameters:

Name	Type	Description	Default
`obj`	`Any`	The object to check. Can be any Python object.	required

Returns:

Type	Description
`bool`	`True` if the input is a lambda function, otherwise `False`

Example

>>> from coola.utils.introspection import is_lambda_function
>>> is_lambda_function(lambda value: value + 1)
True
>>> def my_function(value: int) -> int:
...     return value + 1
...
>>> is_lambda_function(my_function)
False
>>> is_lambda_function(1)
False

coola.utils.introspection.is_same_type ¶

is_same_type(
    value: object, reference: T
) -> TypeGuard[T]

Indicate if an object has exactly the same type as a reference.

The types must be identical: an instance of a subclass is not considered to have the same type as an instance of its parent class. This is the check used by the equal methods to compare two objects of the same class. When it returns True, type checkers know that value has the type of reference, so its attributes can be accessed.

Parameters:

Name	Type	Description	Default
`value`	`object`	The object to check. It is the first argument because type guards narrow the first argument.	required
`reference`	`T`	The object that gives the expected type.	required

Returns:

Type	Description
`TypeGuard[T]`	`True` if `type(value) is type(reference)`, otherwise `False`.

Example

>>> from coola.utils.introspection import is_same_type
>>> is_same_type(1, 2)
True
>>> is_same_type(1, 2.0)
False
>>> is_same_type(True, 1)
False

coola.utils.introspection.supports_methods ¶

supports_methods(obj: Any, *method_names: str) -> bool

Indicate whether an object has all the given methods and they are callable.

Parameters:

Name	Type	Description	Default
`obj`	`Any`	The object to check.	required
`method_names`	`str`	The names of the methods to check.	`()`

Returns:

Type	Description
`bool`	`True` if `obj` has every method in `method_names` and each is callable, otherwise `False`.

Example

>>> from coola.utils.introspection import supports_methods
>>> supports_methods(1, "bit_length")
True
>>> supports_methods(1, "bit_length", "does_not_exist")
False

coola.utils.logging ¶

Implement logging utilities.

coola.utils.logging.CaptureHandler ¶

Bases: Handler

Collect the records logged by the thread that created the handler.

Records emitted from other threads are ignored by default, so concurrent comparisons do not see each other's records.

Parameters:

Name	Type	Description	Default
`level`	`int`	The minimum level of the records to capture. Defaults to `logging.INFO`.	`INFO`
`all_threads`	`bool`	If `True`, the records of all the threads are captured. Defaults to `False`.	`False`

Attributes:

Name	Type	Description
`records`	`list[LogRecord]`	The captured records, in emission order.

Example

>>> import logging
>>> from coola.utils.logging import CaptureHandler
>>> handler = CaptureHandler()
>>> logger = logging.getLogger("capture_example")
>>> logger.addHandler(handler)
>>> logger.warning("hello")
>>> [record.getMessage() for record in handler.records]
['hello']
>>> logger.removeHandler(handler)

coola.utils.lru ¶

Define a bounded, dict-like LRU (least-recently-used) cache.

A standalone utility for the data structures that need a bounded lookup cache: it caps memory usage while keeping the most recently used entries around. It is not used by the registries of coola (TypeRegistry uses a WeakKeyDictionary as cache). This class is not thread-safe on its own — callers that share an instance across threads are expected to guard access with their own lock.

coola.utils.lru.LRUCache ¶

Bases: Generic[K, V]

A bounded, dict-like cache that evicts the least-recently-used entry once it grows past maxsize.

Both reading (__getitem__/get) and writing (__setitem__) an existing key mark it as most-recently-used. The class supports the common dict-like operations (__contains__, __len__, __iter__, __delitem__, keys/values/items, clear, pop) needed to act as a drop-in cache store.

Warning

The cache is not thread-safe, unlike the registries of coola: even a read (__getitem__, get) modifies the order of the entries. Protect it with a lock if it is shared between threads.

Parameters:

Name	Type	Description	Default
`maxsize`	`int`	The maximum number of entries the cache holds. Must be a positive integer. Once a new entry would push the cache past this size, the least-recently-used entry is evicted.	required

Raises:

Type	Description
`ValueError`	if `maxsize` is not a positive integer.

Example

>>> from coola.utils.lru import LRUCache
>>> cache = LRUCache[str, int](maxsize=2)
>>> cache["a"] = 1
>>> cache["b"] = 2
>>> cache["a"]  # accessing "a" marks it as most-recently-used
1
>>> cache["c"] = 3  # evicts "b", the least-recently-used entry
>>> list(cache)
['a', 'c']

coola.utils.lru.LRUCache.maxsize `property` ¶

maxsize: int

The maximum number of entries the cache holds.

Returns:

Type	Description
`int`	The maximum number of entries.

coola.utils.lru.LRUCache.getitem ¶

__getitem__(key: K) -> V

Get the value associated with key and mark it as most- recently-used.

Parameters:

Name	Type	Description	Default
`key`	`K`	The key to look up.	required

Returns:

Type	Description
`V`	The value associated with `key`.

Raises:

Type	Description
`KeyError`	if `key` is not in the cache.

coola.utils.lru.LRUCache.setitem ¶

__setitem__(key: K, value: V) -> None

Set value for key, marking it as most-recently-used, and evict the least-recently-used entry if the cache is now over capacity.

Parameters:

Name	Type	Description	Default
`key`	`K`	The key to set.	required
`value`	`V`	The value to associate with `key`.	required

coola.utils.lru.LRUCache.clear ¶

clear() -> None

Remove all entries from the cache.

coola.utils.lru.LRUCache.get ¶

get(key: K, default: V | None = None) -> V | None

Get the value associated with key, or default if the key is missing, marking a hit as most-recently-used.

Parameters:

Name	Type	Description	Default
`key`	`K`	The key to look up.	required
`default`	`V \| None`	The value returned if `key` is missing.	`None`

Returns:

Type	Description
`V \| None`	The value associated with `key`, or `default` if `key` is missing.

coola.utils.lru.LRUCache.items ¶

items() -> ItemsView[K, V]

Return a view of the cache's (key, value) pairs, in least- to most-recently-used order.

Returns:

Type	Description
`ItemsView[K, V]`	A view of the `(key, value)` pairs.

coola.utils.lru.LRUCache.keys ¶

keys() -> KeysView[K]

Return a view of the cache's keys, in least- to most- recently-used order.

Returns:

Type	Description
`KeysView[K]`	A view of the keys.

coola.utils.lru.LRUCache.pop ¶

pop(key: K, *args: V) -> V

Remove and return the value associated with key.

Parameters:

Name	Type	Description	Default
`key`	`K`	The key to remove.	required
`*args`	`V`	An optional default value returned instead of raising `KeyError` when `key` is missing, mirroring `dict.pop`.	`()`

Returns:

Type	Description
`V`	The value that was associated with `key`, or the default value if `key` is missing.

Raises:

Type	Description
`KeyError`	if `key` is missing and no default is given.

coola.utils.lru.LRUCache.values ¶

values() -> ValuesView[V]

Return a view of the cache's values, in least- to most- recently-used order.

Returns:

Type	Description
`ValuesView[V]`	A view of the values.

coola.utils.mapping ¶

Contain utility functions for mappings.

coola.utils.mapping.get_first_value ¶

get_first_value(data: Mapping[Any, T]) -> T

Get the first value of a mapping.

Parameters:

Name	Type	Description	Default
`data`	`Mapping[Any, T]`	The input mapping.	required

Returns:

Type	Description
`T`	The first value in the mapping.

Raises:

Type	Description
`ValueError`	if the mapping is empty.

Example

>>> from coola.utils.mapping import get_first_value
>>> get_first_value({"key1": 1, "key2": 2})
1

coola.utils.mapping.sort_by_keys ¶

sort_by_keys(mapping: Mapping[Any, Any]) -> dict[Any, Any]

Sort a dictionary by keys.

If the keys cannot be compared with each other (for example 1 and "a"), they are sorted by type name first, then by key. If two keys of the same type cannot be compared either (for example the tuples (1,) and ("a",)), they are sorted by type name first, then by repr of the key.

Parameters:

Name	Type	Description	Default
`mapping`	`Mapping[Any, Any]`	The dictionary to sort.	required

Returns:

Type	Description
`dict[Any, Any]`	The sorted dictionary.

Example

>>> from coola.utils.mapping import sort_by_keys
>>> sort_by_keys({"dog": 1, "cat": 5, "fish": 2})
{'cat': 5, 'dog': 1, 'fish': 2}

coola.utils.mapping.sort_by_values ¶

sort_by_values(
    mapping: Mapping[Any, Any],
) -> dict[Any, Any]

Sort a dictionary by values.

If the values cannot be compared with each other (for example 1 and "a"), they are sorted as in sort_by_keys.

Parameters:

Name	Type	Description	Default
`mapping`	`Mapping[Any, Any]`	The dictionary to sort.	required

Returns:

Type	Description
`dict[Any, Any]`	The sorted dictionary.

Example

>>> from coola.utils.mapping import sort_by_values
>>> sort_by_values({"dog": 1, "cat": 5, "fish": 2})
{'dog': 1, 'fish': 2, 'cat': 5}

coola.utils.password ¶

Implement utility functions to securely get passwords from users.

coola.utils.password.get_password ¶

get_password(
    username: str | None = None, *, confirm: bool = False
) -> str

Securely prompt for and retrieve a password from the user.

This function uses getpass to securely prompt for password input without echoing characters to the terminal. The password undergoes validation for length constraints before being returned.

Security features

No password echoing to terminal
Length validation (non-empty, max bound)
Optional confirmation to prevent typos
Prevents use in non-interactive contexts
No logging or storage of password values

Parameters:

Name	Type	Description	Default
`username`	`str \| None`	Optional username for prompt clarity.	`None`
`confirm`	`bool`	Require password confirmation if True.	`False`

Returns:

Type	Description
`str`	The password entered by the user (whitespace preserved).

Raises:

Type	Description
`RuntimeError`	If not running in an interactive terminal.
`KeyboardInterrupt`	If input is interrupted.
`EOFError`	If input is interrupted (e.g. end of stream reached).
`ValueError`	If password validation fails.

Note

This function has no doctest example because it requires interactive terminal input, which cannot be simulated in a doctest.

coola.utils.path ¶

Contain path utility functions.

coola.utils.path.add_uuid_suffix ¶

add_uuid_suffix(path: Path) -> Path

Return a new path with a unique UUID suffix added to the file name.

This is typically used to derive a unique staging path from a target path, e.g. to write to a temporary file before atomically renaming it to the target path.

Parameters:

Name	Type	Description	Default
`path`	`Path`	The input path.	required

Returns:

Type	Description
`Path`	A new path with the same parent and extension, and a unique UUID appended to the stem. The extension is the last suffix, or a compound extension such as `.tar.gz`.

Example

>>> import tempfile
>>> from pathlib import Path
>>> from coola.utils.path import add_uuid_suffix
>>> with tempfile.TemporaryDirectory() as tmpdir:
...     path = add_uuid_suffix(Path(tmpdir).joinpath("data.pt"))
...     path
...
PosixPath('/.../data-....pt')

coola.utils.path.find_root_package_parent ¶

find_root_package_parent(
    start_path: str | PathLike[str],
) -> Path

Given a file or directory path, walk upward through directories that contain __init__.py (i.e., are part of a package) and return the parent directory of the outermost (root) package.

Parameters:

Name	Type	Description	Default
`start_path`	`str \| PathLike[str]`	The path to a file or directory inside the package. Can be any path-like object, i.e. a string or an object implementing the `os.PathLike` protocol such as `pathlib.Path`.	required

Returns:

Type	Description
`Path`	The absolute path to the parent directory of the root package. If `start_path` itself is not part of a package (no `__init__.py` in its directory), its own directory is returned.

Example

>>> from coola.utils.path import find_root_package_parent
>>> find_root_package_parent("something")  # doctest: +SKIP
PosixPath('.../my_project')

Given the following directory layout::

my_project/
    pkg/
        __init__.py
        sub/
            __init__.py
            module.py

find_root_package_parent(".../pkg/sub/module.py") returns Path(".../my_project").

coola.utils.path.sanitize_path ¶

sanitize_path(
    path: str | PathLike[str], *, resolve: bool = True
) -> Path

Sanitize the given path.

Parameters:

Name	Type	Description	Default
`path`	`str \| PathLike[str]`	The path to sanitize. This can be any path-like object, i.e. a string or an object implementing the `os.PathLike` protocol such as `pathlib.Path`. `file://` URIs are recognized and decoded, but only if given as a `str` (or a `PathLike` whose `__fspath__` returns the raw URI string). Wrapping a URI in `pathlib.Path` first is not supported: `Path` collapses `"file:///a"` to `"file:/a"` before this function ever sees it, so the URI can no longer be detected and it is treated as a plain relative path.	required
`resolve`	`bool`	If `True` (default), the path is made absolute and the symbolic links are resolved. If `False`, the path is only converted to a `Path` (`~` and `file://` URIs are still expanded): a relative path stays relative and a symbolic link is kept.	`True`

Returns:

Type	Description
`Path`	The sanitized path as a `pathlib.Path` object.

Raises:

Type	Description
`ValueError`	if a `file://` URI has a host other than `localhost` (for example `file://server/share/x`), or if the home directory of `~user` cannot be found.

Example

>>> from pathlib import Path
>>> from coola.utils.path import sanitize_path
>>> sanitize_path("something")
PosixPath('.../something')
>>> sanitize_path("")
PosixPath('...')
>>> sanitize_path(Path("something"))
PosixPath('.../something')
>>> sanitize_path(Path("something/./../"))
PosixPath('...')
>>> sanitize_path("something", resolve=False)
PosixPath('something')

coola.utils.recursion ¶

Define a helper to report an interpreter stack overflow.

coola.utils.recursion.raise_recursion_limit_error ¶

raise_recursion_limit_error(
    action: str, exc: RecursionError
) -> NoReturn

Raise the error that explains a Python recursion limit overflow.

The recursive functions of coola use several interpreter frames per nesting level, so the interpreter limit (sys.getrecursionlimit()) is reached before a structure of a thousand levels is done. This function gives them one actionable message.

Parameters:

Name	Type	Description	Default
`action`	`str`	What was done to the structure, in the present participle (for example `"comparing"`).	required
`exc`	`RecursionError`	The `RecursionError` of the interpreter, which is the cause of the raised error.	required

Raises:

Type	Description
`RecursionError`	Always, with a message that explains how to fix the error.

Example

>>> from coola.utils.recursion import raise_recursion_limit_error
>>> try:
...     raise_recursion_limit_error("summarizing", RecursionError())
... except RecursionError as exc:
...     print(exc)
...
The Python recursion limit was exceeded while summarizing deeply nested objects. Each nesting level uses several interpreter frames; reduce the nesting or increase the limit with sys.setrecursionlimit().

coola.utils.safe ¶

Conversions of objects to strings that never raise.

These helpers are meant for error messages, logs and reports, where a broken __str__ or __repr__ (or an int above the string conversion limit) must not hide the information being reported.

coola.utils.safe.safe_repr ¶

safe_repr(obj: object) -> str

Return repr(obj), or a placeholder if the conversion fails.

Parameters:

Name	Type	Description	Default
`obj`	`object`	The object to convert.	required

Returns:

Type	Description
`str`	The representation of the object, or `<ClassName object at 0x...>` if `repr` raised an exception.

Example

>>> from coola.utils.safe import safe_repr
>>> safe_repr("abc")
"'abc'"
>>> class Bad:
...     def __repr__(self):
...         raise RuntimeError
...
>>> safe_repr(Bad()).startswith("<Bad object at 0x")
True

coola.utils.safe.safe_str ¶

safe_str(obj: object) -> str

Return str(obj), or a placeholder if the conversion fails.

Parameters:

Name	Type	Description	Default
`obj`	`object`	The object to convert.	required

Returns:

Type	Description
`str`	The string representation of the object, or `<ClassName object at 0x...>` if `str` raised an exception.

Example

>>> from coola.utils.safe import safe_str
>>> safe_str(42)
'42'
>>> class Bad:
...     def __str__(self):
...         raise RuntimeError
...
>>> safe_str(Bad()).startswith("<Bad object at 0x")
True

coola.utils.selection ¶

Contain functions to select items of a collection.

coola.utils.selection.smallest_items ¶

smallest_items(
    data: Collection[Any], *, limit: int | None = None
) -> list[Any]

Return the smallest items in order.

The items are selected in one pass without sorting all of them. If the items cannot be compared (for example an int and a str), they are sorted with the fallbacks of sort_items, so the result does not depend on the iteration order of data.

Parameters:

Name	Type	Description	Default
`data`	`Collection[Any]`	The items, for example a set. It is a collection (not an iterator) because it is iterated twice when the items cannot be compared.	required
`limit`	`int \| None`	The maximum number of items to return. `None` returns all the items.	`None`

Returns:

Type	Description
`list[Any]`	The `limit` smallest items in order.

Raises:

Type	Description
`ValueError`	If `limit` is negative.

Example

>>> from coola.utils.selection import smallest_items
>>> smallest_items({3, 1, 2})
[1, 2, 3]
>>> smallest_items({3, 1, 2}, limit=2)
[1, 2]
>>> smallest_items([1, "a"], limit=1)
[1]
>>> smallest_items({1, "a"})
[1, 'a']

coola.utils.selection.sort_items ¶

sort_items(
    data: Iterable[Any],
    *,
    key: Callable[[Any], Any] | None = None,
) -> list[Any]

Sort items that may not be comparable with each other.

The items are sorted by key(item) (or by the item if key is None). If the sorted values cannot be compared with each other (for example 1 and "a"), they are sorted by type name first, then by value. If two values of the same type cannot be compared either (for example the tuples (1,) and ("a",)), they are sorted by type name first, then by repr of the value. The result does not depend on the order of data, so it can be used for a set.

Parameters:

Name	Type	Description	Default
`data`	`Iterable[Any]`	The items to sort.	required
`key`	`Callable[[Any], Any] \| None`	A function that returns the value to sort by, or `None` to sort the items themselves.	`None`

Returns:

Type	Description
`list[Any]`	The sorted items.

Example

>>> from coola.utils.selection import sort_items
>>> sort_items({3, 1, 2})
[1, 2, 3]
>>> sort_items([2, "b", 1, "a"])
[1, 2, 'a', 'b']
>>> sort_items([("b", 1), ("a", 2)], key=lambda x: x[0])
[('a', 2), ('b', 1)]

coola.utils.singleton ¶

Define a thread-safe lazy singleton helper.

Several modules expose a get_default_registry() function that builds and caches a registry on first use. Building it eagerly at import time is not an option here because some registries are constructed via imports that are deliberately deferred to avoid circular imports between coola.equality and coola.registry. LazySingleton keeps the construction lazy while making the first-call race thread-safe via double-checked locking.

coola.utils.singleton.LazySingleton ¶

Bases: Generic[T]

Thread-safe holder that builds a value lazily on first access and caches it.

Parameters:

Name	Type	Description	Default
`factory`	`Callable[[], T]`	Callable used to build the value on first access. It is called at most once, even if multiple threads call `get` concurrently before the value exists.	required

Raises:

Type	Description
`RuntimeError`	If `get` is called by the factory itself (on the same thread) while the value is being built. The value does not exist yet, so waiting for it would block forever. For a default registry, this happens when a plugin calls a function that uses the registry that is being built; the plugin should use the registry it receives instead.

Example

>>> from coola.utils.singleton import LazySingleton
>>> counter = {"calls": 0}
>>> def build() -> int:
...     counter["calls"] += 1
...     return 42
...
>>> singleton = LazySingleton(build)
>>> singleton.get()
42
>>> singleton.get()
42
>>> counter["calls"]
1

coola.utils.singleton.LazySingleton.get ¶

get() -> T

Return the cached value, building it on the first call.

Returns:

Type	Description
`T`	The singleton value.

Raises:

Type	Description
`RuntimeError`	If the value is requested again by the thread that is building it (the factory requests the singleton it builds), which would otherwise wait forever.

coola.utils.singleton.LazySingleton.is_built ¶

is_built() -> bool

Indicate if the value has been built.

Returns:

Type	Description
`bool`	`True` if the value has been built, otherwise `False`.

coola.utils.singleton.LazySingleton.reset ¶

reset() -> None

Forget the cached value, so the next get call builds a new one.

Raises:

Type	Description
`RuntimeError`	If `reset` is called by the thread that is building the value (the factory resets the singleton it builds), which would otherwise wait forever.

coola.utils.singleton.LazySingleton.set ¶

set(value: T) -> None

Replace the cached value.

A reference to the previous value is not changed. If the value was not built yet, the factory is not needed anymore.

Parameters:

Name	Type	Description	Default
`value`	`T`	The new value.	required

Raises:

Type	Description
`RuntimeError`	If `set` is called by the thread that is building the value (the factory sets the singleton it builds), which would otherwise wait forever.

Example

>>> from coola.utils.singleton import LazySingleton
>>> singleton = LazySingleton(lambda: 1)
>>> singleton.set(2)
>>> singleton.get()
2

coola.utils.stats ¶

Implement some utility functions to compute statistics.

coola.utils.stats.quantile ¶

quantile(
    values: Sequence[float | int],
    quantiles: Sequence[float],
) -> list[float]

Compute the quantiles with the linear method.

Parameters:

Name	Type	Description	Default
`values`	`Sequence[float \| int]`	The values.	required
`quantiles`	`Sequence[float]`	The quantile values in the range `[0, 1]`.	required

Returns:

Type	Description
`list[float]`	The quantiles.

Raises:

Type	Description
`ValueError`	If a quantile is not in the range `[0, 1]`, or if `values` is empty and `quantiles` is not.

Notes

If values contains a NaN, every quantile is NaN, like with NumPy: the sort order of the NaN is undefined, so the quantiles would depend on the position of the NaN.

Example

>>> from coola.utils.stats import quantile
>>> quantile([0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10], (0.2, 0.5, 0.9))
[2.0, 5.0, 9.0]
>>> quantile([1.0, float("nan")], (0.5,))
[nan]

coola.utils.tensor ¶

Implement some utility functions for torch.Tensors.

coola.utils.tensor.get_available_devices ¶

get_available_devices() -> tuple[str, ...]

Get the available PyTorch devices on the machine.

The result is cached until clear_availability_caches is called.

Returns:

Type	Description
`tuple[str, ...]`	The available devices: `"cpu"`, `"cuda:<i>"` for each CUDA device and `"mps:0"` if MPS is available.

Example

>>> from coola.utils.tensor import get_available_devices
>>> get_available_devices()
('cpu'...)

coola.utils.tensor.is_cuda_available ¶

is_cuda_available() -> bool

Indicate if CUDA is currently available.

Returns:

Type	Description
`bool`	A boolean indicating if CUDA is currently available.

Example

>>> from coola.utils.tensor import is_cuda_available
>>> is_cuda_available()

coola.utils.tensor.is_mps_available ¶

is_mps_available() -> bool

Indicate if MPS is currently available.

It does not allocate a tensor on the device.

Returns:

Type	Description
`bool`	A boolean indicating if MPS is currently available.

Example

>>> from coola.utils.tensor import is_mps_available
>>> is_mps_available()

coola.utils.tensor.sparse_components ¶

sparse_components(tensor: Tensor) -> tuple[Tensor, ...]

Return the strided tensors that store a sparse tensor.

The positions are returned first, then the values. A COO tensor is coalesced first, so two COO tensors that hold the same values have the same components. No dense copy is made.

Parameters:

Name	Type	Description	Default
`tensor`	`Tensor`	The sparse tensor (COO, CSR, CSC, BSR or BSC layout).	required

Returns:

Type	Description
`tuple[Tensor, ...]`	The positions (one or two tensors) followed by the values.

Raises:

Type	Description
`TypeError`	If the layout of the tensor is not supported.

Example

>>> import torch
>>> from coola.utils.tensor import sparse_components
>>> indices, values = sparse_components(torch.eye(2).to_sparse())
>>> indices
tensor([[0, 1],
        [0, 1]])
>>> values
tensor([1., 1.])

coola.utils.tensor.to_tensor ¶

to_tensor(
    data: Sequence[int | float] | Tensor | ndarray,
) -> Tensor

Convert the input to a torch.Tensor.

Parameters:

Name	Type	Description	Default
`data`	`Sequence[int \| float] \| Tensor \| ndarray`	The data to convert to a tensor.	required

Returns:

Type	Description
`Tensor`	A tensor. A tensor is returned as is, and the tensor of a NumPy array may share its memory with the array; a list is copied.

Raises:

Type	Description
`MissingPackageError`	if `torch` is not installed.

Example

>>> from coola.utils.tensor import to_tensor
>>> x = to_tensor([1, 2, 3, 4, 5])
>>> x
tensor([1, 2, 3, 4, 5])
>>> import numpy as np
>>> x = to_tensor(np.array([1, 2, 3, 4, 5]))
>>> x
tensor([1, 2, 3, 4, 5])

coola.utils.timing ¶

Contain utility functions to measure time.

coola.utils.timing.TimingResult ¶

Hold the result of a timed block.

Parameters:

Name	Type	Description	Default
`started_at`	`float \| None`	The time when the block started, or `None` if it has not yet started.	`None`
`finished_at`	`float \| None`	The time when the block finished, or `None` if it has not yet finished.	`None`

coola.utils.timing.TimingResult.elapsed `property` ¶

elapsed: float | None

The elapsed time in seconds, or None if the block has not yet completed.

coola.utils.timing.timeblock ¶

timeblock(
    message: str = "Total time: {time}",
) -> Generator[TimingResult]

Implement a context manager to measure the execution time of a block of code.

Parameters:

Name	Type	Description	Default
`message`	`str`	The message displayed when the time is logged. It must contain the `{time}` placeholder.	`'Total time: {time}'`

Yields:

Type	Description
`Generator[TimingResult]`	A `TimingResult` whose `elapsed` attribute is set when the block exits.

Raises:

Type	Description
`ValueError`	If `message` does not contain `{time}`.

Example

>>> from coola.utils.timing import timeblock
>>> with timeblock():
...     x = [1, 2, 3]
...
>>> with timeblock("Training: {time}"):
...     y = [1, 2, 3]
...
>>> with timeblock() as t:
...     z = [1, 2, 3]
...
>>> t.elapsed

coola.utils.version ¶

Contain functions to manage package versions.

coola.utils.version.compare_version ¶

compare_version(
    package: str,
    op: Callable[[Version, Version], bool],
    version: str,
) -> bool

Compare a package version to a given version.

Parameters:

Name	Type	Description	Default
`package`	`str`	The package to check.	required
`op`	`Callable[[Version, Version], bool]`	The comparison operator.	required
`version`	`str`	The version to compare with.	required

Returns:

Type	Description
`bool`	The comparison status.

Example

>>> import operator
>>> from coola.utils.version import compare_version
>>> compare_version("pytest", op=operator.ge, version="7.3.0")
True

coola.utils.version.get_package_version ¶

get_package_version(package: str) -> Version | None

Get the package version.

Parameters:

Name	Type	Description	Default
`package`	`str`	The package name.	required

Returns:

Type	Description
`Version \| None`	The package version, or `None` if the package is not installed or if its version is not a valid PEP 440 version. The result is cached; use `clear_availability_caches` to reset it.

Example

>>> from coola.utils.version import get_package_version
>>> get_package_version("pytest")
<Version('...')>

Utils

coola.utils ¶

coola.utils.array ¶

coola.utils.array.to_array ¶

coola.utils.batching ¶

coola.utils.batching.batchify ¶

coola.utils.bloom_filter ¶

coola.utils.bloom_filter.BloomFilter ¶

coola.utils.bloom_filter.BloomFilter.__contains__ ¶

coola.utils.bloom_filter.BloomFilter.add ¶

coola.utils.bloom_filter.BloomFilter.add_and_check ¶

coola.utils.bloom_filter.BloomFilter.fill_ratio ¶

coola.utils.date ¶

coola.utils.date.get_today_date ¶

coola.utils.date.to_date ¶

coola.utils.date.to_datetime ¶

coola.utils.env_vars ¶

coola.utils.env_vars.check_env_vars ¶

coola.utils.env_vars.get_required_env_var ¶

coola.utils.env_vars.temp_env_vars ¶

coola.utils.fallback ¶

coola.utils.file_size ¶

coola.utils.file_size.get_file_size ¶

coola.utils.fork ¶

coola.utils.fork.ForkAware ¶

coola.utils.fork.track_fork ¶

coola.utils.git ¶

coola.utils.git.get_current_branch ¶

coola.utils.git.get_current_commit_hash ¶

coola.utils.git.is_git_repo ¶

coola.utils.git.run_git_command ¶

coola.utils.imports ¶

coola.utils.imports.MissingPackageError ¶

coola.utils.imports.availability_cache ¶

coola.utils.imports.check_colorlog ¶

coola.utils.imports.check_jax ¶

coola.utils.imports.check_numpy ¶

coola.utils.imports.check_package ¶

coola.utils.imports.check_packaging ¶

coola.utils.imports.check_pandas ¶

coola.utils.imports.check_polars ¶

coola.utils.imports.check_pyarrow ¶

coola.utils.imports.check_pydantic ¶

coola.utils.imports.check_rich ¶

coola.utils.imports.check_scipy ¶

coola.utils.imports.check_torch ¶

coola.utils.imports.check_torch_numpy ¶

coola.utils.imports.check_xarray ¶

coola.utils.imports.clear_availability_caches ¶

coola.utils.imports.is_colorlog_available ¶

coola.utils.imports.is_jax_available ¶

coola.utils.imports.is_module_imported ¶

coola.utils.imports.is_numpy_available ¶

coola.utils.imports.is_packaging_available ¶

coola.utils.imports.is_pandas_available ¶

coola.utils.imports.is_polars_available ¶

coola.utils.imports.is_pyarrow_available ¶

coola.utils.imports.is_pydantic_available ¶

coola.utils.imports.is_rich_available ¶

coola.utils.imports.is_scipy_available ¶

coola.utils.imports.is_torch_available ¶

coola.utils.imports.is_torch_numpy_available ¶

coola.utils.imports.is_xarray_available ¶

coola.utils.imports.lazy_exports ¶

coola.utils.imports.module_available ¶

coola.utils.imports.package_available ¶

coola.utils.imports.raise_package_missing_error ¶

coola.utils.integer ¶

coola.utils.integer.int_repr ¶

coola.utils.introspection ¶

coola.utils.introspection.check_not_lambda_function ¶

coola.utils.introspection.get_all_child_classes ¶

coola.utils.introspection.get_qualified_class_name ¶

coola.utils.introspection.get_type_constructor ¶

coola.utils.introspection.has_generated_dataclass_eq ¶

coola.utils.introspection.is_lambda_function ¶

coola.utils.introspection.is_same_type ¶

coola.utils.introspection.supports_methods ¶

coola.utils.logging ¶

coola.utils.logging.CaptureHandler ¶

coola.utils.bloom_filter.BloomFilter.contains ¶

coola.utils.lru.LRUCache.maxsize `property` ¶

coola.utils.lru.LRUCache.getitem ¶

coola.utils.lru.LRUCache.setitem ¶

coola.utils.timing.TimingResult.elapsed `property` ¶