Exceptions

zenml.exceptions

ZenML specific exception definitions.

ArtifactInterfaceError (ZenMLBaseException)

Raises exception when interacting with the Artifact interface in an unsupported way.

Source code in zenml/exceptions.py

class ArtifactInterfaceError(ZenMLBaseException):
    """Raises exception when interacting with the Artifact interface in an unsupported way."""

ArtifactStoreInterfaceError (ZenMLBaseException)

Raises exception when interacting with the Artifact Store interface in an unsupported way.

Source code in zenml/exceptions.py

class ArtifactStoreInterfaceError(ZenMLBaseException):
    """Raises exception when interacting with the Artifact Store interface in an unsupported way."""

AuthorizationException (ZenMLBaseException)

Raised when an authorization error occurred while trying to access a ZenML resource .

Source code in zenml/exceptions.py

class AuthorizationException(ZenMLBaseException):
    """Raised when an authorization error occurred while trying to access a ZenML resource ."""

DoesNotExistException (ZenMLBaseException)

Raises exception when the entity does not exist in the system but an action is being done that requires it to be present.

Source code in zenml/exceptions.py

class DoesNotExistException(ZenMLBaseException):
    """Raises exception when the entity does not exist in the system but an action is being done that requires it to be present."""

    def __init__(self, message: str):
        """Initializes the exception.

        Args:
            message: Message with details of exception.
        """
        super().__init__(message)

__init__(self, message) special

Initializes the exception.

Parameters:

Name	Type	Description	Default
message	str	Message with details of exception.	required

Source code in zenml/exceptions.py

def __init__(self, message: str):
    """Initializes the exception.

    Args:
        message: Message with details of exception.
    """
    super().__init__(message)

DuplicateRunNameError (RuntimeError)

Raises exception when a run with the same name already exists.

Source code in zenml/exceptions.py

class DuplicateRunNameError(RuntimeError):
    """Raises exception when a run with the same name already exists."""

    def __init__(
        self,
        message: str = "Unable to run a pipeline with a run name that "
        "already exists.",
    ):
        """Initializes the exception.

        Args:
            message: Message with details of exception.
        """
        super().__init__(message)

__init__(self, message='Unable to run a pipeline with a run name that already exists.') special

Initializes the exception.

Parameters:

Name	Type	Description	Default
message	str	Message with details of exception.	'Unable to run a pipeline with a run name that already exists.'

Source code in zenml/exceptions.py

def __init__(
    self,
    message: str = "Unable to run a pipeline with a run name that "
    "already exists.",
):
    """Initializes the exception.

    Args:
        message: Message with details of exception.
    """
    super().__init__(message)

DuplicatedConfigurationError (ZenMLBaseException)

Raised when a configuration parameter is set twice.

Source code in zenml/exceptions.py

class DuplicatedConfigurationError(ZenMLBaseException):
    """Raised when a configuration parameter is set twice."""

EntityExistsError (ZenMLBaseException)

Raised when trying to register an entity that already exists.

Source code in zenml/exceptions.py

class EntityExistsError(ZenMLBaseException):
    """Raised when trying to register an entity that already exists."""

GitException (ZenMLBaseException)

Raises exception when a problem occurs in git resolution.

Source code in zenml/exceptions.py

class GitException(ZenMLBaseException):
    """Raises exception when a problem occurs in git resolution."""

    def __init__(
        self,
        message: str = "There is a problem with git resolution. "
        "Please make sure that all relevant files "
        "are committed.",
    ):
        """Initializes the exception.

        Args:
            message: Message with details of exception.
        """
        super().__init__(message)

__init__(self, message='There is a problem with git resolution. Please make sure that all relevant files are committed.') special

Initializes the exception.

Parameters:

Name	Type	Description	Default
message	str	Message with details of exception.	'There is a problem with git resolution. Please make sure that all relevant files are committed.'

Source code in zenml/exceptions.py

def __init__(
    self,
    message: str = "There is a problem with git resolution. "
    "Please make sure that all relevant files "
    "are committed.",
):
    """Initializes the exception.

    Args:
        message: Message with details of exception.
    """
    super().__init__(message)

GitNotFoundError (ImportError)

Raised when ZenML CLI is used to interact with examples on a machine with no git installation.

Source code in zenml/exceptions.py

class GitNotFoundError(ImportError):
    """Raised when ZenML CLI is used to interact with examples on a machine with no git installation."""

IllegalOperationError (ZenMLBaseException)

