Storage

Stores persist run records and artifacts. metalab supports file-based storage for local development and PostgreSQL for production deployments.

FileStore

metalab.FileStore

FileStore(config: FileStoreConfig)

Filesystem-based storage backend.

Provides atomic writes and per-run locking for concurrent access. Uses FileStoreLayout for all path construction.

Create via FileStoreConfig:

config = FileStoreConfig(root="./experiments")
store = config.connect()

Initialize from config.

Use FileStoreConfig(...).connect() to create instances.

Parameters:

Name	Type	Description	Default
config	FileStoreConfig	The FileStoreConfig for this store.	required

config property

config: FileStoreConfig

The configuration for this store.

is_scoped property

is_scoped: bool

True if this store is scoped to a specific experiment.

layout property

layout: FileStoreLayout

The layout configuration for this store.

root property

root: Path

The root directory of this store.

__repr__

__repr__() -> str

Return a string representation of the store.

delete_run

delete_run(run_id: str) -> None

Delete a run and all its artifacts/logs/derived metrics/results.

derived_exists

derived_exists(run_id: str) -> bool

Check if derived metrics exist for a run.

get_artifact

get_artifact(uri: str) -> bytes

Retrieve artifact data.

get_derived

get_derived(run_id: str) -> dict[str, Metric] | None

Retrieve derived metrics for a run.

get_experiment_manifest

get_experiment_manifest(experiment_id: str) -> dict[str, Any] | None

Retrieve experiment manifest by ID (most recent version).

get_log

get_log(run_id: str, name: str) -> str | None

Retrieve a log file.

Searches for logs matching the run_id. Handles legacy formats.

get_log_path

get_log_path(run_id: str, name: str) -> Path

Get the path where a log file should be written.

Enables streaming loggers to write directly to the store.

get_result

get_result(run_id: str, name: str) -> dict[str, Any] | None

Retrieve structured result data.

get_run_record

get_run_record(run_id: str) -> RunRecord | None

Retrieve a run record by ID.

get_working_directory

get_working_directory() -> Path

Return the root directory of this store.

Implements SupportsWorkingDirectory capability.

list_artifacts

list_artifacts(run_id: str) -> list[ArtifactDescriptor]

List artifacts for a run.

list_logs

list_logs(run_id: str) -> list[str]

List available log names for a run.

Returns a list of log names (e.g., ["run", "stdout", "stderr"]).

list_results

list_results(run_id: str) -> list[str]

List result names for a run.

list_run_records

list_run_records(experiment_id: str | None = None) -> list[RunRecord]

List run records, optionally filtered by experiment.

open_artifact

open_artifact(uri: str) -> 'BinaryIO'

Open an artifact for reading.

Implements SupportsArtifactOpen capability.

put_artifact

put_artifact(data: bytes | Path, descriptor: ArtifactDescriptor) -> ArtifactDescriptor

Store an artifact.

put_derived

put_derived(run_id: str, derived: dict[str, Metric]) -> None

Persist derived metrics for a run.

put_experiment_manifest

put_experiment_manifest(experiment_id: str, manifest: dict[str, Any], timestamp: str | None = None) -> None

Store an experiment manifest.

put_log

put_log(run_id: str, name: str, content: str) -> None

Store a log file for a run.

put_result

put_result(run_id: str, name: str, data: Any, dtype: str | None = None, shape: list[int] | None = None, metadata: dict[str, Any] | None = None) -> None

Store structured result data for a run.

Results are stored in JSON format at results/{run_id}/{name}.json.

put_run_record

put_run_record(record: RunRecord) -> None

Persist a run record atomically.

run_exists

run_exists(run_id: str) -> bool

Check if a run record exists.

scoped

scoped(experiment_id: str) -> 'FileStore'

Return a store scoped to the given experiment.

Parameters:

Name	Type	Description	Default
experiment_id	str	The experiment identifier.	required

Returns:

Type	Description
'FileStore'	A new FileStore scoped to the experiment.

metalab.FileStoreConfig dataclass

FileStoreConfig(*, experiment_id: str | None = None, root: str)

Bases: StoreConfig

Configuration for FileStore.

Example:

config = FileStoreConfig(root="./experiments")
scoped = config.scoped("my_exp:1.0")
store = scoped.connect()

# Or let runner handle scoping:
metalab.run(exp, store=config)  # auto-scopes to experiment

connect

connect() -> 'FileStore'

Create a FileStore from this config.

for_experiment

for_experiment(experiment_id: str) -> 'FileStoreConfig'

Get a scoped config for a specific experiment.

Alias for scoped() with a clearer name for loading/browsing context.

Parameters:

Name	Type	Description	Default
experiment_id	str	The experiment ID to scope to.	required

Returns:

Type	Description
'FileStoreConfig'	A new FileStoreConfig scoped to the experiment.

Example:

collection = FileStoreConfig(root="./experiments")
config = collection.for_experiment("my_exp:1.0")
results = load_results(config)

from_locator classmethod

from_locator(info: 'LocatorInfo', **kwargs: Any) -> 'FileStoreConfig'

Parse file:// locator into config.

list_experiments

list_experiments() -> list[str]

List all experiment IDs in this collection.

Discovers experiments by scanning subdirectories for _meta.json files that contain experiment_id. Only works on unscoped configs.

Returns:

Type	Description
list[str]	List of experiment IDs found in this collection.

Raises:

Type	Description
ValueError	If called on a scoped config.

Example:

config = FileStoreConfig(root="./experiments")
experiments = config.list_experiments()
# ['my_exp:1.0', 'my_exp:2.0', 'other_exp:1.0']

PostgresStore

For query-accelerated storage. Requires the postgres extra (uv add metalab[postgres]).

metalab.store.postgres.PostgresStoreConfig dataclass

PostgresStoreConfig(*, experiment_id: str | None = None, connection_string: str, file_root: str, schema: str = 'public', auto_migrate: bool = True, connect_timeout: float = 10.0, pool_min_size: int = 1, pool_max_size: int = 2)

Bases: StoreConfig

Configuration for PostgresStore.

Example:

config = PostgresStoreConfig(
    connection_string="postgresql://localhost/metalab",
    file_root="/data/experiments",
)
scoped = config.scoped("my_exp:1.0")
store = scoped.connect()

# Or let runner handle scoping:
metalab.run(exp, store=config)  # auto-scopes to experiment

connect

connect() -> 'PostgresStore'

Create a PostgresStore from this config.

If the connection_string lacks credentials, attempts to discover them from service.json at {file_root}/services/postgres/service.json.

for_experiment

for_experiment(experiment_id: str) -> 'PostgresStoreConfig'

Get a scoped config for a specific experiment.

Alias for scoped() with a clearer name for loading/browsing context.

Parameters:

Name	Type	Description	Default
experiment_id	str	The experiment ID to scope to.	required

Returns:

Type	Description
'PostgresStoreConfig'	A new PostgresStoreConfig scoped to the experiment.

from_locator classmethod

from_locator(info: 'LocatorInfo', **kwargs: Any) -> 'PostgresStoreConfig'

Parse postgresql:// locator into config.

list_experiments

list_experiments() -> list[str]

List all experiment IDs in this collection.

Discovers experiments by scanning subdirectories of file_root for _meta.json files that contain experiment_id. Only works on unscoped configs.

Returns:

Type	Description
list[str]	List of experiment IDs found in this collection.

Raises:

Type	Description
ValueError	If called on a scoped config.