Artifact Stores

zenml.artifact_stores special

An artifact store is a place where artifacts are stored. These artifacts may have been produced by the pipeline steps, or they may be the data first ingested into a pipeline via an ingestion step.

Definitions of the BaseArtifactStore class and the LocalArtifactStore that builds on it are in this module.

Other artifact stores corresponding to specific integrations are to be found in the integrations module. For example, the GCPArtifactStore, used when running ZenML on Google Cloud Platform, is defined in integrations.gcp.artifact_stores.

base_artifact_store

Definition of an Artifact Store

BaseArtifactStore (BaseComponent) pydantic-model

Base class for all ZenML Artifact Store.

Every ZenML Artifact Store should override this class.

Source code in zenml/artifact_stores/base_artifact_store.py

class BaseArtifactStore(BaseComponent):
    """Base class for all ZenML Artifact Store.

    Every ZenML Artifact Store should override this class.
    """

    path: str
    _ARTIFACT_STORE_DIR_NAME: str = "artifact_stores"

    def __init__(self, repo_path: str, **kwargs: Any) -> None:
        """Initializes a BaseArtifactStore instance.

        Args:
            repo_path: Path to the repository of this artifact store.
        """
        serialization_dir = os.path.join(
            get_zenml_config_dir(repo_path),
            self._ARTIFACT_STORE_DIR_NAME,
        )
        super().__init__(serialization_dir=serialization_dir, **kwargs)

    @staticmethod
    def get_component_name_from_uri(artifact_uri: str) -> str:
        """Gets component name from artifact URI.

        Args:
          artifact_uri: URI to artifact.

        Returns:
            Name of the component.
        """
        return fileio.get_grandparent(artifact_uri)

    def resolve_uri_locally(
        self, artifact_uri: str, path: Optional[str] = None
    ) -> str:
        """Takes a URI that points within the artifact store, downloads the
        URI locally, then returns local URI.

        Args:
          artifact_uri: uri to artifact.
          path: optional path to download to. If None, is inferred.

        Returns:
            Locally resolved uri.
        """
        if not fileio.is_remote(artifact_uri):
            # It's already local
            return artifact_uri

        if path is None:
            # Create a unique path in local machine
            path = os.path.join(
                GlobalConfig().get_serialization_dir(),
                str(self.uuid),
                BaseArtifactStore.get_component_name_from_uri(artifact_uri),
                fileio.get_parent(artifact_uri),  # unique ID from MLMD
            )

        # Create if not exists and download
        fileio.create_dir_recursive_if_not_exists(path)
        fileio.copy_dir(artifact_uri, path, overwrite=True)

        return path

    class Config:
        """Configuration of settings."""

        env_prefix = "zenml_artifact_store_"

Config

Configuration of settings.

Source code in zenml/artifact_stores/base_artifact_store.py

class Config:
    """Configuration of settings."""

    env_prefix = "zenml_artifact_store_"

__init__(self, repo_path, **kwargs) special

Initializes a BaseArtifactStore instance.

Parameters:

Name	Type	Description	Default
repo_path	str	Path to the repository of this artifact store.	required

Source code in zenml/artifact_stores/base_artifact_store.py

def __init__(self, repo_path: str, **kwargs: Any) -> None:
    """Initializes a BaseArtifactStore instance.

    Args:
        repo_path: Path to the repository of this artifact store.
    """
    serialization_dir = os.path.join(
        get_zenml_config_dir(repo_path),
        self._ARTIFACT_STORE_DIR_NAME,
    )
    super().__init__(serialization_dir=serialization_dir, **kwargs)

get_component_name_from_uri(artifact_uri) staticmethod

Gets component name from artifact URI.

Parameters:

Name	Type	Description	Default
artifact_uri	str	URI to artifact.	required

Returns:

Type	Description
str	Name of the component.

Source code in zenml/artifact_stores/base_artifact_store.py

@staticmethod
def get_component_name_from_uri(artifact_uri: str) -> str:
    """Gets component name from artifact URI.

    Args:
      artifact_uri: URI to artifact.

    Returns:
        Name of the component.
    """
    return fileio.get_grandparent(artifact_uri)

resolve_uri_locally(self, artifact_uri, path=None)

Takes a URI that points within the artifact store, downloads the URI locally, then returns local URI.

Parameters:

Name	Type	Description	Default
artifact_uri	str	uri to artifact.	required
path	Optional[str]	optional path to download to. If None, is inferred.	None

Returns:

Type	Description
str	Locally resolved uri.

Source code in zenml/artifact_stores/base_artifact_store.py

def resolve_uri_locally(
    self, artifact_uri: str, path: Optional[str] = None
) -> str:
    """Takes a URI that points within the artifact store, downloads the
    URI locally, then returns local URI.

    Args:
      artifact_uri: uri to artifact.
      path: optional path to download to. If None, is inferred.

    Returns:
        Locally resolved uri.
    """
    if not fileio.is_remote(artifact_uri):
        # It's already local
        return artifact_uri

    if path is None:
        # Create a unique path in local machine
        path = os.path.join(
            GlobalConfig().get_serialization_dir(),
            str(self.uuid),
            BaseArtifactStore.get_component_name_from_uri(artifact_uri),
            fileio.get_parent(artifact_uri),  # unique ID from MLMD
        )

    # Create if not exists and download
    fileio.create_dir_recursive_if_not_exists(path)
    fileio.copy_dir(artifact_uri, path, overwrite=True)

    return path

local_artifact_store

LocalArtifactStore (BaseArtifactStore) pydantic-model

Artifact Store for local artifacts.

Source code in zenml/artifact_stores/local_artifact_store.py

class LocalArtifactStore(BaseArtifactStore):
    """Artifact Store for local artifacts."""

    @validator("path")
    def must_be_local_path(cls, v: str) -> str:
        """Validates that the path is a local path."""
        if any([v.startswith(prefix) for prefix in REMOTE_FS_PREFIX]):
            raise ValueError("Must be a local path.")
        return v

must_be_local_path(v) classmethod

Validates that the path is a local path.

Source code in zenml/artifact_stores/local_artifact_store.py

@validator("path")
def must_be_local_path(cls, v: str) -> str:
    """Validates that the path is a local path."""
    if any([v.startswith(prefix) for prefix in REMOTE_FS_PREFIX]):
        raise ValueError("Must be a local path.")
    return v