p failure using the Alerter from the active stack.We use @step for success notification to only notify the user about a fully successful pipeline run and not about every successful step.

Inside the helper function build_message(), you will find an example on how developers can work with StepContext to form a proper notification:

from zenml import get_step_context

def build_message(status: str) -> str:

"""Builds a message to post.

Args:

status: Status to be set in text.

Returns:

str: Prepared message.

"""

step_context = get_step_context()

run_url = get_run_url(step_context.pipeline_run)

return (

f"Pipeline `{step_context.pipeline.name}` [{str(step_context.pipeline.id)}] {status}!\n"

f"Run `{step_context.pipeline_run.name}` [{str(step_context.pipeline_run.id)}]\n"

f"URL: {run_url}"

@step(enable_cache=False)

def notify_on_success() -> None:

"""Notifies user on pipeline success."""

step_context = get_step_context()

if alerter and step_context.pipeline_run.config.extra["notify_on_success"]:

alerter.post(message=build_message(status="succeeded"))

Linking to the Alerter Stack component

A common use case is to use the Alerter component inside the failure or success hooks to notify relevant people. It is quite easy to do this:

from zenml import get_step_context

from zenml.client import Client

def on_failure():

step_name = get_step_context().step_run.name

Client().active_stack.alerter.post(f"{step_name} just failed!")

ZenML provides standard failure and success hooks that use the alerter you have configured in your stack. Here's an example of how to use them in your pipelines:

from zenml.hooks import alerter_success_hook, alerter_failure_hook

@step(on_failure=alerter_failure_hook, on_success=alerter_success_hook)

def my_step(...):

...

To set up the local environment used below, follow the recommendations from the Project templates.