Raised when an illegal operation is attempted.

Source code in zenml/exceptions.py

class IllegalOperationError(ZenMLBaseException):
    """Raised when an illegal operation is attempted."""

InitializationException (ZenMLBaseException)

Raised when an error occurred during initialization of a ZenML repository.

Source code in zenml/exceptions.py

class InitializationException(ZenMLBaseException):
    """Raised when an error occurred during initialization of a ZenML repository."""

InputResolutionError (ZenMLBaseException)

Raised when step input resolving failed.

Source code in zenml/exceptions.py

class InputResolutionError(ZenMLBaseException):
    """Raised when step input resolving failed."""

IntegrationError (ZenMLBaseException)

Raises exceptions when a requested integration can not be activated.

Source code in zenml/exceptions.py

class IntegrationError(ZenMLBaseException):
    """Raises exceptions when a requested integration can not be activated."""

MaterializerInterfaceError (ZenMLBaseException)

Raises exception when interacting with the Materializer interface in an unsupported way.

Source code in zenml/exceptions.py

class MaterializerInterfaceError(ZenMLBaseException):
    """Raises exception when interacting with the Materializer interface in an unsupported way."""

MissingStepParameterError (ZenMLBaseException)

Raises exceptions when a step parameter is missing when running a pipeline.

Source code in zenml/exceptions.py

class MissingStepParameterError(ZenMLBaseException):
    """Raises exceptions when a step parameter is missing when running a pipeline."""

    def __init__(
        self,
        step_name: str,
        missing_parameters: List[str],
        parameters_class: Type["BaseParameters"],
    ):
        """Initializes a MissingStepParameterError object.

        Args:
            step_name: Name of the step for which one or more parameters
                are missing.
            missing_parameters: Names of all parameters which are missing.
            parameters_class: Class of the parameters object for which
                the parameters are missing.
        """
        import textwrap

        message = textwrap.fill(
            textwrap.dedent(
                f"""
            Missing parameters {missing_parameters} for '{step_name}' step.
            There are three ways to solve this issue:
            (1) Specify a default value in the parameters class
            `{parameters_class.__name__}`
            (2) Specify the parameters in code when creating the pipeline:
            `my_pipeline({step_name}(params={parameters_class.__name__}(...))`
            (3) Specify the parameters in a yaml configuration file and pass
            it to the pipeline: `my_pipeline(...).run(config_path='path_to_yaml')`
            """
            )
        )
        super().__init__(message)

__init__(self, step_name, missing_parameters, parameters_class) special

Initializes a MissingStepParameterError object.

Parameters:

Name	Type	Description	Default
step_name	str	Name of the step for which one or more parameters are missing.	required
missing_parameters	List[str]	Names of all parameters which are missing.	required
parameters_class	Type[BaseParameters]	Class of the parameters object for which the parameters are missing.	required

Source code in zenml/exceptions.py

def __init__(
    self,
    step_name: str,
    missing_parameters: List[str],
    parameters_class: Type["BaseParameters"],
):
    """Initializes a MissingStepParameterError object.

    Args:
        step_name: Name of the step for which one or more parameters
            are missing.
        missing_parameters: Names of all parameters which are missing.
        parameters_class: Class of the parameters object for which
            the parameters are missing.
    """
    import textwrap

    message = textwrap.fill(
        textwrap.dedent(
            f"""
        Missing parameters {missing_parameters} for '{step_name}' step.
        There are three ways to solve this issue:
        (1) Specify a default value in the parameters class
        `{parameters_class.__name__}`
        (2) Specify the parameters in code when creating the pipeline:
        `my_pipeline({step_name}(params={parameters_class.__name__}(...))`
        (3) Specify the parameters in a yaml configuration file and pass
        it to the pipeline: `my_pipeline(...).run(config_path='path_to_yaml')`
        """
        )
    )
    super().__init__(message)

OAuthError (ValueError)

OAuth2 error.

Source code in zenml/exceptions.py

class OAuthError(ValueError):
    """OAuth2 error."""

    def __init__(
        self,
        error: str,
        status_code: int = 400,
        error_description: Optional[str] = None,
        error_uri: Optional[str] = None,
    ) -> None:
        """Initializes the OAuthError.

        Args:
            status_code: HTTP status code.
            error: Error code.
            error_description: Error description.
            error_uri: Error URI.
        """
        self.status_code = status_code
        self.error = error
        self.error_description = error_description
        self.error_uri = error_uri

    def to_dict(self) -> Dict[str, Optional[str]]:
        """Returns the OAuthError as a dictionary.

        Returns:
            The OAuthError as a dictionary.
        """
        return {
            "error": self.error,
            "error_description": self.error_description,
            "error_uri": self.error_uri,
        }

    def __str__(self) -> str:
        """String function.

        Returns:
            the error message
        """
        return f"{self.error}: {self.error_description or ''}"

