Skip to content

Io

coola.io

Contain data loaders and savers.

coola.io.BaseFileSaver

Bases: BaseSaver[T]

Define the base class to implement a file saver.

Example
>>> import tempfile
>>> from pathlib import Path
>>> from coola.io import JsonSaver, JsonLoader
>>> with tempfile.TemporaryDirectory() as tmpdir:
...     path = Path(tmpdir).joinpath("data.json")
...     JsonSaver().save({"key1": [1, 2, 3], "key2": "abc"}, path)
...     data = JsonLoader().load(path)
...     data
...
{'key1': [1, 2, 3], 'key2': 'abc'}

coola.io.BaseFileSaver.save

save(
    to_save: T, path: Path, *, exist_ok: bool = False
) -> None

Save the data into the given path.

Parameters:

Name Type Description Default
to_save T

The data to save. The data should be compatible with the saving engine.

required
path Path

The path where to save the data.

required
exist_ok bool

If exist_ok is False (the default), FileExistsError is raised if the target file already exists. If exist_ok is True, FileExistsError will not be raised unless the given path already exists in the file system and is not a file.

Note: the exist_ok=False guarantee is protected against a concurrent creation of path between the initial existence check and the commit step (the commit uses os.link which atomically fails with FileExistsError in that case).

Note: concurrent save calls targeting the same path, whether from within this process (an in-memory lock) or from other processes (a path.lock lock file created next to path), are serialized so they cannot interleave their write/commit steps and corrupt or partially overwrite each other's data. This holds for exist_ok=True too: two concurrent calls with exist_ok=True can no longer race each other's tmp_path.replace(path) commit; each call's replace happens fully before the next one starts. Which call's data ends up on disk is still whichever one wins the lock last (no ordering is guaranteed beyond "no interleaving"), and the lock file is removed once the holder releases it, so it does not accumulate on disk.

False

Raises:

Type Description
FileExistsError

if the file already exists.

Example
>>> import tempfile
>>> from pathlib import Path
>>> from coola.io import JsonSaver, JsonLoader
>>> with tempfile.TemporaryDirectory() as tmpdir:
...     path = Path(tmpdir).joinpath("data.json")
...     JsonSaver().save({"key1": [1, 2, 3], "key2": "abc"}, path)
...     data = JsonLoader().load(path)
...     data
...
{'key1': [1, 2, 3], 'key2': 'abc'}

coola.io.BaseLoader

Bases: ABC, Generic[T]

Define the base class to implement a data loader.

Example
>>> import tempfile
>>> from pathlib import Path
>>> from coola.io import save_json, JsonLoader
>>> with tempfile.TemporaryDirectory() as tmpdir:
...     path = Path(tmpdir).joinpath("data.json")
...     save_json({"key1": [1, 2, 3], "key2": "abc"}, path)
...     data = JsonLoader().load(path)
...     data
...
{'key1': [1, 2, 3], 'key2': 'abc'}

coola.io.BaseLoader.equal abstractmethod

equal(other: Any, equal_nan: bool = False) -> bool

Indicate if two objects are equal or not.

Parameters:

Name Type Description Default
other Any

The object to compare with.

required
equal_nan bool

If True, then two NaNs will be considered equal.

False

Returns:

Type Description
bool

True if the two objects are equal, otherwise False.

Example
>>> from coola.io import JsonLoader, TextLoader
>>> JsonLoader().equal(JsonLoader())
True
>>> JsonLoader().equal(TextLoader())
False

coola.io.BaseLoader.load abstractmethod

load(path: Path) -> T

Load the data from the given path.

Parameters:

Name Type Description Default
path Path

The path with the data to load.

required

Returns:

Type Description
T

The data

Example
>>> import tempfile
>>> from pathlib import Path
>>> from coola.io import save_json, JsonLoader
>>> with tempfile.TemporaryDirectory() as tmpdir:
...     path = Path(tmpdir).joinpath("data.json")
...     save_json({"key1": [1, 2, 3], "key2": "abc"}, path)
...     data = JsonLoader().load(path)
...     data
...
{'key1': [1, 2, 3], 'key2': 'abc'}

coola.io.BaseSaver

Bases: ABC, Generic[T]

Define the base class to implement a data saver.

Example
>>> import tempfile
>>> from pathlib import Path
>>> from coola.io import JsonSaver, JsonLoader
>>> with tempfile.TemporaryDirectory() as tmpdir:
...     path = Path(tmpdir).joinpath("data.json")
...     JsonSaver().save({"key1": [1, 2, 3], "key2": "abc"}, path)
...     data = JsonLoader().load(path)
...     data
...
{'key1': [1, 2, 3], 'key2': 'abc'}

coola.io.BaseSaver.equal abstractmethod

equal(other: Any, equal_nan: bool = False) -> bool

Indicate if two objects are equal or not.

Parameters:

Name Type Description Default
other Any

The object to compare with.

required
equal_nan bool

If True, then two NaNs will be considered equal.

False

Returns:

Type Description
bool

True if the two objects are equal, otherwise False.

Example
>>> from coola.io import JsonSaver, TextSaver
>>> JsonSaver().equal(JsonSaver())
True
>>> JsonSaver().equal(TextSaver())
False

coola.io.BaseSaver.save abstractmethod

save(
    to_save: T, path: Path, *, exist_ok: bool = False
) -> None

Save the data into the given path.

Parameters:

Name Type Description Default
to_save T

The data to save. The data should be compatible with the saving engine.

required
path Path

The path where to save the data.

required
exist_ok bool

If exist_ok is False (the default), an exception is raised if the target path already exists.

