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.

LazyRunMetadataResponse (BaseModel)

Lazy run metadata response.

Used if the run metadata is accessed from the model in a pipeline context available only during pipeline compilation.

Source code in zenml/metadata/lazy_load.py 
class LazyRunMetadataResponse(BaseModel):
    """Lazy run metadata response.

    Used if the run metadata is accessed from the model in
    a pipeline context available only during pipeline compilation.
    """

    lazy_load_artifact_name: Optional[str] = None
    lazy_load_artifact_version: Optional[str] = None
    lazy_load_metadata_name: Optional[str] = None
    lazy_load_model_name: str
    lazy_load_model_version: Optional[str] = None

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) -> MetadataType:
        """Get the metadata for the given key.

        Args:
            key: The metadata key.

        Returns:
            The metadata lazy loader wrapper for the given key.
        """
        return LazyRunMetadataResponse(  # type: ignore[return-value]
            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
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 metadata lazy loader wrapper for the given key.
Source code in zenml/metadata/lazy_load.py 
def __getitem__(self, key: str) -> MetadataType:
    """Get the metadata for the given key.

    Args:
        key: The metadata key.

    Returns:
        The metadata lazy loader wrapper for the given key.
    """
    return LazyRunMetadataResponse(  # type: ignore[return-value]
        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.")

validate_metadata(metadata)

Validate metadata.

This function excludes and warns about metadata values that are too long or of an unsupported type.

Parameters:

Name Type Description Default
metadata Dict[str, 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 metadata to validate.

 required

Returns:

Type Description
Dict[str, 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 validated metadata.
Source code in zenml/metadata/metadata_types.py 
def validate_metadata(
    metadata: Dict[str, MetadataType],
) -> Dict[str, MetadataType]:
    """Validate metadata.

    This function excludes and warns about metadata values that are too long
    or of an unsupported type.

    Args:
        metadata: The metadata to validate.

    Returns:
        The validated metadata.
    """
    validated_metadata = {}

    for key, value in metadata.items():
        if len(key) > STR_FIELD_MAX_LENGTH:
            logger.warning(
                f"Metadata key '{key}' is too large to be "
                "stored in the database. Skipping."
            )
            continue

        if len(json.dumps(value)) > TEXT_FIELD_MAX_LENGTH:
            logger.warning(
                f"Metadata value for key '{key}' is too large to be "
                "stored in the database. Skipping."
            )
            continue

        try:
            get_metadata_type(value)
        except ValueError as e:
            logger.warning(
                f"Metadata value for key '{key}' is not of a supported "
                f"type. Skipping. Full error: {e}"
            )

        validated_metadata[key] = value

    return validated_metadata