Define scheduled task triggers

By the end of this tutorial, you will have a flyte.TaskEnvironment containing a task with a named schedule, scheduled-run defaults, and a kickoff-time input. You will also attach two schedules to one task, reuse a trigger specification, and deploy the environment so Flyte converts the task-level definitions into remote task triggers.

Prerequisites

You need a working flyte-sdk installation and a Flyte configuration that can be loaded by flyte.init_from_config(). Deployment also requires a configured Flyte client. The public API is re-exported from flyte, so use flyte.TaskEnvironment, flyte.Trigger, flyte.Cron, flyte.FixedRate, and flyte.TriggerTime rather than importing the private implementation module.

1. Attach a trigger to a task

Create an environment and pass a trigger directly to the @env.task decorator:

from datetime import datetime

import flyte

env = flyte.TaskEnvironment(
    name="example_task",
)


@env.task(triggers=flyte.Trigger.hourly())
def example_task(trigger_time: datetime, x: int = 1) -> str:
    return f"Task executed at {trigger_time.isoformat()} with x={x}"


assert len(example_task.triggers) == 1
assert example_task.triggers[0].name == "hourly"


if __name__ == "__main__":
    flyte.init_from_config()
    flyte.deploy(env)

Trigger.hourly() creates a Cron trigger with the expression 0 * * * *. Although the decorator accepts the single Trigger object, TaskEnvironment.task() normalizes it to a one-element tuple. The decorated object is therefore a task template whose triggers field is (trigger,).

TaskEnvironment.task() creates the task name by joining the environment name and Python function name: this task is named example_task.example_task. The optional short_name decorator argument is a UI-oriented value and does not change that fully qualified task name. The standard task is an AsyncFunctionTaskTemplate; when the environment has registered plugin configuration, the decorator instead resolves a plugin task-template class through TaskPluginRegistry.

The if __name__ == "__main__" block shows the deployment boundary. Defining the Trigger is local; flyte.deploy(env) converts it into a deployable trigger and sends it with the task deployment request.

2. Choose a schedule

The convenience constructors create cron schedules with these expressions:

import flyte

hourly = flyte.Trigger.hourly()
daily = flyte.Trigger.daily()
minutely = flyte.Trigger.minutely()
weekly = flyte.Trigger.weekly()
monthly = flyte.Trigger.monthly()

assert hourly.automation.expression == "0 * * * *"
assert daily.automation.expression == "0 0 * * *"
assert minutely.automation.expression == "* * * * *"
assert weekly.automation.expression == "0 0 * * 0"
assert monthly.automation.expression == "0 0 1 * *"

The constructors default the trigger names to hourly, daily, minutely, weekly, and monthly. They accept name= when those defaults are not appropriate. They also accept trigger_time_input_key=, which is covered in the input section below.

For a calendar schedule in a particular timezone, construct Cron explicitly. The timezone belongs to the Cron, not to the environment or task. For example, the trigger examples use Cron("1 12 * * *", timezone="America/New_York") and Cron("0 9 * * *", timezone="America/Los_Angeles") as two schedules on one task. Cron.timezone defaults to UTC, and the deployment serializer sends the expression and timezone separately. Cron schedules can skip or duplicate runs across daylight-saving-time transitions, so review timezone-specific schedules accordingly.

Use FixedRate when the schedule is an interval rather than a cron expression. It stores interval_minutes and an optional start_time. Without start_time, the first fixed-rate run occurs one interval after deployment or activation. If start_time is provided, the serializer converts that datetime to the schedule's protobuf timestamp and uses it as the interval reference for subsequent activations.

3. Name and activate triggers deliberately

A trigger requires a non-empty name and an automation value (Cron or FixedRate). The convenience constructors also let you assign a stable name:

import flyte

named_trigger = flyte.Trigger.hourly(
    name="nightly_orders",
    description="Process orders at midnight",
    auto_activate=False,
)

assert named_trigger.name == "nightly_orders"
assert named_trigger.auto_activate is False