False
Example
>>> import tempfile
>>> from pathlib import Path
>>> from coola.io import JsonSaver, JsonLoader
>>> with tempfile.TemporaryDirectory() as tmpdir:
...     path = Path(tmpdir).joinpath("data.json")
...     JsonSaver().save({"key1": [1, 2, 3], "key2": "abc"}, path)
...     data = JsonLoader().load(path)
...     data
...
{'key1': [1, 2, 3], 'key2': 'abc'}

coola.io.JsonLoader

Bases: InlineDisplayMixin, BaseLoader[T]

Implement a data loader to load data in a JSON file.

Example
>>> import tempfile
>>> from pathlib import Path
>>> from coola.io import save_json, JsonLoader
>>> with tempfile.TemporaryDirectory() as tmpdir:
...     path = Path(tmpdir).joinpath("data.json")
...     save_json({"key1": [1, 2, 3], "key2": "abc"}, path)
...     data = JsonLoader().load(path)
...     data
...
{'key1': [1, 2, 3], 'key2': 'abc'}

coola.io.JsonSaver

Bases: InlineDisplayMixin, BaseFileSaver[T]

Implement a file saver to save data with a JSON file.

Parameters:

Name Type Description Default
encoding str

The file encoding to use when writing the file. Defaults to "utf-8".

DEFAULT_ENCODING
**kwargs Any

Additional arguments passed to json.dump, e.g. indent or sort_keys.

{}
Example
>>> import tempfile
>>> from pathlib import Path
>>> from coola.io import JsonSaver, JsonLoader
>>> with tempfile.TemporaryDirectory() as tmpdir:
...     path = Path(tmpdir).joinpath("data.json")
...     JsonSaver().save({"key1": [1, 2, 3], "key2": "abc"}, path)
...     data = JsonLoader().load(path)
...     data
...
{'key1': [1, 2, 3], 'key2': 'abc'}

coola.io.PickleLoader

Bases: InlineDisplayMixin, BaseLoader[T]

Implement a data loader to load data in a pickle file.

Warning

load calls pickle.load on the given file. Unpickling can execute arbitrary code as a side effect of deserialization. Only load pickle files that come from a trusted source; never load a pickle file that comes from an untrusted or unauthenticated source.

Example
>>> import tempfile
>>> from pathlib import Path
>>> from coola.io import save_pickle, PickleLoader
>>> with tempfile.TemporaryDirectory() as tmpdir:
...     path = Path(tmpdir).joinpath("data.pkl")
...     save_pickle({"key1": [1, 2, 3], "key2": "abc"}, path)
...     data = PickleLoader().load(path)
...     data
...
{'key1': [1, 2, 3], 'key2': 'abc'}

coola.io.PickleSaver

Bases: InlineDisplayMixin, BaseFileSaver[T]

Implement a file saver to save data with a pickle file.

Parameters:

Name Type Description Default
**kwargs Any

Additional arguments passed to pickle.dump.

{}
Example
>>> import tempfile
>>> from pathlib import Path
>>> from coola.io import PickleSaver, PickleLoader
>>> with tempfile.TemporaryDirectory() as tmpdir:
...     path = Path(tmpdir).joinpath("data.pkl")
...     PickleSaver().save({"key1": [1, 2, 3], "key2": "abc"}, path)
...     data = PickleLoader().load(path)
...     data
...
{'key1': [1, 2, 3], 'key2': 'abc'}

coola.io.TextLoader

Bases: InlineDisplayMixin, BaseLoader[str]

Implement a data loader to load data from a text file.

Parameters:

Name Type Description Default
encoding str

The file encoding to use when reading the file. Defaults to "utf-8".

DEFAULT_ENCODING
Example
>>> import tempfile
>>> from pathlib import Path
>>> from coola.io import save_text, TextLoader
>>> with tempfile.TemporaryDirectory() as tmpdir:
...     path = Path(tmpdir).joinpath("data.txt")
...     save_text("hello", path)
...     data = TextLoader().load(path)
...     data
...
'hello'

coola.io.TextSaver

Bases: InlineDisplayMixin, BaseFileSaver[str]

Implement a file saver to save data to a text file.

Parameters:

Name Type Description Default
encoding str

The file encoding to use when writing the file. Defaults to "utf-8".

DEFAULT_ENCODING
Note

If the data to save is not a string, it is converted to a string before being saved by using str.

Example
>>> import tempfile
>>> from pathlib import Path
>>> from coola.io import TextSaver, TextLoader
>>> with tempfile.TemporaryDirectory() as tmpdir:
...     path = Path(tmpdir).joinpath("data.txt")
...     TextSaver().save("hello", path)
...     data = TextLoader().load(path)
...     data
...
'hello'

coola.io.TorchLoader

Bases: InlineDisplayMixin, BaseLoader[T]

Implement a data loader to load data in a PyTorch file.

Parameters:

Name Type Description Default
**kwargs Any

Additional arguments passed to torch.load.

{}
Example
>>> import tempfile
>>> from pathlib import Path
>>> from coola.io import save_torch, TorchLoader
>>> with tempfile.TemporaryDirectory() as tmpdir:
...     path = Path(tmpdir).joinpath("data.pt")
...     save_torch({"key1": [1, 2, 3], "key2": "abc"}, path)
...     data = TorchLoader().load(path)
...     data
...
{'key1': [1, 2, 3], 'key2': 'abc'}

coola.io.TorchSaver

Bases: InlineDisplayMixin, BaseFileSaver[T]

Implement a file saver to save data with a PyTorch file.

Parameters:

Name Type Description Default
**kwargs Any

