Skip to content

Metadata

zenml.metadata special

Initialization of ZenML metadata.

ZenML metadata is any additional, dynamic information that is associated with your pipeline runs and artifacts at runtime.

lazy_load

Run Metadata Lazy Loader definition.

RunMetadataLazyGetter

Run Metadata Lazy Getter helper class.

It serves the purpose to feed back to the user the metadata lazy loader wrapper for any given key, if called inside a pipeline design time context.

Source code in zenml/metadata/lazy_load.py
class RunMetadataLazyGetter:
    """Run Metadata Lazy Getter helper class.

    It serves the purpose to feed back to the user the metadata
    lazy loader wrapper for any given key, if called inside a pipeline
    design time context.
    """

    def __init__(
        self,
        _lazy_load_model_name: str,
        _lazy_load_model_version: Optional[str],
        _lazy_load_artifact_name: Optional[str] = None,
        _lazy_load_artifact_version: Optional[str] = None,
    ):
        """Initialize a RunMetadataLazyGetter.

        Args:
            _lazy_load_model_name: The model name.
            _lazy_load_model_version: The model version.
            _lazy_load_artifact_name: The artifact name.
            _lazy_load_artifact_version: The artifact version.
        """
        self._lazy_load_model_name = _lazy_load_model_name
        self._lazy_load_model_version = _lazy_load_model_version
        self._lazy_load_artifact_name = _lazy_load_artifact_name
        self._lazy_load_artifact_version = _lazy_load_artifact_version

    def __getitem__(self, key: str) -> "RunMetadataResponse":
        """Get the metadata for the given key.

        Args:
            key: The metadata key.

        Returns:
            The metadata lazy loader wrapper for the given key.
        """
        from zenml.models.v2.core.run_metadata import LazyRunMetadataResponse

        return LazyRunMetadataResponse(
            lazy_load_model_name=self._lazy_load_model_name,
            lazy_load_model_version=self._lazy_load_model_version,
            lazy_load_artifact_name=self._lazy_load_artifact_name,
            lazy_load_artifact_version=self._lazy_load_artifact_version,
            lazy_load_metadata_name=key,
        )
__getitem__(self, key) special

Get the metadata for the given key.

Parameters:

Name Type Description Default
key str

The metadata key.

required

Returns:

Type Description
RunMetadataResponse

The metadata lazy loader wrapper for the given key.

Source code in zenml/metadata/lazy_load.py
def __getitem__(self, key: str) -> "RunMetadataResponse":
    """Get the metadata for the given key.

    Args:
        key: The metadata key.

    Returns:
        The metadata lazy loader wrapper for the given key.
    """
    from zenml.models.v2.core.run_metadata import LazyRunMetadataResponse

    return LazyRunMetadataResponse(
        lazy_load_model_name=self._lazy_load_model_name,
        lazy_load_model_version=self._lazy_load_model_version,
        lazy_load_artifact_name=self._lazy_load_artifact_name,
        lazy_load_artifact_version=self._lazy_load_artifact_version,
        lazy_load_metadata_name=key,
    )
__init__(self, _lazy_load_model_name, _lazy_load_model_version, _lazy_load_artifact_name=None, _lazy_load_artifact_version=None) special

Initialize a RunMetadataLazyGetter.

Parameters:

Name Type Description Default
_lazy_load_model_name str

The model name.

required
_lazy_load_model_version Optional[str]

The model version.

required
_lazy_load_artifact_name Optional[str]

The artifact name.

None
_lazy_load_artifact_version Optional[str]

The artifact version.

None
Source code in zenml/metadata/lazy_load.py
def __init__(
    self,
    _lazy_load_model_name: str,
    _lazy_load_model_version: Optional[str],
    _lazy_load_artifact_name: Optional[str] = None,
    _lazy_load_artifact_version: Optional[str] = None,
):
    """Initialize a RunMetadataLazyGetter.

    Args:
        _lazy_load_model_name: The model name.
        _lazy_load_model_version: The model version.
        _lazy_load_artifact_name: The artifact name.
        _lazy_load_artifact_version: The artifact version.
    """
    self._lazy_load_model_name = _lazy_load_model_name
    self._lazy_load_model_version = _lazy_load_model_version
    self._lazy_load_artifact_name = _lazy_load_artifact_name
    self._lazy_load_artifact_version = _lazy_load_artifact_version

metadata_types

Custom types that can be used as metadata of ZenML artifacts.

DType (str)

