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: "Model",
        _lazy_load_artifact_name: Optional[str],
        _lazy_load_artifact_version: Optional[str],
    ):
        """Initialize a RunMetadataLazyGetter.

        Args:
            _lazy_load_model: The model version.
            _lazy_load_artifact_name: The artifact name.
            _lazy_load_artifact_version: The artifact version.
        """
        self._lazy_load_model = _lazy_load_model
        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=self._lazy_load_model,
            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=self._lazy_load_model,
        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, _lazy_load_artifact_name, _lazy_load_artifact_version) special

Initialize a RunMetadataLazyGetter.

Parameters:

Name	Type	Description	Default
_lazy_load_model	Model	The model version.	required
_lazy_load_artifact_name	Optional[str]	The artifact name.	required
_lazy_load_artifact_version	Optional[str]	The artifact version.	required

Source code in zenml/metadata/lazy_load.py

def __init__(
    self,
    _lazy_load_model: "Model",
    _lazy_load_artifact_name: Optional[str],
    _lazy_load_artifact_version: Optional[str],
):
    """Initialize a RunMetadataLazyGetter.

    Args:
        _lazy_load_model: The model version.
        _lazy_load_artifact_name: The artifact name.
        _lazy_load_artifact_version: The artifact version.
    """
    self._lazy_load_model = _lazy_load_model
    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.")