Additional arguments passed to torch.save.

{}
Example
>>> import tempfile
>>> from pathlib import Path
>>> from coola.io import TorchSaver, TorchLoader
>>> with tempfile.TemporaryDirectory() as tmpdir:
...     path = Path(tmpdir).joinpath("data.pt")
...     TorchSaver().save({"key1": [1, 2, 3], "key2": "abc"}, path)
...     data = TorchLoader().load(path)
...     data
...
{'key1': [1, 2, 3], 'key2': 'abc'}

coola.io.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.

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

coola.io.is_loader_config

is_loader_config(config: dict[Any, Any]) -> bool

Indicate if the input configuration is a configuration for a BaseLoader.

This function only checks if the value of the key _target_ is valid. It does not check the other values. If _target_ indicates a function, the returned type hint is used to check the class.

Parameters:

Name Type Description Default
config dict[Any, Any]

The configuration to check.

required

Returns:

Type Description
bool

True if the input configuration is a configuration for a BaseLoader object.

Example
>>> from coola.io import is_loader_config
>>> is_loader_config({"_target_": "coola.io.JsonLoader"})
True

coola.io.is_saver_config

is_saver_config(config: dict[Any, Any]) -> bool

Indicate if the input configuration is a configuration for a BaseSaver.

This function only checks if the value of the key _target_ is valid. It does not check the other values. If _target_ indicates a function, the returned type hint is used to check the class.

Parameters:

Name Type Description Default
config dict[Any, Any]

The configuration to check.

required

Returns:

Type Description
bool

True if the input configuration is a configuration for a BaseSaver object.

Example
>>> from coola.io import is_saver_config
>>> is_saver_config({"_target_": "coola.io.JsonSaver"})
True

coola.io.load_json

load_json(path: Path) -> Any

Load the data from a given JSON file.

Parameters:

Name Type Description Default
path Path

The path to the JSON file.

required

Returns:

Type Description
Any

The data from the JSON file.

Example
>>> import tempfile
>>> from pathlib import Path
>>> from coola.io import save_json, load_json
>>> with tempfile.TemporaryDirectory() as tmpdir:
...     path = Path(tmpdir).joinpath("data.json")
...     save_json({"key1": [1, 2, 3], "key2": "abc"}, path)
...     data = load_json(path)
...     data
...
{'key1': [1, 2, 3], 'key2': 'abc'}

coola.io.load_pickle

load_pickle(path: Path) -> Any

Load the data from a given pickle file.

Parameters:

Name Type Description Default
path Path

The path to the pickle file.

required

Returns:

Type Description
Any

The data from the pickle file.

Warning

This function unpickles the file's content, which can execute arbitrary code as a side effect of deserialization. Only load pickle files that come from a trusted source; never load a pickle file that comes from an untrusted or unauthenticated source.

Example
>>> import tempfile
>>> from pathlib import Path
>>> from coola.io import save_pickle, load_pickle
>>> with tempfile.TemporaryDirectory() as tmpdir:
...     path = Path(tmpdir).joinpath("data.pkl")
...     save_pickle({"key1": [1, 2, 3], "key2": "abc"}, path)
...     data = load_pickle(path)
...     data
...
{'key1': [1, 2, 3], 'key2': 'abc'}

coola.io.load_text

load_text(
    path: Path, encoding: str = DEFAULT_ENCODING
) -> str

Load the data from a given text file.

Parameters:

Name Type Description Default
path Path

The path to the text file.

required
encoding str

The file encoding to use when reading the file. Defaults to "utf-8".

DEFAULT_ENCODING

Returns:

Type Description
str

The text content of the file as a string.

Example
>>> import tempfile
>>> from pathlib import Path
>>> from coola.io import save_text, load_text
>>> with tempfile.TemporaryDirectory() as tmpdir:
...     path = Path(tmpdir).joinpath("data.txt")
...     save_text("hello", path)
...     data = load_text(path)
...     data
...
'hello'

coola.io.load_torch

load_torch(path: Path, **kwargs: Any) -> Any

Load the data from a given PyTorch file.

Parameters:

Name Type Description Default
path Path

The path to the PyTorch file.

required
**kwargs Any

Additional arguments passed to torch.load.

{}

Returns:

Type Description
Any

The data from the PyTorch file.

Example
>>> import tempfile
>>> from pathlib import Path
>>> from coola.io import save_torch, load_torch
>>> with tempfile.TemporaryDirectory() as tmpdir:
...     path = Path(tmpdir).joinpath("data.pt")
...     save_torch({"key1": [1, 2, 3], "key2": "abc"}, path)
...     data = load_torch(path)
...     data
...
{'key1': [1, 2, 3], 'key2': 'abc'}

coola.io.resolve_loader

resolve_loader(
    loader: BaseLoader[T] | dict[Any, Any],
) -> BaseLoader[T]

Set up a data loader.

The data loader is instantiated from its configuration by using the BaseLoader factory function.

Parameters:

Name Type Description Default
loader BaseLoader[T] | dict[Any, Any]

The data loader or its configuration.

required

Returns:

Type Description
BaseLoader[T]

The instantiated data loader.

Security

When loader is a configuration :class:dict, its "_target_" value is imported and instantiated with no allowlist (see :func:coola.factory.factory). Only call this function with a configuration that comes from a trusted source; never resolve a loader configuration derived from untrusted input, as this is a remote-code-execution vector.

Example
>>> from coola.io import resolve_loader
>>> loader = resolve_loader({"_target_": "coola.io.JsonLoader"})
>>> loader
JsonLoader()

coola.io.resolve_saver

