Storage API Reference

ArtifactStore

flowyml.storage.artifacts.ArtifactStore

Bases: ABC

Base class for artifact storage backends.

Functions

delete(path: str) -> None abstractmethod

Delete artifact at path.

Source code in flowyml/storage/artifacts.py

@abstractmethod
def delete(self, path: str) -> None:
    """Delete artifact at path."""
    pass

exists(path: str) -> bool abstractmethod

Check if artifact exists at path.

Source code in flowyml/storage/artifacts.py

@abstractmethod
def exists(self, path: str) -> bool:
    """Check if artifact exists at path."""
    pass

list_artifacts(prefix: str = '') -> list[str] abstractmethod

List all artifacts with optional prefix filter.

Source code in flowyml/storage/artifacts.py

@abstractmethod
def list_artifacts(self, prefix: str = "") -> list[str]:
    """List all artifacts with optional prefix filter."""
    pass

load(path: str) -> Any abstractmethod

Load an artifact from storage.

Parameters:

Name	Type	Description	Default
path	str	Storage path of the artifact	required

Returns:

Type	Description
Any	The loaded artifact

Source code in flowyml/storage/artifacts.py

@abstractmethod
def load(self, path: str) -> Any:
    """Load an artifact from storage.

    Args:
        path: Storage path of the artifact

    Returns:
        The loaded artifact
    """
    pass

materialize(obj: Any, name: str, run_id: str, step_name: str, project_name: str = 'default') -> str

Materialize artifact to structured storage.

Parameters:

Name	Type	Description	Default
obj	Any	Object to materialize	required
name	str	Name of the artifact	required
run_id	str	ID of the current run	required
step_name	str	Name of the step producing the artifact	required
project_name	str	Name of the project	'default'

Returns:

Type	Description
str	Path where artifact was saved

Source code in flowyml/storage/artifacts.py

def materialize(self, obj: Any, name: str, run_id: str, step_name: str, project_name: str = "default") -> str:
    """Materialize artifact to structured storage.

    Args:
        obj: Object to materialize
        name: Name of the artifact
        run_id: ID of the current run
        step_name: Name of the step producing the artifact
        project_name: Name of the project

    Returns:
        Path where artifact was saved
    """
    pass

save(artifact: Any, path: str, metadata: dict | None = None) -> str abstractmethod

Save an artifact to storage.

Parameters:

Name	Type	Description	Default
artifact	Any	The artifact to save	required
path	str	Storage path for the artifact	required
metadata	dict | None	Optional metadata dictionary	None

Returns:

Type	Description
str	Full path where artifact was saved

Source code in flowyml/storage/artifacts.py

@abstractmethod
def save(self, artifact: Any, path: str, metadata: dict | None = None) -> str:
    """Save an artifact to storage.

    Args:
        artifact: The artifact to save
        path: Storage path for the artifact
        metadata: Optional metadata dictionary

    Returns:
        Full path where artifact was saved
    """
    pass

MetadataStore

flowyml.storage.metadata.MetadataStore

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