Special string class to indicate a data type.

Source code in zenml/metadata/metadata_types.py
class DType(str):
    """Special string class to indicate a data type."""

    @classmethod
    def __get_pydantic_core_schema__(
        cls, source_type: Any, handler: GetCoreSchemaHandler
    ) -> CoreSchema:
        """Additional method for pydantic to recognize it as a valid type.

        Args:
            source_type: the source type
            handler: the handler

        Returns:
            the schema for the custom type.
        """
        return core_schema.no_info_after_validator_function(cls, handler(str))
__get_pydantic_core_schema__(source_type, handler) classmethod special

Additional method for pydantic to recognize it as a valid type.

Parameters:

Name Type Description Default
source_type Any

the source type

required
handler GetCoreSchemaHandler

the handler

required

Returns:

Type Description
Union[pydantic_core.core_schema.AnySchema, pydantic_core.core_schema.NoneSchema, pydantic_core.core_schema.BoolSchema, pydantic_core.core_schema.IntSchema, pydantic_core.core_schema.FloatSchema, pydantic_core.core_schema.DecimalSchema, pydantic_core.core_schema.StringSchema, pydantic_core.core_schema.BytesSchema, pydantic_core.core_schema.DateSchema, pydantic_core.core_schema.TimeSchema, pydantic_core.core_schema.DatetimeSchema, pydantic_core.core_schema.TimedeltaSchema, pydantic_core.core_schema.LiteralSchema, pydantic_core.core_schema.EnumSchema, pydantic_core.core_schema.IsInstanceSchema, pydantic_core.core_schema.IsSubclassSchema, pydantic_core.core_schema.CallableSchema, pydantic_core.core_schema.ListSchema, pydantic_core.core_schema.TupleSchema, pydantic_core.core_schema.SetSchema, pydantic_core.core_schema.FrozenSetSchema, pydantic_core.core_schema.GeneratorSchema, pydantic_core.core_schema.DictSchema, pydantic_core.core_schema.AfterValidatorFunctionSchema, pydantic_core.core_schema.BeforeValidatorFunctionSchema, pydantic_core.core_schema.WrapValidatorFunctionSchema, pydantic_core.core_schema.PlainValidatorFunctionSchema, pydantic_core.core_schema.WithDefaultSchema, pydantic_core.core_schema.NullableSchema, pydantic_core.core_schema.UnionSchema, pydantic_core.core_schema.TaggedUnionSchema, pydantic_core.core_schema.ChainSchema, pydantic_core.core_schema.LaxOrStrictSchema, pydantic_core.core_schema.JsonOrPythonSchema, pydantic_core.core_schema.TypedDictSchema, pydantic_core.core_schema.ModelFieldsSchema, pydantic_core.core_schema.ModelSchema, pydantic_core.core_schema.DataclassArgsSchema, pydantic_core.core_schema.DataclassSchema, pydantic_core.core_schema.ArgumentsSchema, pydantic_core.core_schema.CallSchema, pydantic_core.core_schema.CustomErrorSchema, pydantic_core.core_schema.JsonSchema, pydantic_core.core_schema.UrlSchema, pydantic_core.core_schema.MultiHostUrlSchema, pydantic_core.core_schema.DefinitionsSchema, pydantic_core.core_schema.DefinitionReferenceSchema, pydantic_core.core_schema.UuidSchema]

the schema for the custom type.

Source code in zenml/metadata/metadata_types.py
@classmethod
def __get_pydantic_core_schema__(
    cls, source_type: Any, handler: GetCoreSchemaHandler
) -> CoreSchema:
    """Additional method for pydantic to recognize it as a valid type.

    Args:
        source_type: the source type
        handler: the handler

    Returns:
        the schema for the custom type.
    """
    return core_schema.no_info_after_validator_function(cls, handler(str))

MetadataTypeEnum (StrEnum)

String Enum of all possible types that metadata can have.

Source code in zenml/metadata/metadata_types.py
class MetadataTypeEnum(StrEnum):
    """String Enum of all possible types that metadata can have."""

    STRING = "str"
    INT = "int"
    FLOAT = "float"
    BOOL = "bool"
    LIST = "list"
    DICT = "dict"
    TUPLE = "tuple"
    SET = "set"
    URI = "Uri"
    PATH = "Path"
    DTYPE = "DType"
    STORAGE_SIZE = "StorageSize"

Path (str)

Special string class to indicate a path.