resolve_saver(
    saver: BaseSaver[T] | dict[Any, Any],
) -> BaseSaver[T]

Set up a data saver.

The data saver is instantiated from its configuration by using the BaseSaver factory function.

Parameters:

Name Type Description Default
saver BaseSaver[T] | dict[Any, Any]

The data saver or its configuration.

required

Returns:

Type Description
BaseSaver[T]

The instantiated data saver.

Security

When saver is a configuration :class:dict, its "_target_" value is imported and instantiated with no allowlist (see :func:coola.factory.factory). Only call this function with a configuration that comes from a trusted source; never resolve a saver configuration derived from untrusted input, as this is a remote-code-execution vector.

Example
>>> from coola.io import resolve_saver
>>> saver = resolve_saver({"_target_": "coola.io.JsonSaver"})
>>> saver
JsonSaver(encoding='utf-8')

coola.io.save_json

save_json(
    to_save: Any,
    path: Path,
    *,
    encoding: str = DEFAULT_ENCODING,
    exist_ok: bool = False,
    **kwargs: Any,
) -> None

Save the given data in a JSON file.

Parameters:

Name Type Description Default
to_save Any

The data to write in a JSON file.

required
path Path

The path where to write the JSON file.

required
encoding str

The file encoding to use when writing the file. Defaults to "utf-8".

DEFAULT_ENCODING
exist_ok bool

If exist_ok is False (the default), FileExistsError is raised if the target file already exists. If exist_ok is True, FileExistsError will not be raised unless the given path already exists in the file system and is not a file.

False
**kwargs Any

Additional arguments passed to json.dump, e.g. indent or sort_keys.

{}

Raises:

Type Description
FileExistsError

if the file already exists.

Example
>>> import tempfile
>>> from pathlib import Path
>>> from coola.io import save_json, load_json
>>> with tempfile.TemporaryDirectory() as tmpdir:
...     path = Path(tmpdir).joinpath("data.json")
...     save_json({"key1": [1, 2, 3], "key2": "abc"}, path)
...     data = load_json(path)
...     data
...
{'key1': [1, 2, 3], 'key2': 'abc'}

coola.io.save_pickle

save_pickle(
    to_save: Any,
    path: Path,
    *,
    exist_ok: bool = False,
    **kwargs: Any,
) -> None

Save the given data in a pickle file.

Parameters:

Name Type Description Default
to_save Any

The data to write in a pickle file.

required
path Path

The path where to write the pickle file.

required
exist_ok bool

If exist_ok is False (the default), FileExistsError is raised if the target file already exists. If exist_ok is True, FileExistsError will not be raised unless the given path already exists in the file system and is not a file.

False
**kwargs Any

Additional arguments passed to pickle.dump.

{}

Raises:

Type Description
FileExistsError

if the file already exists.

Example
>>> import tempfile
>>> from pathlib import Path
>>> from coola.io import save_pickle, load_pickle
>>> with tempfile.TemporaryDirectory() as tmpdir:
...     path = Path(tmpdir).joinpath("data.pkl")
...     save_pickle({"key1": [1, 2, 3], "key2": "abc"}, path)
...     data = load_pickle(path)
...     data
...
{'key1': [1, 2, 3], 'key2': 'abc'}

coola.io.save_text

save_text(
    to_save: Any,
    path: Path,
    *,
    encoding: str = DEFAULT_ENCODING,
    exist_ok: bool = False,
) -> None

Save the given data to a text file.

Parameters:

Name Type Description Default
to_save Any

The data to write to the text file.

required
path Path

The path where to write the text file.

required
encoding str

The file encoding to use when writing the file. Defaults to "utf-8".

DEFAULT_ENCODING
exist_ok bool

If exist_ok is False (the default), FileExistsError is raised if the target file already exists. If exist_ok is True, FileExistsError will not be raised unless the given path already exists in the file system and is not a file.

False

Raises:

Type Description
FileExistsError

if the file already exists.

Note

If the data to save is not a string, it is converted to a string before being saved by using str.

Example
>>> import tempfile
>>> from pathlib import Path
>>> from coola.io import save_text, load_text
>>> with tempfile.TemporaryDirectory() as tmpdir:
...     path = Path(tmpdir).joinpath("data.txt")
...     save_text("hello", path)
...     data = load_text(path)
...     data
...
'hello'

coola.io.save_torch

save_torch(
    to_save: Any,
    path: Path,
    *,
    exist_ok: bool = False,
    **kwargs: Any,
) -> None

Save the given data in a PyTorch file.

Parameters:

Name Type Description Default
to_save Any

The data to write in a PyTorch file.

required
path Path

The path where to write the PyTorch file.

required
exist_ok bool

If exist_ok is False (the default), FileExistsError is raised if the target file already exists. If exist_ok is True, FileExistsError will not be raised unless the given path already exists in the file system and is not a file.

False
**kwargs Any

Additional arguments passed to torch.save.

{}

Raises:

Type Description
FileExistsError

if the file already exists.

Example
>>> import tempfile
>>> from pathlib import Path
>>> from coola.io import save_torch, load_torch
>>> with tempfile.TemporaryDirectory() as tmpdir:
...     path = Path(tmpdir).joinpath("data.pt")
...     save_torch({"key1": [1, 2, 3], "key2": "abc"}, path)
...     data = load_torch(path)
...     data
...
{'key1': [1, 2, 3], 'key2': 'abc'}

coola.io.base

Contain the base class to implement a data loader or saver object.

coola.io.base.BaseFileSaver

Bases: BaseSaver[T]

Define the base class to implement a file saver.