The Trigger documentation describes name as unique, but local construction only checks that it is non-empty. Choose stable names appropriate to the task; conflicts are handled outside this local dataclass. auto_activate defaults to True. During deployment it is copied directly to the remote trigger's active field, so False creates an inactive trigger rather than changing local function execution.

After deployment, the CLI can change the remote state:

flyte update trigger nightly_orders timezone_tasks.my_task --activate
flyte update trigger nightly_orders timezone_tasks.my_task --deactivate

The command takes the trigger name, task name, and exactly one of --activate or --deactivate. This is remote trigger management; it does not add a trigger to a task template.

4. Supply scheduled inputs and the kickoff time

Use inputs for values passed to a triggered run. A convenience constructor can add the scheduled kickoff input for you:

from datetime import datetime

import flyte

env = flyte.TaskEnvironment(name="input_tasks")

trigger = flyte.Trigger.hourly(
    trigger_time_input_key="scheduled_at",
    name="custom_hourly",
    description="Custom hourly trigger",
    auto_activate=False,
    inputs={"batch_size": 100},
    env_vars={"ENV": "prod"},
    interruptible=True,
    overwrite_cache=True,
    queue="high-priority",
    labels={"team": "ml"},
    annotations={"owner": "data-team"},
)


@env.task(triggers=trigger)
async def process_batch(scheduled_at: datetime, batch_size: int, mode: str = "dev") -> str:
    return f"{scheduled_at.isoformat()} {batch_size} {mode}"


assert trigger.inputs["batch_size"] == 100
assert process_batch.triggers == (trigger,)

TriggerTime is a marker singleton, not a datetime value. The trigger_time_input_key convenience argument inserts that marker under scheduled_at; the serializer removes it from the ordinary default-input map and places scheduled_at in the schedule's kickoff_time_input_arg. At runtime, the scheduled kickoff is supplied to that task input. The key passed to trigger_time_input_key must therefore match an input in the decorated function's interface.

The other input values are converted to Flyte literals during deployment. Trigger values take precedence over task defaults with the same name; task defaults not supplied by the trigger are appended. For example, x and y remain task defaults while batch_size is supplied by the trigger, and a trigger value can replace a task default:

from datetime import datetime

import flyte

env = flyte.TaskEnvironment(name="default_inputs")

trigger = flyte.Trigger.hourly(
    trigger_time_input_key="trigger_time",
    name="custom",
    inputs={
        "batch_size": 50,
        "x": 20,
    },
)


@env.task(triggers=trigger)
async def task_with_defaults(
    trigger_time: datetime,
    batch_size: int,
    x: int = 10,
    y: str = "default",
) -> str:
    return f"{trigger_time}, {batch_size}, {x}, {y}"

Every ordinary trigger input key must be a task input. Deployment raises ValueError for an unknown key and also raises ValueError when a TriggerTime key is not present in the task interface. A required task input without a trigger value or a task default remains unsatisfied for the scheduled run.

5. Attach multiple schedules and reuse a trigger

Pass a tuple to preserve multiple trigger definitions and their order:

from datetime import datetime

import flyte

env = flyte.TaskEnvironment(name="multi_schedule")

hourly_trigger = flyte.Trigger.hourly(trigger_time_input_key="trigger_time")
daily_trigger = flyte.Trigger.daily(trigger_time_input_key="trigger_time")


@env.task(triggers=(hourly_trigger, daily_trigger))
async def task_with_triggers(trigger_time: datetime, x: int = 1) -> str:
    return f"Executed at {trigger_time.isoformat()}"


assert task_with_triggers.triggers == (hourly_trigger, daily_trigger)
assert len(task_with_triggers.triggers) == 2

Each trigger is serialized independently, with its own schedule and input defaults. The task interface must contain every input key referenced by every trigger. Different triggers may use different kickoff-input names as long as each name exists on the task.

Trigger is a frozen dataclass, and its documented API supports associating the same trigger object with multiple tasks. The environment decorator stores that same specification in each task template's tuple:

