pytupli.storage.TupliStorage

class TupliStorage[source]

Bases: object

Base class for storing StorableObjects.

Methods

delete_artifact

Deletes the specified artifact from the storage.

delete_benchmark

Deletes the specified benchmark from the storage.

delete_episode

Deletes the specified episode from the 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 storage that match the specified filter.

load_artifact

Loads the artifact from the storage.

load_benchmark

Loads data from the specified URI.

publish_episode

Publishes the specified episode in the storage.

record_episode

Records an episode in the storage.

store_artifact

Stores the artifact in the storage.

store_benchmark

Saves the serialized object to the specified storage.
delete_artifact(uri: str) → None[source]

Deletes the specified artifact from the storage.

Parameters:

uri (str) – The URI/ID of the artifact to delete.

delete_benchmark(uri: str) → None[source]

Deletes the specified benchmark from the storage. :param uri: The URI/ID of the benchmark to delete. :type uri: str

delete_episode(uri: str) → None[source]

Deletes the specified episode from the storage.

Parameters:

uri (str) – The URI/ID of the episode to delete.

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

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

Parameters:

filter (BaseFilter) – The filter to apply when listing artifacts.

Returns:

list[ArtifactMetadataItem] – A list of artifacts that match the filter.

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

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

Parameters:

filter (BaseFilter) – 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 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 with tuples.

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

Loads the artifact from the storage.

Parameters:

  • uri (str) – The URI/ID of the artifact to load.

  • kwargs – Additional arguments for loading the artifact.

load_benchmark(uri: str) → Benchmark[source]

Loads data from the specified URI.

Parameters:

uri (str) – The URI of the data to be loaded.

Returns:

Benchmark – The loaded benchmark object.

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

Publishes the specified episode in the storage.

Parameters:

  • uri (str) – The URI/ID of the episode to publish.

  • publish_in (str) – The group to publish the episode in. Defaults to ‘global’.

record_episode(episode: Episode) → EpisodeHeader[source]

Records an episode in the storage.

Parameters:

episode (Episode) – The episode to record.

Returns:

EpisodeHeader – The header of the recorded episode.

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

Stores the artifact in the storage.

Parameters:

  • artifact (bytes) – The artifact to store.

  • metadata (ArtifactMetadata) – Metadata for the artifact.

Returns:

ArtifactMetadataItem – Metadata item for the stored artifact.

store_benchmark(benchmark_query: BenchmarkQuery) → BenchmarkHeader[source]

Saves the serialized object to the specified storage.

Parameters:

benchmark_query (BenchmarkQuery) – The serialized benchmark object to be saved.

Returns:

BenchmarkHeader – The header of the saved benchmark.