Example
>>> import tempfile
>>> from pathlib import Path
>>> from coola.io import JsonSaver, JsonLoader
>>> with tempfile.TemporaryDirectory() as tmpdir:
...     path = Path(tmpdir).joinpath("data.json")
...     JsonSaver().save({"key1": [1, 2, 3], "key2": "abc"}, path)
...     data = JsonLoader().load(path)
...     data
...
{'key1': [1, 2, 3], 'key2': 'abc'}

coola.io.base.BaseFileSaver.save

save(
    to_save: T, path: Path, *, exist_ok: bool = False
) -> None

Save the data into the given path.

Parameters:

Name Type Description Default
to_save T

The data to save. The data should be compatible with the saving engine.

required
path Path

The path where to save the data.

required
exist_ok bool

If exist_ok is False (the default), FileExistsError is raised if the target file already exists. If exist_ok is True, FileExistsError will not be raised unless the given path already exists in the file system and is not a file.

Note: the exist_ok=False guarantee is protected against a concurrent creation of path between the initial existence check and the commit step (the commit uses os.link which atomically fails with FileExistsError in that case).

Note: concurrent save calls targeting the same path, whether from within this process (an in-memory lock) or from other processes (a path.lock lock file created next to path), are serialized so they cannot interleave their write/commit steps and corrupt or partially overwrite each other's data. This holds for exist_ok=True too: two concurrent calls with exist_ok=True can no longer race each other's tmp_path.replace(path) commit; each call's replace happens fully before the next one starts. Which call's data ends up on disk is still whichever one wins the lock last (no ordering is guaranteed beyond "no interleaving"), and the lock file is removed once the holder releases it, so it does not accumulate on disk.

False

Raises:

Type Description
FileExistsError

if the file already exists.

Example
>>> import tempfile
>>> from pathlib import Path
>>> from coola.io import JsonSaver, JsonLoader
>>> with tempfile.TemporaryDirectory() as tmpdir:
...     path = Path(tmpdir).joinpath("data.json")
...     JsonSaver().save({"key1": [1, 2, 3], "key2": "abc"}, path)
...     data = JsonLoader().load(path)
...     data
...
{'key1': [1, 2, 3], 'key2': 'abc'}

coola.io.base.BaseLoader

Bases: ABC, Generic[T]

Define the base class to implement a data loader.

Example
>>> import tempfile
>>> from pathlib import Path
>>> from coola.io import save_json, JsonLoader
>>> with tempfile.TemporaryDirectory() as tmpdir:
...     path = Path(tmpdir).joinpath("data.json")
...     save_json({"key1": [1, 2, 3], "key2": "abc"}, path)
...     data = JsonLoader().load(path)
...     data
...
{'key1': [1, 2, 3], 'key2': 'abc'}

coola.io.base.BaseLoader.equal abstractmethod

equal(other: Any, equal_nan: bool = False) -> bool

Indicate if two objects are equal or not.

Parameters:

Name Type Description Default
other Any

The object to compare with.

required
equal_nan bool

If True, then two NaNs will be considered equal.

False

Returns:

Type Description
bool

True if the two objects are equal, otherwise False.

Example
>>> from coola.io import JsonLoader, TextLoader
>>> JsonLoader().equal(JsonLoader())
True
>>> JsonLoader().equal(TextLoader())
False

coola.io.base.BaseLoader.load abstractmethod

load(path: Path) -> T

Load the data from the given path.

Parameters:

Name Type Description Default
path Path

The path with the data to load.

required

Returns:

Type Description
T

The data

Example
>>> import tempfile
>>> from pathlib import Path
>>> from coola.io import save_json, JsonLoader
>>> with tempfile.TemporaryDirectory() as tmpdir:
...     path = Path(tmpdir).joinpath("data.json")
...     save_json({"key1": [1, 2, 3], "key2": "abc"}, path)
...     data = JsonLoader().load(path)
...     data
...
{'key1': [1, 2, 3], 'key2': 'abc'}

coola.io.base.BaseSaver

Bases: ABC, Generic[T]

Define the base class to implement a data saver.

Example
>>> import tempfile
>>> from pathlib import Path
>>> from coola.io import JsonSaver, JsonLoader
>>> with tempfile.TemporaryDirectory() as tmpdir:
...     path = Path(tmpdir).joinpath("data.json")
...     JsonSaver().save({"key1": [1, 2, 3], "key2": "abc"}, path)
...     data = JsonLoader().load(path)
...     data
...
{'key1': [1, 2, 3], 'key2': 'abc'}

coola.io.base.BaseSaver.equal abstractmethod

equal(other: Any, equal_nan: bool = False) -> bool

Indicate if two objects are equal or not.

Parameters:

Name Type Description Default
other Any

The object to compare with.

required
equal_nan bool

If True, then two NaNs will be considered equal.

False

Returns:

Type Description
bool

True if the two objects are equal, otherwise False.

Example
>>> from coola.io import JsonSaver, TextSaver
>>> JsonSaver().equal(JsonSaver())
True
>>> JsonSaver().equal(TextSaver())
False

coola.io.base.BaseSaver.save abstractmethod

save(
    to_save: T, path: Path, *, exist_ok: bool = False
) -> None

Save the data into the given path.

Parameters:

Name Type Description Default
to_save T

The data to save. The data should be compatible with the saving engine.

required
path Path

The path where to save the data.

required
exist_ok bool

If exist_ok is False (the default), an exception is raised if the target path already exists.