import flyte

env = flyte.TaskEnvironment(name="shared_schedule")
shared_trigger = flyte.Trigger.hourly(name="shared_hourly")


@env.task(triggers=shared_trigger)
def first_task(x: int = 1) -> int:
    return x


@env.task(triggers=shared_trigger)
def second_task(x: int = 2) -> int:
    return x


assert first_task.triggers == (shared_trigger,)
assert second_task.triggers == (shared_trigger,)
assert first_task.triggers[0] is second_task.triggers[0]

Deployment still validates the trigger against each task's interface and defaults separately. Treat the trigger and its mappings as configuration after construction; although the dataclass itself is frozen, referenced dictionaries can still be changed if you mutate them.

6. Configure the triggered run context

Trigger fields can override execution metadata for scheduled runs. The serializer maps env_vars, interruptible, queue, labels, annotations, notifications, and custom_context into the triggered run specification. overwrite_cache=True is also passed through to force a cache refresh for that scheduled run.

A task can read propagated custom context through flyte.ctx().custom_context:

from datetime import datetime

import flyte

env = flyte.TaskEnvironment(
    name="trigger_with_custom_context",
    image=flyte.Image.from_debian_base(),
)

minutely_monitor = flyte.Trigger.minutely(
    trigger_time_input_key="trigger_time",
    name="minutely_monitor",
    custom_context={"team": "infra", "alert_channel": "#ops"},
)


@env.task(triggers=minutely_monitor)
async def monitor_task(trigger_time: datetime) -> str:
    ctx = flyte.ctx().custom_context
    return f"Monitor ran for team={ctx.get('team')}"

A non-None trigger queue is serialized as the triggered run's cluster, and a non-None trigger interruptible is serialized as an explicit boolean override. Trigger environment variables are serialized separately from the task's environment settings. Labels and annotations are attached to the run specification; notification settings are resolved by the notification serializer.

How deployment consumes the task template

TaskTemplate is the common dataclass base for executable and plugin-defined tasks. Its triggers field is always tuple-valued. For ordinary Python functions, TaskEnvironment.task() creates the standard AsyncFunctionTaskTemplate, which carries the interface and trigger tuple. Plugin task classes can also subclass TaskTemplate and provide their own NativeInterface and extension methods such as custom_config.

During deployment, flyte-sdk loops over task.triggers and calls to_task_trigger() with the serialized task input interface and task default inputs. That serializer:

• converts Cron or FixedRate to the schedule protobuf;
• turns TriggerTime into kickoff_time_input_arg;
• converts ordinary trigger values with flyte.types.TypeEngine;
• appends task defaults not replaced by trigger values; and
• maps activation and run metadata into the remote trigger specification.

TaskTemplate.override() does not accept a triggers parameter. Triggers are attached at task-definition time through @env.task(triggers=...), not at the environment level or invocation override level.

Troubleshooting checklist

• Empty trigger name: Trigger raises ValueError("Trigger name cannot be empty").
• Missing schedule: automation=None raises ValueError.
• Long description: descriptions longer than 255 characters are truncated by the description parser. The tested result is exactly 255 characters and ends with ...(tr.).
• Unknown input: deployment rejects every trigger input key that is not present in the task interface.
• Missing kickoff input: a TriggerTime key must name an actual task input.
• Unexpected activation state: auto_activate=False is serialized as an inactive remote trigger; use flyte update trigger ... --activate when it is ready.
• Timezone behavior: Cron defaults to UTC, while an explicit timezone can encounter daylight-saving-time skips or duplicates.
• Fixed-rate timing: without start_time, the first run is one interval after deployment or activation.
• No invocation-time trigger override: use the decorator to add schedules; task.override() cannot add or replace triggers.
• Reusable environments: unrelated environment constraints can affect task definition: reusable environments reject pod templates at the decorator, and synchronous tasks with reusable concurrency greater than one are rejected. TaskEnvironment also rejects plugin_config combined with reusable.