Skip to content

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_utils

Utility functions for alerters.

get_active_alerter(context)

Get the alerter component of the active stack.

Parameters:

Name Type Description Default
context StepContext

StepContext of the ZenML repository.

required

Returns:

Type Description
BaseAlerter

Alerter component of the active stack.

Exceptions:

Type Description
DoesNotExistException

if active stack has no slack alerter.

Source code in zenml/alerter/alerter_utils.py
def get_active_alerter(context: StepContext) -> BaseAlerter:
    """Get the alerter component of the active stack.

    Args:
        context: StepContext of the ZenML repository.

    Returns:
        Alerter component of the active stack.

    Raises:
        DoesNotExistException: if active stack has no slack alerter.
    """
    # TODO: duplicate code with examples/feast_feature_store/run.py
    if not context.stack:
        raise DoesNotExistException(
            "No active stack is available. "
            "Please make sure that you have registered and set a stack."
        )
    if not context.stack.alerter:
        raise DoesNotExistException(
            "The active stack needs to have an alerter component registered "
            "to be able to use an `alerter_step`. "
            "You can create a new stack with e.g. a Slack alerter component or update "
            "your existing stack to add this component, e.g.:\n\n"
            "  'zenml alerter register slack_alerter --flavor=slack' ...\n"
            "  'zenml stack register stack-name -al slack_alerter ...'\n"
        )
    return context.stack.alerter

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