Alerter

zenml.alerter special

Alerters allow you to send alerts from within your pipeline.

This is useful to immediately get notified when failures happen, and also for general monitoring / reporting.

alerter_step

The alerter step allows you to post messages to registered alerter components.

alerter_ask_step (BaseStep)

Posts a message to the alerter component and waits for approval.

This can be useful, e.g. to easily get a human in the loop before deploying models.

Parameters:

Name	Type	Description	Default
config		Runtime configuration for the slack alerter.	required
context		StepContext of the ZenML repository.	required
message		Initial message to be posted.	required

Returns:

Type	Description
	True if a user approved the operation, else False.

CONFIG_CLASS (BaseStepConfig) pydantic-model

Step config definition for all alerters.

Source code in zenml/alerter/alerter_step.py

class BaseAlerterStepConfig(BaseStepConfig):
    """Step config definition for all alerters."""

entrypoint(config, context, message) staticmethod

Posts a message to the alerter component and waits for approval.

This can be useful, e.g. to easily get a human in the loop before deploying models.

Parameters:

Name	Type	Description	Default
config	BaseAlerterStepConfig	Runtime configuration for the slack alerter.	required
context	StepContext	StepContext of the ZenML repository.	required
message	str	Initial message to be posted.	required

Returns:

Type	Description
bool	True if a user approved the operation, else False.

Source code in zenml/alerter/alerter_step.py

@step
def alerter_ask_step(
    config: BaseAlerterStepConfig, context: StepContext, message: str
) -> bool:
    """Posts a message to the alerter component and waits for approval.

    This can be useful, e.g. to easily get a human in the loop before
    deploying models.

    Args:
        config: Runtime configuration for the slack alerter.
        context: StepContext of the ZenML repository.
        message: Initial message to be posted.

    Returns:
        True if a user approved the operation, else False.
    """
    alerter = _get_active_alerter(context)
    return alerter.ask(message, config)

alerter_post_step (BaseStep)

Post a message to the alerter component of the active stack.

Parameters:

Name	Type	Description	Default
config		Runtime configuration for the slack alerter.	required
context		StepContext of the ZenML repository.	required
message		Message to be posted.	required

Returns:

Type	Description
	True if operation succeeded, else False.

CONFIG_CLASS (BaseStepConfig) pydantic-model

Step config definition for all alerters.

Source code in zenml/alerter/alerter_step.py

class BaseAlerterStepConfig(BaseStepConfig):
    """Step config definition for all alerters."""

entrypoint(config, context, message) staticmethod

Post a message to the alerter component of the active stack.

Parameters:

Name	Type	Description	Default
config	BaseAlerterStepConfig	Runtime configuration for the slack alerter.	required
context	StepContext	StepContext of the ZenML repository.	required
message	str	Message to be posted.	required

Returns:

Type	Description
bool	True if operation succeeded, else False.

Source code in zenml/alerter/alerter_step.py

@step
def alerter_post_step(
    config: BaseAlerterStepConfig, context: StepContext, message: str
) -> bool:
    """Post a message to the alerter component of the active stack.

    Args:
        config: Runtime configuration for the slack alerter.
        context: StepContext of the ZenML repository.
        message: Message to be posted.

    Returns:
        True if operation succeeded, else False.
    """
    alerter = _get_active_alerter(context)
    return alerter.post(message, config)

base_alerter

Base class for all ZenML alerters.

BaseAlerter (StackComponent, ABC) pydantic-model

Base class for all ZenML alerters.

Source code in zenml/alerter/base_alerter.py

class BaseAlerter(StackComponent, ABC):
    """Base class for all ZenML alerters."""

    # Class configuration
    TYPE: ClassVar[StackComponentType] = StackComponentType.ALERTER
    FLAVOR: ClassVar[str]

    def post(
        self, message: str, config: Optional[BaseAlerterStepConfig]
    ) -> bool:
        """Post a message to a chat service.

        Args:
            message (str): Message to be posted.
            config (Optional[BaseAlerterStepConfig]): Optional runtime
                configuration of this function.

        Returns:
            bool: True if operation succeeded, else False.
        """
        return True

    def ask(
        self, question: str, config: Optional[BaseAlerterStepConfig]
    ) -> bool:
        """Post a message to a chat service and wait for approval.

        This can be useful to easily get a human in the loop, e.g., when
        deploying models.

        Args:
            question (str): Question to ask (message to be posted).
            config (Optional[BaseAlerterStepConfig]): Optional runtime
                configuration of this function.

        Returns:
            bool: True if operation succeeded and was approved, else False.
        """
        return True

ask(self, question, config)

Post a message to a chat service and wait for approval.

This can be useful to easily get a human in the loop, e.g., when deploying models.

Parameters:

Name	Type	Description	Default
question	str	Question to ask (message to be posted).	required
config	Optional[BaseAlerterStepConfig]	Optional runtime configuration of this function.	required

Returns:

Type	Description
bool	True if operation succeeded and was approved, else False.

Source code in zenml/alerter/base_alerter.py

def ask(
    self, question: str, config: Optional[BaseAlerterStepConfig]
) -> bool:
    """Post a message to a chat service and wait for approval.

    This can be useful to easily get a human in the loop, e.g., when
    deploying models.

    Args:
        question (str): Question to ask (message to be posted).
        config (Optional[BaseAlerterStepConfig]): Optional runtime
            configuration of this function.

    Returns:
        bool: True if operation succeeded and was approved, else False.
    """
    return True

post(self, message, config)

Post a message to a chat service.

Parameters:

Name	Type	Description	Default
message	str	Message to be posted.	required
config	Optional[BaseAlerterStepConfig]	Optional runtime configuration of this function.	required

Returns:

Type	Description
bool	True if operation succeeded, else False.

Source code in zenml/alerter/base_alerter.py

def post(
    self, message: str, config: Optional[BaseAlerterStepConfig]
) -> bool:
    """Post a message to a chat service.

    Args:
        message (str): Message to be posted.
        config (Optional[BaseAlerterStepConfig]): Optional runtime
            configuration of this function.

    Returns:
        bool: True if operation succeeded, else False.
    """
    return True