Notifications for triggers and runs (local and remote)

examples/notification/slack_webhook.py

@@ -0,0 +1,104 @@
+"""Slack + Email notification example.
+
+Sends a rich Slack notification and an email when a task succeeds or fails.
+
+Usage:
+    export SLACK_WEBHOOK_URL="https://hooks.slack.com/services/YOUR/WEBHOOK/URL"
+    export NOTIFICATION_EMAIL="oncall@example.com"
+    python examples/notification/slack_webhook.py
+
+To receive emails locally, start a debug SMTP server before running:
+
+    # Python < 3.12
+    sudo python -m smtpd -n -c DebuggingServer localhost:25
+
+    # Python >= 3.12 (smtpd was removed; use aiosmtpd)
+    pip install aiosmtpd
+    sudo python -m aiosmtpd -n -l localhost:25
+
+Both print received emails to stdout. Port 25 requires root (sudo);
+alternatively use port 1025, but update the SMTP port in _sender.py.
+"""
+
+import os
+
+import flyte
+from flyte import notify
+from flyte.models import ActionPhase
+
+env = flyte.TaskEnvironment(name="notify_example")
+
+SLACK_WEBHOOK_URL = os.environ["SLACK_WEBHOOK_URL"]
+NOTIFICATION_EMAIL = os.environ["NOTIFICATION_EMAIL"]
+
+
+@env.task
+def compute(x: int, y: int) -> int:
+    return x + y
+
+
+slack_success = notify.Slack(
+    on_phase=ActionPhase.SUCCEEDED,
+    webhook_url=SLACK_WEBHOOK_URL,
+    blocks=[
+        {
+            "type": "header",
+            "text": {"type": "plain_text", "text": "Task Succeeded"},
+        },
+        {
+            "type": "section",
+            "fields": [
+                {"type": "mrkdwn", "text": "*Task:*\n{task.name}"},
+                {"type": "mrkdwn", "text": "*Run:*\n{run.name}"},
+                {"type": "mrkdwn", "text": "*Duration:*\n{run.duration}"},
+                {"type": "mrkdwn", "text": "*Phase:*\n{run.phase}"},
+            ],
+        },
+        {"type": "divider"},
+        {
+            "type": "context",
+            "elements": [
+                {"type": "mrkdwn", "text": "{project}/{domain}"},
+            ],
+        },
+    ],
+)
+
+slack_failure = notify.Slack(
+    on_phase=ActionPhase.FAILED,
+    webhook_url=SLACK_WEBHOOK_URL,
+    blocks=[
+        {
+            "type": "header",
+            "text": {"type": "plain_text", "text": "Task Failed"},
+        },
+        {
+            "type": "section",
+            "fields": [
+                {"type": "mrkdwn", "text": "*Task:*\n{task.name}"},
+                {"type": "mrkdwn", "text": "*Error:*\n{run.error}"},
+            ],
+        },
+    ],
+)
+
+email_success = notify.Email(
+    on_phase=ActionPhase.SUCCEEDED,
+    recipients=[NOTIFICATION_EMAIL],
+    subject="Task {task.name} succeeded",
+    body=("Task: {task.name}\nRun: {run.name}\nDuration: {run.duration}\nPhase: {run.phase}\n"),
+)
+
+email_failure = notify.Email(
+    on_phase=ActionPhase.FAILED,
+    recipients=[NOTIFICATION_EMAIL],
+    subject="ALERT: Task {task.name} failed",
+    body=("Task: {task.name}\nRun: {run.name}\nError: {run.error}\n"),
+)
+
+if __name__ == "__main__":
+    result = flyte.with_runcontext(
+        mode="local",
+        notifications=(slack_success, slack_failure, email_success, email_failure),
+    ).run(compute, x=3, y=7)
+    print(f"Result: {result}")

examples/triggers/notifications.py

@@ -0,0 +1,94 @@
+from datetime import datetime
+
+import flyte.notify
+from flyte.models import ActionPhase
+
+env = flyte.TaskEnvironment(
+    name="example_task",
+)
+
+trig1 = flyte.Trigger(
+    name="hourly",
+    auto_activate=True,
+    automation=flyte.Cron("* * * * *"),
+    notifications=flyte.notify.Slack(
+        on_phase=ActionPhase.FAILED,
+        webhook_url="https://webhook.site/",
+        message="Hello world! from {task.name}",
+        blocks=(
+            {"type": "header", "text": {"type": "plain_text", "text": "🚨 Run Failed"}},
+            {
+                "type": "section",
+                "fields": [
+                    {"type": "mrkdwn", "text": "*Workflow:*\n{run.name}"},
+                    {"type": "mrkdwn", "text": "*Error:*\n{run.error}"},
+                ],
+            },
+            {
+                "type": "actions",
+                "elements": [
+                    {
+                        "type": "button",
+                        "text": {"type": "plain_text", "text": "View Logs"},
+                        "url": "https://logs.example.com/{run.name}",
+                    }
+                ],
+            },
+        ),
+    ),
+)
+
+trig2 = flyte.Trigger(
+    name="hourly",
+    auto_activate=True,
+    automation=flyte.Cron("* * * * *"),
+    notifications=(
+        flyte.notify.Slack(
+            on_phase=(ActionPhase.FAILED, ActionPhase.TIMED_OUT),
+            webhook_url="https://webhook.site/",
+            message="Hello world! from {task.name}",
+        ),
+        flyte.notify.Webhook(
+            on_phase=ActionPhase.FAILED,
+            headers={"Content-Type": "application/json"},
+            url="https://webhook.site/",
+            body={
+                "xyz": "{run.name}",
+            },
+        ),
+        flyte.notify.Email(
+            on_phase=ActionPhase.SUCCEEDED,
+            subject="Hello world! from {task.name}",
+            body="Hello world! from {task.name}",
+            recipients=("<EMAIL>",),
+        ),
+    ),
+)
+
+
+# Use a pre-defined named rule (configured by your Flyte admin)
+trig3 = flyte.Trigger(
+    name="hourly-with-named-rule",
+    auto_activate=True,
+    automation=flyte.Cron("0 * * * *"),
+    notifications=flyte.notify.NamedRule("oncall-alerts"),
+)
+
+# Mix named delivery configs with inline notifications
+trig4 = flyte.Trigger(
+    name="hourly-with-named-delivery",
+    auto_activate=True,
+    automation=flyte.Cron("0 * * * *"),
+    notifications=(
+        flyte.notify.NamedDelivery(on_phase=ActionPhase.FAILED, name="slack-oncall"),
+        flyte.notify.Email(
+            on_phase=ActionPhase.SUCCEEDED,
+            recipients=("team@example.com",),
+        ),
+    ),
+)
+
+
+@env.task(triggers=(trig1, trig2))  # Every hour
+def example_task(trigger_time: datetime, x: int = 1) -> str:
+    return f"Task executed at {trigger_time.isoformat()} with x={x}"