Metadata Stores API 🗄️

Metadata Stores track pipeline runs, step status, and artifact lineage.

Base Metadata Store

Bases: ABC

Base class for metadata storage backends.

Functions

list_assets(limit: int | None = None, **filters) -> list[dict] abstractmethod

List assets with optional filters.

Source code in flowyml/storage/metadata.py

@abstractmethod
def list_assets(self, limit: int | None = None, **filters) -> list[dict]:
    """List assets with optional filters."""
    pass

list_pipelines() -> list[str] abstractmethod

List all unique pipeline names.

Source code in flowyml/storage/metadata.py

@abstractmethod
def list_pipelines(self) -> list[str]:
    """List all unique pipeline names."""
    pass

list_runs(limit: int | None = None) -> list[dict] abstractmethod

List all runs.

Source code in flowyml/storage/metadata.py

@abstractmethod
def list_runs(self, limit: int | None = None) -> list[dict]:
    """List all runs."""
    pass

load_artifact(artifact_id: str) -> dict | None abstractmethod

Load artifact metadata.

Source code in flowyml/storage/metadata.py

@abstractmethod
def load_artifact(self, artifact_id: str) -> dict | None:
    """Load artifact metadata."""
    pass

load_run(run_id: str) -> dict | None abstractmethod

Load run metadata.

Source code in flowyml/storage/metadata.py

@abstractmethod
def load_run(self, run_id: str) -> dict | None:
    """Load run metadata."""
    pass

query(**filters) -> list[dict] abstractmethod

Query runs with filters.

Source code in flowyml/storage/metadata.py

@abstractmethod
def query(self, **filters) -> list[dict]:
    """Query runs with filters."""
    pass

save_artifact(artifact_id: str, metadata: dict) -> None abstractmethod

Save artifact metadata.

Source code in flowyml/storage/metadata.py

@abstractmethod
def save_artifact(self, artifact_id: str, metadata: dict) -> None:
    """Save artifact metadata."""
    pass

save_run(run_id: str, metadata: dict) -> None abstractmethod

Save run metadata.

Source code in flowyml/storage/metadata.py

@abstractmethod
def save_run(self, run_id: str, metadata: dict) -> None:
    """Save run metadata."""
    pass

SQLite Metadata Store