__init__(self, error, status_code=400, error_description=None, error_uri=None) special

Initializes the OAuthError.

Parameters:

Name	Type	Description	Default
status_code	int	HTTP status code.	400
error	str	Error code.	required
error_description	Optional[str]	Error description.	None
error_uri	Optional[str]	Error URI.	None

Source code in zenml/exceptions.py

def __init__(
    self,
    error: str,
    status_code: int = 400,
    error_description: Optional[str] = None,
    error_uri: Optional[str] = None,
) -> None:
    """Initializes the OAuthError.

    Args:
        status_code: HTTP status code.
        error: Error code.
        error_description: Error description.
        error_uri: Error URI.
    """
    self.status_code = status_code
    self.error = error
    self.error_description = error_description
    self.error_uri = error_uri

__str__(self) special

String function.

Returns:

Type	Description
str	the error message

Source code in zenml/exceptions.py

def __str__(self) -> str:
    """String function.

    Returns:
        the error message
    """
    return f"{self.error}: {self.error_description or ''}"

to_dict(self)

Returns the OAuthError as a dictionary.

Returns:

Type	Description
Dict[str, Optional[str]]	The OAuthError as a dictionary.

Source code in zenml/exceptions.py

def to_dict(self) -> Dict[str, Optional[str]]:
    """Returns the OAuthError as a dictionary.

    Returns:
        The OAuthError as a dictionary.
    """
    return {
        "error": self.error,
        "error_description": self.error_description,
        "error_uri": self.error_uri,
    }

PipelineConfigurationError (ZenMLBaseException)

Raises exceptions when a pipeline configuration contains invalid values.

Source code in zenml/exceptions.py

class PipelineConfigurationError(ZenMLBaseException):
    """Raises exceptions when a pipeline configuration contains invalid values."""

PipelineInterfaceError (ZenMLBaseException)

Raises exception when interacting with the Pipeline interface in an unsupported way.

Source code in zenml/exceptions.py

class PipelineInterfaceError(ZenMLBaseException):
    """Raises exception when interacting with the Pipeline interface in an unsupported way."""

PipelineNotSucceededException (ZenMLBaseException)

Raises exception when trying to fetch artifacts from a not succeeded pipeline.

Source code in zenml/exceptions.py

class PipelineNotSucceededException(ZenMLBaseException):
    """Raises exception when trying to fetch artifacts from a not succeeded pipeline."""

    def __init__(
        self,
        name: str = "",
        message: str = "{} is not yet completed successfully.",
    ):
        """Initializes the exception.

        Args:
            name: Name of the pipeline.
            message: Message with details of exception.
        """
        super().__init__(message.format(name))

__init__(self, name='', message='{} is not yet completed successfully.') special

Initializes the exception.

Parameters:

Name	Type	Description	Default
name	str	Name of the pipeline.	''
message	str	Message with details of exception.	'{} is not yet completed successfully.'

Source code in zenml/exceptions.py

def __init__(
    self,
    name: str = "",
    message: str = "{} is not yet completed successfully.",
):
    """Initializes the exception.

    Args:
        name: Name of the pipeline.
        message: Message with details of exception.
    """
    super().__init__(message.format(name))

ProvisioningError (ZenMLBaseException)

Raised when an error occurs when provisioning resources for a StackComponent.

Source code in zenml/exceptions.py

class ProvisioningError(ZenMLBaseException):
    """Raised when an error occurs when provisioning resources for a StackComponent."""

SecretExistsError (EntityExistsError)

Raised when trying to register a secret with existing name.

Source code in zenml/exceptions.py

class SecretExistsError(EntityExistsError):
    """Raised when trying to register a secret with existing name."""

SettingsResolvingError (ZenMLBaseException)

Raised when resolving settings failed.

Source code in zenml/exceptions.py

class SettingsResolvingError(ZenMLBaseException):
    """Raised when resolving settings failed."""

StackComponentDeploymentError (ZenMLBaseException)

Raises exception when deploying a stack component fails.

Source code in zenml/exceptions.py

