pytupli.storage.FileStorage

class FileStorage(storage_base_dir: str = '_local_storage')[source]

Bases: TupliStorage

Storage class for saving and loading benchmarks to/from files.

Initializes the FileStorage with the specified base directory.

Parameters:: storage_base_dir (str) – The base directory for storage.
Returns:: FileStorage – The initialized FileStorage object.
Raises:: TupliStorageError – If the base directory cannot be created.

Methods

`convert_filter_to_function`	Convert a BaseFilter object to a filter function that can be applied to dictionaries.
`delete_artifact`	Deletes the specified artifact from the storage.
`delete_benchmark`	Deletes the specified benchmark from the storage.
`delete_episode`	Deletes an episode from the local file storage.
`list_artifacts`	Lists all artifacts in the storage that match the specified filter.
`list_benchmarks`	Lists all benchmarks in the storage that match the specified filter.
`list_episodes`	Lists all episodes in the local file storage that match the specified filter.
`load_artifact`	Loads an artifact from a file.
`load_benchmark`	Loads a benchmark from the file using the benchmark ID.
`publish_artifact`	Publishing functionality is not available in FileStorage.
`publish_benchmark`	Publishing functionality is not available in FileStorage.
`publish_episode`	Publishes an episode in the specified group in local file storage.
`record_episode`	Records an episode in the local file storage.
`store_artifact`	Stores an artifact as a file and returns its metadata.
`store_benchmark`	Saves the benchmark object to a file.

convert_filter_to_function(filter_obj: BaseFilter) → Callable[[Dict[str, Any]], bool][source]

Convert a BaseFilter object to a filter function that can be applied to dictionaries. Supports nested dictionary access with keys in the form of “a.b.key”.

Parameters:: filter_obj (BaseFilter) – The filter object to convert.
Returns:: Callable[[Dict[str, Any]], bool] – A function that takes a dictionary and returns True if the dictionary matches the filter.
Raises:: TupliStorageError – If the filter type is unknown.

delete_artifact(uri: str) → None[source]

Deletes the specified artifact from the storage.

Parameters:: uri (str) – The ID of the artifact to delete.

delete_benchmark(uri: str) → None[source]

Deletes the specified benchmark from the storage.

Parameters:: uri (str) – The ID of the benchmark to delete.

delete_episode(uri: str) → None[source]

Deletes an episode from the local file storage.

Parameters:: uri (str) – The ID of the episode to delete.
Raises:: TupliStorageError – If the episode cannot be deleted.

list_artifacts(filter: BaseFilter = None) → list[ArtifactMetadataItem][source]

Lists all artifacts in the storage that match the specified filter.

Parameters:: filter (BaseFilter, optional) – The filter to apply when listing artifacts.
Returns:: list[ArtifactMetadataItem] – A list of artifacts that match the filter.

list_benchmarks(filter: BaseFilter = None) → list[BenchmarkHeader][source]

Lists all benchmarks in the storage that match the specified filter.

Parameters:: filter (BaseFilter, optional) – The filter to apply when listing benchmarks.
Returns:: list[BenchmarkHeader] – A list of benchmark headers that match the filter.

list_episodes(filter: BaseFilter = None, include_tuples: bool = False) → list[EpisodeHeader] | list[EpisodeItem][source]

Lists all episodes in the local file storage that match the specified filter.

Parameters:

filter (BaseFilter, optional) – The filter to apply when listing episodes.
include_tuples (bool, optional) – Whether to include tuples in the episode data.

Returns:

list[EpisodeHeader] | list[EpisodeItem] – A list of episode headers or full episode items.

load_artifact(uri: str, **kwargs) → bytes[source]

Loads an artifact from a file.

Parameters:

uri (str) – The ID of the artifact to load.
kwargs – Additional arguments (ignored in file storage)

Returns:

bytes – The raw artifact data

Raises:

TupliStorageError – If the artifact cannot be loaded.

load_benchmark(uri: str) → Benchmark[source]

Loads a benchmark from the file using the benchmark ID.

Parameters:: uri (str) – The ID of the benchmark to be loaded.
Returns:: Benchmark – The loaded benchmark object.
Raises:: TupliStorageError – If the benchmark cannot be loaded or parsed.

publish_artifact(uri: str, publish_in: str = 'global') → None[source]

Publishing functionality is not available in FileStorage. This is a placeholder to implement the interface.

Parameters:

uri (str) – The URI of the artifact to publish.
publish_in (str) – The group to publish the artifact in. Defaults to ‘global’.

publish_benchmark(uri: str, publish_in: str = 'global') → None[source]

Publishing functionality is not available in FileStorage. This is a placeholder to implement the interface.

Parameters:

uri (str) – The URI of the benchmark to publish.
publish_in (str) – The group to publish the benchmark in. Defaults to ‘global’.

Returns:

str – The URI of the benchmark.

publish_episode(uri: str, publish_in: str = 'global') → None[source]: Publishes an episode in the specified group in local file storage. This is a placeholder to implement the interface. :param uri: The ID of the episode to publish. :type uri: str :param publish_in: The group to publish the episode in. Defaults to ‘global’. :type publish_in: str

record_episode(episode: Episode) → EpisodeHeader[source]

Records an episode in the local file storage.

Parameters:: episode (Episode) – The episode to record.
Returns:: EpisodeHeader – The header of the recorded episode.
Raises:: TupliStorageError – If the episode cannot be saved or serialized.

store_artifact(artifact: bytes, metadata: ArtifactMetadata) → ArtifactMetadataItem[source]

Stores an artifact as a file and returns its metadata.

Parameters:

artifact (bytes) – The artifact data to store.
metadata (ArtifactMetadata) – Metadata for the artifact.

Returns:

ArtifactMetadataItem – Metadata for the stored artifact.

Raises:

TupliStorageError – If the artifact cannot be stored or metadata cannot be serialized.

store_benchmark(benchmark_query: BenchmarkQuery) → BenchmarkHeader[source]

Saves the benchmark object to a file.

Parameters:: benchmark_query (BenchmarkQuery) – The benchmark query to be saved.
Returns:: BenchmarkHeader – The header of the saved benchmark.
Raises:: TupliStorageError – If the benchmark cannot be saved or serialized.