Source code in zenml/metadata/metadata_types.py
class Path(str):
    """Special string class to indicate a path."""

    @classmethod
    def __get_pydantic_core_schema__(
        cls, source_type: Any, handler: GetCoreSchemaHandler
    ) -> CoreSchema:
        """Additional method for pydantic to recognize it as a valid type.

        Args:
            source_type: the source type
            handler: the handler

        Returns:
            the schema for the custom type.
        """
        return core_schema.no_info_after_validator_function(cls, handler(str))
__get_pydantic_core_schema__(source_type, handler) classmethod special

Additional method for pydantic to recognize it as a valid type.

Parameters:

Name Type Description Default
source_type Any

the source type

required
handler GetCoreSchemaHandler

the handler

required

Returns:

Type Description
Union[pydantic_core.core_schema.AnySchema, pydantic_core.core_schema.NoneSchema, pydantic_core.core_schema.BoolSchema, pydantic_core.core_schema.IntSchema, pydantic_core.core_schema.FloatSchema, pydantic_core.core_schema.DecimalSchema, pydantic_core.core_schema.StringSchema, pydantic_core.core_schema.BytesSchema, pydantic_core.core_schema.DateSchema, pydantic_core.core_schema.TimeSchema, pydantic_core.core_schema.DatetimeSchema, pydantic_core.core_schema.TimedeltaSchema, pydantic_core.core_schema.LiteralSchema, pydantic_core.core_schema.EnumSchema, pydantic_core.core_schema.IsInstanceSchema, pydantic_core.core_schema.IsSubclassSchema, pydantic_core.core_schema.CallableSchema, pydantic_core.core_schema.ListSchema, pydantic_core.core_schema.TupleSchema, pydantic_core.core_schema.SetSchema, pydantic_core.core_schema.FrozenSetSchema, pydantic_core.core_schema.GeneratorSchema, pydantic_core.core_schema.DictSchema, pydantic_core.core_schema.AfterValidatorFunctionSchema, pydantic_core.core_schema.BeforeValidatorFunctionSchema, pydantic_core.core_schema.WrapValidatorFunctionSchema, pydantic_core.core_schema.PlainValidatorFunctionSchema, pydantic_core.core_schema.WithDefaultSchema, pydantic_core.core_schema.NullableSchema, pydantic_core.core_schema.UnionSchema, pydantic_core.core_schema.TaggedUnionSchema, pydantic_core.core_schema.ChainSchema, pydantic_core.core_schema.LaxOrStrictSchema, pydantic_core.core_schema.JsonOrPythonSchema, pydantic_core.core_schema.TypedDictSchema, pydantic_core.core_schema.ModelFieldsSchema, pydantic_core.core_schema.ModelSchema, pydantic_core.core_schema.DataclassArgsSchema, pydantic_core.core_schema.DataclassSchema, pydantic_core.core_schema.ArgumentsSchema, pydantic_core.core_schema.CallSchema, pydantic_core.core_schema.CustomErrorSchema, pydantic_core.core_schema.JsonSchema, pydantic_core.core_schema.UrlSchema, pydantic_core.core_schema.MultiHostUrlSchema, pydantic_core.core_schema.DefinitionsSchema, pydantic_core.core_schema.DefinitionReferenceSchema, pydantic_core.core_schema.UuidSchema]

the schema for the custom type.

Source code in zenml/metadata/metadata_types.py
@classmethod
def __get_pydantic_core_schema__(
    cls, source_type: Any, handler: GetCoreSchemaHandler
) -> CoreSchema:
    """Additional method for pydantic to recognize it as a valid type.

    Args:
        source_type: the source type
        handler: the handler

    Returns:
        the schema for the custom type.
    """
    return core_schema.no_info_after_validator_function(cls, handler(str))

StorageSize (int)

Special int class to indicate the storage size in number of bytes.

Source code in zenml/metadata/metadata_types.py
class StorageSize(int):
    """Special int class to indicate the storage size in number of bytes."""

    @classmethod
    def __get_pydantic_core_schema__(
        cls, source_type: Any, handler: GetCoreSchemaHandler
    ) -> CoreSchema:
        """Additional method for pydantic to recognize it as a valid type.

        Args:
            source_type: the source type
            handler: the handler

        Returns:
            the schema for the custom type.
        """
        return core_schema.no_info_after_validator_function(cls, handler(int))
__get_pydantic_core_schema__(source_type, handler) classmethod special

Additional method for pydantic to recognize it as a valid type.