False
Example
>>> import tempfile
>>> from pathlib import Path
>>> from coola.io import JsonSaver, JsonLoader
>>> with tempfile.TemporaryDirectory() as tmpdir:
...     path = Path(tmpdir).joinpath("data.json")
...     JsonSaver().save({"key1": [1, 2, 3], "key2": "abc"}, path)
...     data = JsonLoader().load(path)
...     data
...
{'key1': [1, 2, 3], 'key2': 'abc'}

coola.io.base.is_loader_config

is_loader_config(config: dict[Any, Any]) -> bool

Indicate if the input configuration is a configuration for a BaseLoader.

This function only checks if the value of the key _target_ is valid. It does not check the other values. If _target_ indicates a function, the returned type hint is used to check the class.

Parameters:

Name Type Description Default
config dict[Any, Any]

The configuration to check.

required

Returns:

Type Description
bool

True if the input configuration is a configuration for a BaseLoader object.

Example
>>> from coola.io import is_loader_config
>>> is_loader_config({"_target_": "coola.io.JsonLoader"})
True

coola.io.base.is_saver_config

is_saver_config(config: dict[Any, Any]) -> bool

Indicate if the input configuration is a configuration for a BaseSaver.

This function only checks if the value of the key _target_ is valid. It does not check the other values. If _target_ indicates a function, the returned type hint is used to check the class.

Parameters:

Name Type Description Default
config dict[Any, Any]

The configuration to check.

required

Returns:

Type Description
bool

True if the input configuration is a configuration for a BaseSaver object.

Example
>>> from coola.io import is_saver_config
>>> is_saver_config({"_target_": "coola.io.JsonSaver"})
True

coola.io.base.resolve_loader

resolve_loader(
    loader: BaseLoader[T] | dict[Any, Any],
) -> BaseLoader[T]

Set up a data loader.

The data loader is instantiated from its configuration by using the BaseLoader factory function.

Parameters:

Name Type Description Default
loader BaseLoader[T] | dict[Any, Any]

The data loader or its configuration.

required

Returns:

Type Description
BaseLoader[T]

The instantiated data loader.

Security

When loader is a configuration :class:dict, its "_target_" value is imported and instantiated with no allowlist (see :func:coola.factory.factory). Only call this function with a configuration that comes from a trusted source; never resolve a loader configuration derived from untrusted input, as this is a remote-code-execution vector.

Example
>>> from coola.io import resolve_loader
>>> loader = resolve_loader({"_target_": "coola.io.JsonLoader"})
>>> loader
JsonLoader()

coola.io.base.resolve_saver

resolve_saver(
    saver: BaseSaver[T] | dict[Any, Any],
) -> BaseSaver[T]

Set up a data saver.

The data saver is instantiated from its configuration by using the BaseSaver factory function.

Parameters:

Name Type Description Default
saver BaseSaver[T] | dict[Any, Any]

The data saver or its configuration.

required

Returns:

Type Description
BaseSaver[T]

The instantiated data saver.

Security

When saver is a configuration :class:dict, its "_target_" value is imported and instantiated with no allowlist (see :func:coola.factory.factory). Only call this function with a configuration that comes from a trusted source; never resolve a saver configuration derived from untrusted input, as this is a remote-code-execution vector.

Example
>>> from coola.io import resolve_saver
>>> saver = resolve_saver({"_target_": "coola.io.JsonSaver"})
>>> saver
JsonSaver(encoding='utf-8')

coola.io.json

Contain JSON-based data loaders and savers.

coola.io.json.JsonLoader

Bases: InlineDisplayMixin, BaseLoader[T]

Implement a data loader to load data in a JSON file.

Example
>>> import tempfile
>>> from pathlib import Path
>>> from coola.io import save_json, JsonLoader
>>> with tempfile.TemporaryDirectory() as tmpdir:
...     path = Path(tmpdir).joinpath("data.json")
...     save_json({"key1": [1, 2, 3], "key2": "abc"}, path)
...     data = JsonLoader().load(path)
...     data
...
{'key1': [1, 2, 3], 'key2': 'abc'}

coola.io.json.JsonSaver

Bases: InlineDisplayMixin, BaseFileSaver[T]

Implement a file saver to save data with a JSON file.

Parameters:

Name Type Description Default
encoding str

The file encoding to use when writing the file. Defaults to "utf-8".

DEFAULT_ENCODING
**kwargs Any

Additional arguments passed to json.dump, e.g. indent or sort_keys.

{}
Example
>>> import tempfile
>>> from pathlib import Path
>>> from coola.io import JsonSaver, JsonLoader
>>> with tempfile.TemporaryDirectory() as tmpdir:
...     path = Path(tmpdir).joinpath("data.json")
...     JsonSaver().save({"key1": [1, 2, 3], "key2": "abc"}, path)
...     data = JsonLoader().load(path)
...     data
...
{'key1': [1, 2, 3], 'key2': 'abc'}

coola.io.json.load_json

load_json(path: Path) -> Any

Load the data from a given JSON file.

Parameters:

Name Type Description Default
path Path

The path to the JSON file.

required

Returns:

Type Description
Any

The data from the JSON file.

Example
>>> import tempfile
>>> from pathlib import Path
>>> from coola.io import save_json, load_json
>>> with tempfile.TemporaryDirectory() as tmpdir:
...     path = Path(tmpdir).joinpath("data.json")
...     save_json({"key1": [1, 2, 3], "key2": "abc"}, path)
...     data = load_json(path)
...     data
...
{'key1': [1, 2, 3], 'key2': 'abc'}

coola.io.json.save_json

save_json(
    to_save: Any,
    path: Path,
    *,
    encoding: str = DEFAULT_ENCODING,
    exist_ok: bool = False,
    **kwargs: Any,
) -> None

Save the given data in a JSON file.

