|
| 1 | +"""Triggers without automation: a named, pre-bound launch configuration. |
| 2 | +
|
| 3 | +A `Trigger` does not have to be scheduled. Leave `automation` unset and the |
| 4 | +trigger becomes a saved launch configuration for the task -- default inputs, |
| 5 | +queue, env vars, notifications -- that nothing fires on its own. It is fired |
| 6 | +on demand only (from the UI, or via the API), which makes it a convenient way |
| 7 | +to publish a handful of "blessed" ways to run a task without re-typing inputs. |
| 8 | +
|
| 9 | +Try it (see the env vars below for where notifications are delivered): |
| 10 | +
|
| 11 | + SLACK_WEBHOOK_URL=... NOTIFICATION_EMAIL=... flyte deploy examples/triggers/manual.py env |
| 12 | + flyte get trigger |
| 13 | +
|
| 14 | +`report_on_demand` deploys with two such triggers, `quick-report` and |
| 15 | +`full-report`, alongside a regular scheduled one. Fire either manual trigger |
| 16 | +from the task's Triggers tab in the UI, or from Python with `flyte.run` (see |
| 17 | +`programmatic.py`). The run starts with the trigger's bound inputs, env vars |
| 18 | +and notifications. |
| 19 | +""" |
| 20 | + |
| 21 | +import os |
| 22 | +from datetime import datetime |
| 23 | + |
| 24 | +import flyte |
| 25 | +import flyte.notify |
| 26 | +from flyte.models import ActionPhase |
| 27 | + |
| 28 | +env = flyte.TaskEnvironment(name="manual_trigger_example") |
| 29 | + |
| 30 | +# Where notifications go. Read from the deploying shell so no credentials land |
| 31 | +# in the example: |
| 32 | +# |
| 33 | +# SLACK_WEBHOOK_URL=https://hooks.slack.com/services/... \ |
| 34 | +# NOTIFICATION_EMAIL=you@example.com \ |
| 35 | +# flyte deploy examples/triggers/manual.py env |
| 36 | +# |
| 37 | +# webhook.site is handy for seeing the Slack payload if you have no webhook yet. |
| 38 | +SLACK_WEBHOOK = os.environ.get("SLACK_WEBHOOK_URL", "https://webhook.site/") |
| 39 | +REPORT_RECIPIENTS = (os.environ.get("NOTIFICATION_EMAIL", "<EMAIL>"),) |
| 40 | + |
| 41 | +# No `automation=`: nothing schedules these. Each is just a named set of |
| 42 | +# inputs plus what to do when the run ends. |
| 43 | +# |
| 44 | +# Trigger inputs override the task's own defaults (`region="all"`, `days=7` |
| 45 | +# below) for every run fired through the trigger. Inputs the trigger does not |
| 46 | +# mention keep the task default, so `quick-report` still gets `as_of=None`. |
| 47 | +quick_report = flyte.Trigger( |
| 48 | + name="quick-report", |
| 49 | + inputs={"region": "us-east", "days": 1}, |
| 50 | + description="Yesterday only, for a fast sanity check", |
| 51 | + # A quick check only needs to shout when something goes wrong. |
| 52 | + notifications=flyte.notify.Slack( |
| 53 | + on_phase=(ActionPhase.FAILED, ActionPhase.TIMED_OUT), |
| 54 | + webhook_url=SLACK_WEBHOOK, |
| 55 | + message=":x: quick-report {{.Run.Name}} ended in {{.Phase}}: {{.Error}}", |
| 56 | + ), |
| 57 | +) |
| 58 | + |
| 59 | +full_report = flyte.Trigger( |
| 60 | + name="full-report", |
| 61 | + inputs={"region": "all", "days": 30}, |
| 62 | + description="The full monthly report", |
| 63 | + env_vars={"REPORT_VERBOSE": "1"}, |
| 64 | + # The monthly report is worth an email on success and a Slack ping on failure. |
| 65 | + notifications=( |
| 66 | + flyte.notify.Email( |
| 67 | + on_phase=ActionPhase.SUCCEEDED, |
| 68 | + recipients=REPORT_RECIPIENTS, |
| 69 | + subject="Monthly report {{.Run.Name}} is ready", |
| 70 | + body="The full report finished.\nRun: {{.Run.Name}}\nProject/Domain: {{.Run.Project}}/{{.Run.Domain}}", |
| 71 | + ), |
| 72 | + flyte.notify.Slack( |
| 73 | + on_phase=ActionPhase.FAILED, |
| 74 | + webhook_url=SLACK_WEBHOOK, |
| 75 | + message=":rotating_light: full-report {{.Run.Name}} failed: {{.Error}}", |
| 76 | + ), |
| 77 | + ), |
| 78 | +) |
| 79 | + |
| 80 | +# A scheduled trigger can sit next to the manual ones on the same task. Only a |
| 81 | +# schedule can bind `flyte.TriggerTime`, since a manual trigger has no fire time. |
| 82 | +nightly = flyte.Trigger( |
| 83 | + name="nightly", |
| 84 | + automation=flyte.Cron("0 2 * * *"), |
| 85 | + inputs={"as_of": flyte.TriggerTime, "region": "all", "days": 1}, |
| 86 | + notifications=flyte.notify.Slack( |
| 87 | + on_phase=ActionPhase.FAILED, |
| 88 | + webhook_url=SLACK_WEBHOOK, |
| 89 | + message=":rotating_light: nightly report {{.Run.Name}} failed: {{.Error}}", |
| 90 | + ), |
| 91 | +) |
| 92 | + |
| 93 | + |
| 94 | +@env.task(triggers=(quick_report, full_report, nightly)) |
| 95 | +async def report_on_demand(region: str = "all", days: int = 7, as_of: datetime | None = None) -> str: |
| 96 | + as_of = as_of or datetime.now() |
| 97 | + msg = f"report for region={region!r} over the last {days} day(s), as of {as_of.isoformat()}" |
| 98 | + print(msg) |
| 99 | + return msg |
| 100 | + |
| 101 | + |
| 102 | +if __name__ == "__main__": |
| 103 | + flyte.init_from_config() |
| 104 | + flyte.deploy(env) |
0 commit comments