Parameters:

Name Type Description Default
source_type Any

the source type

required
handler GetCoreSchemaHandler

the handler

required

Returns:

Type Description
Union[pydantic_core.core_schema.AnySchema, pydantic_core.core_schema.NoneSchema, pydantic_core.core_schema.BoolSchema, pydantic_core.core_schema.IntSchema, pydantic_core.core_schema.FloatSchema, pydantic_core.core_schema.DecimalSchema, pydantic_core.core_schema.StringSchema, pydantic_core.core_schema.BytesSchema, pydantic_core.core_schema.DateSchema, pydantic_core.core_schema.TimeSchema, pydantic_core.core_schema.DatetimeSchema, pydantic_core.core_schema.TimedeltaSchema, pydantic_core.core_schema.LiteralSchema, pydantic_core.core_schema.EnumSchema, pydantic_core.core_schema.IsInstanceSchema, pydantic_core.core_schema.IsSubclassSchema, pydantic_core.core_schema.CallableSchema, pydantic_core.core_schema.ListSchema, pydantic_core.core_schema.TupleSchema, pydantic_core.core_schema.SetSchema, pydantic_core.core_schema.FrozenSetSchema, pydantic_core.core_schema.GeneratorSchema, pydantic_core.core_schema.DictSchema, pydantic_core.core_schema.AfterValidatorFunctionSchema, pydantic_core.core_schema.BeforeValidatorFunctionSchema, pydantic_core.core_schema.WrapValidatorFunctionSchema, pydantic_core.core_schema.PlainValidatorFunctionSchema, pydantic_core.core_schema.WithDefaultSchema, pydantic_core.core_schema.NullableSchema, pydantic_core.core_schema.UnionSchema, pydantic_core.core_schema.TaggedUnionSchema, pydantic_core.core_schema.ChainSchema, pydantic_core.core_schema.LaxOrStrictSchema, pydantic_core.core_schema.JsonOrPythonSchema, pydantic_core.core_schema.TypedDictSchema, pydantic_core.core_schema.ModelFieldsSchema, pydantic_core.core_schema.ModelSchema, pydantic_core.core_schema.DataclassArgsSchema, pydantic_core.core_schema.DataclassSchema, pydantic_core.core_schema.ArgumentsSchema, pydantic_core.core_schema.CallSchema, pydantic_core.core_schema.CustomErrorSchema, pydantic_core.core_schema.JsonSchema, pydantic_core.core_schema.UrlSchema, pydantic_core.core_schema.MultiHostUrlSchema, pydantic_core.core_schema.DefinitionsSchema, pydantic_core.core_schema.DefinitionReferenceSchema, pydantic_core.core_schema.UuidSchema]

the schema for the custom type.

Source code in zenml/metadata/metadata_types.py
@classmethod
def __get_pydantic_core_schema__(
    cls, source_type: Any, handler: GetCoreSchemaHandler
) -> CoreSchema:
    """Additional method for pydantic to recognize it as a valid type.

    Args:
        source_type: the source type
        handler: the handler

    Returns:
        the schema for the custom type.
    """
    return core_schema.no_info_after_validator_function(cls, handler(int))

Uri (str)

Special string class to indicate a URI.

Source code in zenml/metadata/metadata_types.py
class Uri(str):
    """Special string class to indicate a URI."""

    @classmethod
    def __get_pydantic_core_schema__(
        cls, source_type: Any, handler: GetCoreSchemaHandler
    ) -> CoreSchema:
        """Additional method for pydantic to recognize it as a valid type.

        Args:
            source_type: the source type
            handler: the handler

        Returns:
            the schema for the custom type.
        """
        return core_schema.no_info_after_validator_function(cls, handler(str))
__get_pydantic_core_schema__(source_type, handler) classmethod special

Additional method for pydantic to recognize it as a valid type.

Parameters:

Name Type Description Default
source_type Any

the source type

required
handler GetCoreSchemaHandler

the handler

required

Returns:

Type Description
Union[pydantic_core.core_schema.AnySchema, pydantic_core.core_schema.NoneSchema, pydantic_core.core_schema.BoolSchema, pydantic_core.core_schema.IntSchema, pydantic_core.core_schema.FloatSchema, pydantic_core.core_schema.DecimalSchema, pydantic_core.core_schema.StringSchema, pydantic_core.core_schema.BytesSchema, pydantic_core.core_schema.DateSchema, pydantic_core.core_schema.TimeSchema, pydantic_core.core_schema.DatetimeSchema, pydantic_core.core_schema.TimedeltaSchema, pydantic_core.core_schema.LiteralSchema, pydantic_core.core_schema.EnumSchema, pydantic_core.core_schema.IsInstanceSchema, pydantic_core.core_schema.IsSubclassSchema, pydantic_core.core_schema.CallableSchema, pydantic_core.core_schema.ListSchema, pydantic_core.core_schema.TupleSchema, pydantic_core.core_schema.SetSchema, pydantic_core.core_schema.FrozenSetSchema, pydantic_core.core_schema.GeneratorSchema, pydantic_core.core_schema.DictSchema, pydantic_core.core_schema.AfterValidatorFunctionSchema, pydantic_core.core_schema.BeforeValidatorFunctionSchema, pydantic_core.core_schema.WrapValidatorFunctionSchema, pydantic_core.core_schema.PlainValidatorFunctionSchema, pydantic_core.core_schema.WithDefaultSchema, pydantic_core.core_schema.NullableSchema, pydantic_core.core_schema.UnionSchema, pydantic_core.core_schema.TaggedUnionSchema, pydantic_core.core_schema.ChainSchema, pydantic_core.core_schema.LaxOrStrictSchema, pydantic_core.core_schema.JsonOrPythonSchema, pydantic_core.core_schema.TypedDictSchema, pydantic_core.core_schema.ModelFieldsSchema, pydantic_core.core_schema.ModelSchema, pydantic_core.core_schema.DataclassArgsSchema, pydantic_core.core_schema.DataclassSchema, pydantic_core.core_schema.ArgumentsSchema, pydantic_core.core_schema.CallSchema, pydantic_core.core_schema.CustomErrorSchema, pydantic_core.core_schema.JsonSchema, pydantic_core.core_schema.UrlSchema, pydantic_core.core_schema.MultiHostUrlSchema, pydantic_core.core_schema.DefinitionsSchema, pydantic_core.core_schema.DefinitionReferenceSchema, pydantic_core.core_schema.UuidSchema]

the schema for the custom type.

Source code in zenml/metadata/metadata_types.py
@classmethod
def __get_pydantic_core_schema__(
    cls, source_type: Any, handler: GetCoreSchemaHandler
) -> CoreSchema:
    """Additional method for pydantic to recognize it as a valid type.

    Args:
        source_type: the source type
        handler: the handler

    Returns:
        the schema for the custom type.
    """
    return core_schema.no_info_after_validator_function(cls, handler(str))

cast_to_metadata_type(value, type_)

Cast an object to a metadata type.

Parameters:

Name Type Description Default
value object

The object to cast.

required
type_ MetadataTypeEnum

The metadata type to cast to.

required

Returns:

Type Description
Union[str, int, float, bool, Dict[Any, Any], List[Any], Set[Any], Tuple[Any, ...], zenml.metadata.metadata_types.Uri, zenml.metadata.metadata_types.Path, zenml.metadata.metadata_types.DType, zenml.metadata.metadata_types.StorageSize]

The value cast to the given metadata type.

Source code in zenml/metadata/metadata_types.py
def cast_to_metadata_type(
    value: object,
    type_: MetadataTypeEnum,
) -> MetadataType:
    """Cast an object to a metadata type.

    Args:
        value: The object to cast.
        type_: The metadata type to cast to.

    Returns:
        The value cast to the given metadata type.
    """
    metadata_type = metadata_enum_to_type_mapping[type_]
    typed_value = metadata_type(value)
    return typed_value  # type: ignore[no-any-return]

get_metadata_type(object_)

Get the metadata type enum for a given object.

Parameters:

Name Type Description Default
object_ object

The object to get the metadata type for.

required

Returns:

Type Description
MetadataTypeEnum

The corresponding metadata type enum.

Exceptions:

Type Description
ValueError

If the metadata type is not supported.

Source code in zenml/metadata/metadata_types.py
def get_metadata_type(
    object_: object,
) -> MetadataTypeEnum:
    """Get the metadata type enum for a given object.

    Args:
        object_: The object to get the metadata type for.

    Returns:
        The corresponding metadata type enum.

    Raises:
        ValueError: If the metadata type is not supported.
    """
    metadata_type = type(object_)
    if metadata_type in metadata_type_to_enum_mapping:
        return metadata_type_to_enum_mapping[metadata_type]
    raise ValueError(f"Metadata type {metadata_type} is not supported.")