Parameters:

Name Type Description Default
to_save Any

The data to write in a JSON file.

required
path Path

The path where to write the JSON file.

required
encoding str

The file encoding to use when writing the file. Defaults to "utf-8".

DEFAULT_ENCODING
exist_ok bool

If exist_ok is False (the default), FileExistsError is raised if the target file already exists. If exist_ok is True, FileExistsError will not be raised unless the given path already exists in the file system and is not a file.

False
**kwargs Any

Additional arguments passed to json.dump, e.g. indent or sort_keys.

{}

Raises:

Type Description
FileExistsError

if the file already exists.

Example
>>> import tempfile
>>> from pathlib import Path
>>> from coola.io import save_json, load_json
>>> with tempfile.TemporaryDirectory() as tmpdir:
...     path = Path(tmpdir).joinpath("data.json")
...     save_json({"key1": [1, 2, 3], "key2": "abc"}, path)
...     data = load_json(path)
...     data
...
{'key1': [1, 2, 3], 'key2': 'abc'}

coola.io.pickle

Contain pickle-based data loaders and savers.

coola.io.pickle.PickleLoader

Bases: InlineDisplayMixin, BaseLoader[T]

Implement a data loader to load data in a pickle file.

Warning

load calls pickle.load on the given file. Unpickling can execute arbitrary code as a side effect of deserialization. Only load pickle files that come from a trusted source; never load a pickle file that comes from an untrusted or unauthenticated source.

Example
>>> import tempfile
>>> from pathlib import Path
>>> from coola.io import save_pickle, PickleLoader
>>> with tempfile.TemporaryDirectory() as tmpdir:
...     path = Path(tmpdir).joinpath("data.pkl")
...     save_pickle({"key1": [1, 2, 3], "key2": "abc"}, path)
...     data = PickleLoader().load(path)
...     data
...
{'key1': [1, 2, 3], 'key2': 'abc'}

coola.io.pickle.PickleSaver

Bases: InlineDisplayMixin, BaseFileSaver[T]

Implement a file saver to save data with a pickle file.

Parameters:

Name Type Description Default
**kwargs Any

Additional arguments passed to pickle.dump.

{}
Example
>>> import tempfile
>>> from pathlib import Path
>>> from coola.io import PickleSaver, PickleLoader
>>> with tempfile.TemporaryDirectory() as tmpdir:
...     path = Path(tmpdir).joinpath("data.pkl")
...     PickleSaver().save({"key1": [1, 2, 3], "key2": "abc"}, path)
...     data = PickleLoader().load(path)
...     data
...
{'key1': [1, 2, 3], 'key2': 'abc'}

coola.io.pickle.load_pickle

load_pickle(path: Path) -> Any

Load the data from a given pickle file.

Parameters:

Name Type Description Default
path Path

The path to the pickle file.

required

Returns:

Type Description
Any

The data from the pickle file.

Warning

This function unpickles the file's content, which can execute arbitrary code as a side effect of deserialization. Only load pickle files that come from a trusted source; never load a pickle file that comes from an untrusted or unauthenticated source.

Example
>>> import tempfile
>>> from pathlib import Path
>>> from coola.io import save_pickle, load_pickle
>>> with tempfile.TemporaryDirectory() as tmpdir:
...     path = Path(tmpdir).joinpath("data.pkl")
...     save_pickle({"key1": [1, 2, 3], "key2": "abc"}, path)
...     data = load_pickle(path)
...     data
...
{'key1': [1, 2, 3], 'key2': 'abc'}

coola.io.pickle.save_pickle

save_pickle(
    to_save: Any,
    path: Path,
    *,
    exist_ok: bool = False,
    **kwargs: Any,
) -> None

Save the given data in a pickle file.

Parameters:

Name Type Description Default
to_save Any

The data to write in a pickle file.

required
path Path

The path where to write the pickle file.

required
exist_ok bool

If exist_ok is False (the default), FileExistsError is raised if the target file already exists. If exist_ok is True, FileExistsError will not be raised unless the given path already exists in the file system and is not a file.

False
**kwargs Any

Additional arguments passed to pickle.dump.

{}

Raises:

Type Description
FileExistsError

if the file already exists.

Example
>>> import tempfile
>>> from pathlib import Path
>>> from coola.io import save_pickle, load_pickle
>>> with tempfile.TemporaryDirectory() as tmpdir:
...     path = Path(tmpdir).joinpath("data.pkl")
...     save_pickle({"key1": [1, 2, 3], "key2": "abc"}, path)
...     data = load_pickle(path)
...     data
...
{'key1': [1, 2, 3], 'key2': 'abc'}

coola.io.text

Contain text-based data loaders and savers.

coola.io.text.TextLoader

Bases: InlineDisplayMixin, BaseLoader[str]

Implement a data loader to load data from a text file.

Parameters:

Name Type Description Default
encoding str

The file encoding to use when reading the file. Defaults to "utf-8".

DEFAULT_ENCODING
Example
>>> import tempfile
>>> from pathlib import Path
>>> from coola.io import save_text, TextLoader
>>> with tempfile.TemporaryDirectory() as tmpdir:
...     path = Path(tmpdir).joinpath("data.txt")
...     save_text("hello", path)
...     data = TextLoader().load(path)
...     data
...
'hello'

coola.io.text.TextSaver

Bases: InlineDisplayMixin, BaseFileSaver[str]

Implement a file saver to save data to a text file.

Parameters:

Name Type Description Default
encoding str

The file encoding to use when writing the file. Defaults to "utf-8".

DEFAULT_ENCODING
Note

