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.")

Metadata

zenml.metadata special

lazy_load

RunMetadataLazyGetter

__getitem__(self, key) special

__init__(self, _lazy_load_model_name, _lazy_load_model_version, _lazy_load_artifact_name=None, _lazy_load_artifact_version=None) special

metadata_types

DType (str)

__get_pydantic_core_schema__(source_type, handler) classmethod special

MetadataTypeEnum (StrEnum)

Path (str)

__get_pydantic_core_schema__(source_type, handler) classmethod special

StorageSize (int)

__get_pydantic_core_schema__(source_type, handler) classmethod special

Uri (str)

__get_pydantic_core_schema__(source_type, handler) classmethod special

cast_to_metadata_type(value, type_)

get_metadata_type(object_)

`zenml.metadata` `special`

`lazy_load`

`RunMetadataLazyGetter`

`getitem(self, key)` `special`

`init(self, _lazy_load_model_name, _lazy_load_model_version, _lazy_load_artifact_name=None, _lazy_load_artifact_version=None)` `special`

`metadata_types`

`DType (str)`

`__get_pydantic_core_schema__(source_type, handler)` `classmethod` `special`

`MetadataTypeEnum (StrEnum)`

`Path (str)`

`__get_pydantic_core_schema__(source_type, handler)` `classmethod` `special`

`StorageSize (int)`

`__get_pydantic_core_schema__(source_type, handler)` `classmethod` `special`

`Uri (str)`

`__get_pydantic_core_schema__(source_type, handler)` `classmethod` `special`

`cast_to_metadata_type(value, type_)`

`get_metadata_type(object_)`