class StackComponentDeploymentError(ZenMLBaseException):
    """Raises exception when deploying a stack component fails."""

StackComponentExistsError (EntityExistsError)

Raised when trying to register a stack component with existing name.

Source code in zenml/exceptions.py

class StackComponentExistsError(EntityExistsError):
    """Raised when trying to register a stack component with existing name."""

StackComponentInterfaceError (ZenMLBaseException)

Raises exception when interacting with the stack components in an unsupported way.

Source code in zenml/exceptions.py

class StackComponentInterfaceError(ZenMLBaseException):
    """Raises exception when interacting with the stack components in an unsupported way."""

StackComponentValidationError (ZenMLBaseException)

Raised when a stack component configuration is not valid.

Source code in zenml/exceptions.py

class StackComponentValidationError(ZenMLBaseException):
    """Raised when a stack component configuration is not valid."""

StackExistsError (EntityExistsError)

Raised when trying to register a stack with name that already exists.

Source code in zenml/exceptions.py

class StackExistsError(EntityExistsError):
    """Raised when trying to register a stack with name that already exists."""

StackValidationError (ZenMLBaseException)

Raised when a stack configuration is not valid.

Source code in zenml/exceptions.py

class StackValidationError(ZenMLBaseException):
    """Raised when a stack configuration is not valid."""

StepContextError (ZenMLBaseException)

Raises exception when interacting with a StepContext in an unsupported way.

Source code in zenml/exceptions.py

class StepContextError(ZenMLBaseException):
    """Raises exception when interacting with a StepContext in an unsupported way."""

StepInterfaceError (ZenMLBaseException)

Raises exception when interacting with the Step interface in an unsupported way.

Source code in zenml/exceptions.py

class StepInterfaceError(ZenMLBaseException):
    """Raises exception when interacting with the Step interface in an unsupported way."""

ValidationError (ZenMLBaseException)

Raised when the Model passed to the ZenStore.

Source code in zenml/exceptions.py

class ValidationError(ZenMLBaseException):
    """Raised when the Model passed to the ZenStore."""

ZenKeyError (KeyError)

Specialized key error which allows error messages with line breaks.

Source code in zenml/exceptions.py

class ZenKeyError(KeyError):
    """Specialized key error which allows error messages with line breaks."""

    def __init__(self, message: str) -> None:
        """Initialization.

        Args:
            message:str, the error message
        """
        self.message = message

    def __str__(self) -> str:
        """String function.

        Returns:
            the error message
        """
        return self.message

__init__(self, message) special

Initialization.

Parameters:

Name	Type	Description	Default
message	str	str, the error message	required

Source code in zenml/exceptions.py

def __init__(self, message: str) -> None:
    """Initialization.

    Args:
        message:str, the error message
    """
    self.message = message

__str__(self) special

String function.

Returns:

Type	Description
str	the error message

Source code in zenml/exceptions.py

def __str__(self) -> str:
    """String function.

    Returns:
        the error message
    """
    return self.message

ZenMLBaseException (Exception)

Base exception for all ZenML Exceptions.

Source code in zenml/exceptions.py

class ZenMLBaseException(Exception):
    """Base exception for all ZenML Exceptions."""

    def __init__(
        self,
        message: Optional[str] = None,
        url: Optional[str] = None,
    ):
        """The BaseException used to format messages displayed to the user.

        Args:
            message: Message with details of exception. This message
                     will be appended with another message directing user to
                     `url` for more information. If `None`, then default
                     Exception behavior is used.
            url: URL to point to in exception message. If `None`, then no url
                 is appended.
        """
        if message and url:
            message += f" For more information, visit {url}."
        super().__init__(message)

__init__(self, message=None, url=None) special

The BaseException used to format messages displayed to the user.

Parameters:

Name	Type	Description	Default
message	Optional[str]	Message with details of exception. This message will be appended with another message directing user to url for more information. If None, then default Exception behavior is used.	None
url	Optional[str]	URL to point to in exception message. If None, then no url is appended.	None

Source code in zenml/exceptions.py

def __init__(
    self,
    message: Optional[str] = None,
    url: Optional[str] = None,
):
    """The BaseException used to format messages displayed to the user.

    Args:
        message: Message with details of exception. This message
                 will be appended with another message directing user to
                 `url` for more information. If `None`, then default
                 Exception behavior is used.
        url: URL to point to in exception message. If `None`, then no url
             is appended.
    """
    if message and url:
        message += f" For more information, visit {url}."
    super().__init__(message)