Manual triggers and firing on demand
Not every repeated run is on a schedule. An analyst re-runs a report with the same few input combinations whenever someone asks, and retypes them each time. A trigger with no automation saves each combination under a name, and anyone can fire it from the UI or from Python.
Triggers without automation
Triggers without automation, and firing a trigger on demand from Python, require flyte 2.7.1 or later.
Leave automation unset and the trigger has nothing that fires it.
It becomes a saved launch configuration for the task: a name, a set of inputs, and optionally env vars, a queue, notifications and the other flyte.Trigger settings.
Nothing runs until someone fires it, from the UI or from Python (see
Firing a trigger on demand).
This is a convenient way to publish a handful of “blessed” ways to run a task without re-typing inputs each time. Here, one task gets a quick sanity-check configuration and a full monthly one, each with its own inputs and notifications:
from datetime import datetime
import flyte
import flyte.notify
from flyte.models import ActionPhase
env = flyte.TaskEnvironment(name="manual_trigger_example")
# No `automation=`: nothing schedules these. Each is a named set of inputs
# plus what to do when the run ends.
#
# Trigger inputs override the task's own defaults (`region="all"`, `days=7`
# below) for every run fired through the trigger. Inputs the trigger does not
# mention keep the task default, so `quick-report` still gets `as_of=None`.
quick_report = flyte.Trigger(
name="quick-report",
inputs={"region": "us-east", "days": 1},
description="Yesterday only, for a fast sanity check",
notifications=flyte.notify.Slack(
on_phase=(ActionPhase.FAILED, ActionPhase.TIMED_OUT),
webhook_url="https://hooks.slack.com/services/YOUR/WEBHOOK/URL",
message=":x: quick-report {{.Run.Name}} ended in {{.Phase}}: {{.Error}}",
),
)
full_report = flyte.Trigger(
name="full-report",
inputs={"region": "all", "days": 30},
description="The full monthly report",
env_vars={"REPORT_VERBOSE": "1"},
notifications=(
flyte.notify.Email(
on_phase=ActionPhase.SUCCEEDED,
recipients=["[email protected]"],
subject="Monthly report {{.Run.Name}} is ready",
body="Run: {{.Run.Name}}\nProject/Domain: {{.Run.Project}}/{{.Run.Domain}}",
),
flyte.notify.Slack(
on_phase=ActionPhase.FAILED,
webhook_url="https://hooks.slack.com/services/YOUR/WEBHOOK/URL",
message=":rotating_light: full-report {{.Run.Name}} failed: {{.Error}}",
),
),
)
Trigger inputs override the task’s own defaults for every run fired through the trigger.
Inputs the trigger does not mention keep the task default, so quick-report above still runs with as_of=None.
A manual trigger can sit next to scheduled ones on the same task. Only a schedule can bind flyte.TriggerTime, since a manual trigger has no fire time:
# A scheduled trigger can sit next to the manual ones on the same task. Only a
# schedule can bind `flyte.TriggerTime`, since a manual trigger has no fire time.
nightly = flyte.Trigger(
name="nightly",
automation=flyte.Cron("0 2 * * *"),
inputs={"as_of": flyte.TriggerTime, "region": "all", "days": 1},
)
@env.task(triggers=(quick_report, full_report, nightly))
async def report_on_demand(region: str = "all", days: int = 7, as_of: datetime | None = None) -> str:
as_of = as_of or datetime.now()
msg = f"report for region={region!r} over the last {days} day(s), as of {as_of.isoformat()}"
print(msg)
return msg
Deploy the task as usual and all three triggers are registered:
flyte.deploy(env)
for trigger in flyte.remote.Trigger.listall():
print(trigger.name, trigger.task_name)flyte deploy manual.py env
flyte get triggerYou can also create a manual trigger for an already-deployed task. In the CLI, leave out --schedule:
flyte.remote.Trigger.create(
flyte.Trigger(name="ad-hoc", description="Fire by hand"),
task_name="manual_trigger_example.report_on_demand",
)flyte create trigger manual_trigger_example.report_on_demand ad-hoc --description "Fire by hand"Firing a trigger on demand
Passing a trigger to flyte.run() requires flyte 2.7.1 or later.
Every deployed trigger can be fired on demand, whether or not it has an automation. The run starts with the inputs, env vars, queue and notification rules the trigger was deployed with, and the platform records the trigger as the run’s origin, exactly as it does for a scheduled fire.
From the UI
Open the task in the UI, go to its Triggers tab, open the trigger you want, and select Run.
From Python
Fetch the trigger with flyte.remote.Trigger.get() and pass it to flyte.run(), exactly like a task.
This needs a remote client, so call flyte.init_from_config() or flyte.init() first.
import flyte
import flyte.remote
TASK_NAME = "manual_trigger_example.report_on_demand"
def fire_as_deployed() -> flyte.remote.Run:
"""Fire `full-report` with exactly what it was deployed with (region="all", days=30)."""
trigger = flyte.remote.Trigger.get(name="full-report", task_name=TASK_NAME)
return flyte.run(trigger)
Keyword arguments override individual inputs. Anything left out keeps the value the trigger was deployed with, or the task default if the trigger never bound that input:
def fire_with_overrides() -> flyte.remote.Run:
"""Override one input; `region` keeps the trigger's value, `days` becomes 3."""
trigger = flyte.remote.Trigger.get(name="full-report", task_name=TASK_NAME)
return flyte.run(trigger, days=3)
Overrides are keyword-only. Positional arguments raise an error, because a trigger already binds a subset of the inputs and positional values would be ambiguous. Passing an input name the task does not have also raises an error, listing the known inputs.
flyte.with_runcontext() layers run-level settings on top of the trigger’s own. The trigger’s env vars, queue and notifications are the floor; anything the run context sets wins over the trigger’s value for that setting:
def fire_with_runcontext() -> flyte.remote.Run:
"""Layer run-level overrides on top of the trigger's run spec.
The trigger's own env vars and notification rules are kept; `EXTRA_FLAG` is added and
the run gets a fixed name. Anything `with_runcontext` sets wins over the trigger's value.
"""
trigger = flyte.remote.Trigger.get(name="quick-report", task_name=TASK_NAME)
return flyte.with_runcontext(env_vars={"EXTRA_FLAG": "1"}, name="quick-report-from-python").run(trigger)
Triggers returned by flyte.remote.Trigger.listall() can be fired the same way. Their full definition is fetched when you run them:
def fire_every_trigger_on_task() -> list[flyte.remote.Run]:
"""Triggers from `listall()` can be fired too; their details are fetched on demand.
Scheduled triggers (`nightly` here) fire just fine off-schedule: the platform stamps the
run start time, and any `flyte.TriggerTime` input is filled from it.
"""
runs = []
for trigger in flyte.remote.Trigger.listall(task_name=TASK_NAME):
runs.append(flyte.run(trigger))
return runs
A scheduled trigger fired this way runs immediately, off its schedule. The platform stamps the run start time, and any input bound to flyte.TriggerTime is filled from it.
If you pass that input explicitly as a keyword override, your value is used instead.
The returned flyte.remote.Run is an ordinary run handle, so you can wait() on it and read its inputs and outputs:
if __name__ == "__main__":
flyte.init_from_config()
run = fire_with_overrides()
print(f"fired full-report with days=3: {run.url}")
run.wait()
print(f"phase={run.phase} inputs={run.inputs()} outputs={run.outputs()}")
run = fire_with_runcontext()
print(f"fired quick-report with run-context overrides: {run.url}")
Running a trigger is remote-only. It is not supported in local mode or with dry_run.