Send notifications from your code

To report a run’s outcome, wrap the work in the driver task in try/except, post the message, and re-raise. The notification is ordinary code, so it reads its destination from a secret when the task runs. The same deployment can then post to a different webhook on each cluster.

Built-in notifications need no code, but their delivery details are fixed when you submit the run. Use this pattern when the destination must be chosen at runtime, or when the message needs results the run computed.

notify_from_code.py
                
                    
                
            
                
            

                
                    
                
            
import os

import httpx

import flyte

env = flyte.TaskEnvironment(
    name="notify_from_code",
    secrets=[flyte.Secret(key="SLACK_WEBHOOK_URL")],
)

@flyte.trace
async def post_slack(text: str) -> None:
    # The secret is read when the task runs, not when it is deployed.
    async with httpx.AsyncClient() as client:
        response = await client.post(os.environ["SLACK_WEBHOOK_URL"], json={"text": text})
        response.raise_for_status()

@env.task(retries=3)
async def train(x: int) -> int:
    if x < 0:
        raise ValueError("x must be non-negative")
    return x * 2

@env.task
async def pipeline(x: int) -> int:
    run_name = flyte.ctx().action.run_name
    try:
        result = await train(x)
    except Exception as e:
        await post_slack(f"Run {run_name} failed: {e}")
        raise
    await post_slack(f"Run {run_name} succeeded with result {result}.")
    return result

Create the secret before you run the task:

flyte create secret SLACK_WEBHOOK_URL --value <your-webhook-url>

Each cluster pool has its own secret store. To post to a different webhook from each pool, create SLACK_WEBHOOK_URL in each one with --cluster-pool. See Secrets.

Notify once per run

Three details keep the message from being sent more than once:

Put retries on the child tasks, not the driver. train retries up to three times, and the driver sees the exception only after the last attempt fails, so it posts once. If the driver had retries, each driver attempt would post again.
Make the post a traced function. A completed trace is a checkpoint. If the driver resumes after an interruption, it doesn’t send the message again.
Re-raise after posting. The run still ends as failed, and the error stays in the run.

The exception from a failed child task reaches the driver as flyte.errors.RuntimeUserError, with the original error message. The example catches Exception so that system errors notify too.

Limitations

The driver sends the notification, so nothing is sent if the driver itself doesn’t reach the except block. This happens when:

The driver task runs out of memory, times out, or crashes.
The run is aborted.