If the data to save is not a string, it is converted to a string before being saved by using str.

Example
>>> import tempfile
>>> from pathlib import Path
>>> from coola.io import TextSaver, TextLoader
>>> with tempfile.TemporaryDirectory() as tmpdir:
...     path = Path(tmpdir).joinpath("data.txt")
...     TextSaver().save("hello", path)
...     data = TextLoader().load(path)
...     data
...
'hello'

coola.io.text.load_text

load_text(
    path: Path, encoding: str = DEFAULT_ENCODING
) -> str

Load the data from a given text file.

Parameters:

Name Type Description Default
path Path

The path to the text file.

required
encoding str

The file encoding to use when reading the file. Defaults to "utf-8".

DEFAULT_ENCODING

Returns:

Type Description
str

The text content of the file as a string.

Example
>>> import tempfile
>>> from pathlib import Path
>>> from coola.io import save_text, load_text
>>> with tempfile.TemporaryDirectory() as tmpdir:
...     path = Path(tmpdir).joinpath("data.txt")
...     save_text("hello", path)
...     data = load_text(path)
...     data
...
'hello'

coola.io.text.save_text

save_text(
    to_save: Any,
    path: Path,
    *,
    encoding: str = DEFAULT_ENCODING,
    exist_ok: bool = False,
) -> None

Save the given data to a text file.

Parameters:

Name Type Description Default
to_save Any

The data to write to the text file.

required
path Path

The path where to write the text file.

required
encoding str

The file encoding to use when writing the file. Defaults to "utf-8".

DEFAULT_ENCODING
exist_ok bool

If exist_ok is False (the default), FileExistsError is raised if the target file already exists. If exist_ok is True, FileExistsError will not be raised unless the given path already exists in the file system and is not a file.

False

Raises:

Type Description
FileExistsError

if the file already exists.

Note

If the data to save is not a string, it is converted to a string before being saved by using str.

Example
>>> import tempfile
>>> from pathlib import Path
>>> from coola.io import save_text, load_text
>>> with tempfile.TemporaryDirectory() as tmpdir:
...     path = Path(tmpdir).joinpath("data.txt")
...     save_text("hello", path)
...     data = load_text(path)
...     data
...
'hello'

coola.io.torch

Contain torch-based data loaders and savers.

coola.io.torch.TorchLoader

Bases: InlineDisplayMixin, BaseLoader[T]

Implement a data loader to load data in a PyTorch file.

Parameters:

Name Type Description Default
**kwargs Any

Additional arguments passed to torch.load.

{}
Example
>>> import tempfile
>>> from pathlib import Path
>>> from coola.io import save_torch, TorchLoader
>>> with tempfile.TemporaryDirectory() as tmpdir:
...     path = Path(tmpdir).joinpath("data.pt")
...     save_torch({"key1": [1, 2, 3], "key2": "abc"}, path)
...     data = TorchLoader().load(path)
...     data
...
{'key1': [1, 2, 3], 'key2': 'abc'}

coola.io.torch.TorchSaver

Bases: InlineDisplayMixin, BaseFileSaver[T]

Implement a file saver to save data with a PyTorch file.

Parameters:

Name Type Description Default
**kwargs Any

Additional arguments passed to torch.save.

{}
Example
>>> import tempfile
>>> from pathlib import Path
>>> from coola.io import TorchSaver, TorchLoader
>>> with tempfile.TemporaryDirectory() as tmpdir:
...     path = Path(tmpdir).joinpath("data.pt")
...     TorchSaver().save({"key1": [1, 2, 3], "key2": "abc"}, path)
...     data = TorchLoader().load(path)
...     data
...
{'key1': [1, 2, 3], 'key2': 'abc'}

coola.io.torch.load_torch

load_torch(path: Path, **kwargs: Any) -> Any

Load the data from a given PyTorch file.

Parameters:

Name Type Description Default
path Path

The path to the PyTorch file.

required
**kwargs Any

Additional arguments passed to torch.load.

{}

Returns:

Type Description
Any

The data from the PyTorch file.

Example
>>> import tempfile
>>> from pathlib import Path
>>> from coola.io import save_torch, load_torch
>>> with tempfile.TemporaryDirectory() as tmpdir:
...     path = Path(tmpdir).joinpath("data.pt")
...     save_torch({"key1": [1, 2, 3], "key2": "abc"}, path)
...     data = load_torch(path)
...     data
...
{'key1': [1, 2, 3], 'key2': 'abc'}

coola.io.torch.save_torch

save_torch(
    to_save: Any,
    path: Path,
    *,
    exist_ok: bool = False,
    **kwargs: Any,
) -> None

Save the given data in a PyTorch file.

Parameters:

Name Type Description Default
to_save Any

The data to write in a PyTorch file.

required
path Path

The path where to write the PyTorch file.

required
exist_ok bool

If exist_ok is False (the default), FileExistsError is raised if the target file already exists. If exist_ok is True, FileExistsError will not be raised unless the given path already exists in the file system and is not a file.

False
**kwargs Any

Additional arguments passed to torch.save.

{}

Raises:

Type Description
FileExistsError

if the file already exists.

Example
>>> import tempfile
>>> from pathlib import Path
>>> from coola.io import save_torch, load_torch
>>> with tempfile.TemporaryDirectory() as tmpdir:
...     path = Path(tmpdir).joinpath("data.pt")
...     save_torch({"key1": [1, 2, 3], "key2": "abc"}, path)
...     data = load_torch(path)
...     data
...
{'key1': [1, 2, 3], 'key2': 'abc'}

coola.io.utils

Contain I/O utility functions.

coola.io.utils.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.

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