Trigger

Package: flyte

Specification for a trigger that can be associated with any Flyte task.

A trigger is a named, pre-bound launch configuration for a task (default inputs, queue, env vars, notifications, …). It may optionally carry an automation that fires it on a schedule (Cron/FixedRate) or when a new artifact version is published (OnArtifact). A trigger with no automation is fired only on demand (for example from the UI or via the API). Triggers are set only in the @env.task decorator via the triggers parameter. The same Trigger object can be associated with multiple tasks.

Predefined convenience constructors are available: Trigger.hourly(), Trigger.daily(), Trigger.weekly(), Trigger.monthly(), and Trigger.minutely().

        
    
my_trigger = flyte.Trigger(
    name="my_trigger",
    description="A trigger that runs every hour",
    inputs={"start_time": flyte.TriggerTime, "x": 1},
    automation=flyte.FixedRate(60),
)

@env.task(triggers=[my_trigger])
async def my_task(start_time: datetime, x: int) -> str:
    ...

Example (keep a production entrypoint on a released version while CI deploys pre-releases):

        
prod = flyte.Trigger(name="prod", automation=flyte.Cron("0 6 * * *"), task_version="v1.4.0")

@env.task(triggers=[prod])
async def entrypoint() -> None:
    ...

Parameters

        
    
class Trigger(
    name: str,
    automation: Union[Cron, FixedRate, OnArtifact, None] = None,
    description: str = '',
    auto_activate: bool = True,
    inputs: Dict[str, Any] | None = None,
    env_vars: Dict[str, str] | None = None,
    interruptible: bool | None = None,
    overwrite_cache: bool = False,
    queue: str | None = None,
    max_action_concurrency: int | None = None,
    labels: Mapping[str, str] | None = None,
    annotations: Mapping[str, str] | None = None,
    notifications: NamedRule | Notification | Tuple[Notification, ...] | None = None,
    custom_context: Mapping[str, str] | None = None,
    task_version: str | None = None,
)

Parameter	Type	Description
`name`	`str`	Unique name for the trigger (required).
`automation`	`Union[Cron, FixedRate, OnArtifact, None]`	What fires the trigger — `Cron(...)`, `FixedRate(...)` or `OnArtifact(...)`. `None` (default) means the trigger has no automation and is only fired on demand.
`description`	`str`	Human-readable description (max 255 characters). Default `""`.
`auto_activate`	`bool`	Whether to activate the trigger automatically on deployment. Default `True`.
`inputs`	`Dict[str, Any] \| None`	Default input values for triggered runs. Use `flyte.TriggerTime` to bind the trigger’s scheduled time to an input parameter.
`env_vars`	`Dict[str, str] \| None`	Environment variables for triggered runs (overrides the task’s configured values).
`interruptible`	`bool \| None`	Whether triggered runs use spot/preemptible instances. `None` (default) preserves the task’s configured behavior. Overrides the task’s configured value.
`overwrite_cache`	`bool`	Force cache refresh on triggered runs. Default `False`.
`queue`	`str \| None`	Queue name for triggered runs (overrides the task’s configured value).
`max_action_concurrency`	`int \| None`	Maximum number of actions that can run concurrently within a triggered run. `None` (default) defers to the platform default (the `run.max_action_concurrency` setting). Must be 0 (platform default) or at least 2 — a value of 1 would deadlock the run, since the parent action holds a concurrency slot while waiting for its child actions.
`labels`	`Mapping[str, str] \| None`	Kubernetes labels to attach to triggered runs.
`annotations`	`Mapping[str, str] \| None`	Kubernetes annotations to attach to triggered runs.
`notifications`	`NamedRule \| Notification \| Tuple[Notification, ...] \| None`
`custom_context`	`Mapping[str, str] \| None`	Metadata propagated through the entire task hierarchy of triggered runs. Readable inside any task via `flyte.ctx().custom_context`.
`task_version`	`str \| None`	Pin the trigger to this version of the task. `None` (default) means the trigger follows the task: every `flyte deploy` points it at the version being deployed. When set, deploying other versions leaves the trigger on this version, while the rest of its spec (automation, inputs, activation, run settings) is still updated from code. The version must already be deployed (or be the one being deployed). Promote or roll back by changing the pin and redeploying.

Methods

Method	Description
`daily()`	Creates a Cron trigger that runs daily at midnight.
`hourly()`	Creates a Cron trigger that runs every hour.
`minutely()`	Creates a Cron trigger that runs every minute.
`monthly()`	Creates a Cron trigger that runs monthly on the 1st at midnight.
`weekly()`	Creates a Cron trigger that runs weekly on Sundays at midnight.

daily()

        
    
def daily(
    trigger_time_input_key: str | None = None,
    name: str = 'daily',
    description: str = 'A trigger that runs daily at midnight',
    auto_activate: bool = True,
    inputs: Dict[str, Any] | None = None,
    env_vars: Dict[str, str] | None = None,
    interruptible: bool | None = None,
    overwrite_cache: bool = False,
    queue: str | None = None,
    max_action_concurrency: int | None = None,
    labels: Mapping[str, str] | None = None,
    annotations: Mapping[str, str] | None = None,
    custom_context: Mapping[str, str] | None = None,
) -> Trigger

Creates a Cron trigger that runs daily at midnight.

Parameter	Type	Description
`trigger_time_input_key`	`str \| None`	The input key for the trigger time. If None, no trigger time input is added.
`name`	`str`	The name of the trigger, default is “daily”.
`description`	`str`	A description of the trigger.
`auto_activate`	`bool`	Whether the trigger should be automatically activated.
`inputs`	`Dict[str, Any] \| None`	Optional inputs for the trigger.
`env_vars`	`Dict[str, str] \| None`	Optional environment variables.
`interruptible`	`bool \| None`	Whether the triggered run is interruptible.
`overwrite_cache`	`bool`	Whether to overwrite the cache.
`queue`	`str \| None`	Optional queue to run the trigger in.
`max_action_concurrency`	`int \| None`	Optional maximum number of actions that can run concurrently within a triggered run.
`labels`	`Mapping[str, str] \| None`	Optional labels to attach to the trigger.
`annotations`	`Mapping[str, str] \| None`	Optional annotations to attach to the trigger.
`custom_context`	`Mapping[str, str] \| None`	Optional context metadata propagated to triggered runs.

Returns: Trigger: A trigger that runs daily at midnight.

hourly()

        
    
def hourly(
    trigger_time_input_key: str | None = None,
    name: str = 'hourly',
    description: str = 'A trigger that runs every hour',
    auto_activate: bool = True,
    inputs: Dict[str, Any] | None = None,
    env_vars: Dict[str, str] | None = None,
    interruptible: bool | None = None,
    overwrite_cache: bool = False,
    queue: str | None = None,
    max_action_concurrency: int | None = None,
    labels: Mapping[str, str] | None = None,
    annotations: Mapping[str, str] | None = None,
    custom_context: Mapping[str, str] | None = None,
) -> Trigger

Creates a Cron trigger that runs every hour.

Parameter	Type	Description
`trigger_time_input_key`	`str \| None`	The input parameter for the trigger time. If None, no trigger time input is added.
`name`	`str`	The name of the trigger, default is “hourly”.
`description`	`str`	A description of the trigger.
`auto_activate`	`bool`	Whether the trigger should be automatically activated.
`inputs`	`Dict[str, Any] \| None`	Optional inputs for the trigger.
`env_vars`	`Dict[str, str] \| None`	Optional environment variables.
`interruptible`	`bool \| None`	Whether the trigger is interruptible.
`overwrite_cache`	`bool`	Whether to overwrite the cache.
`queue`	`str \| None`	Optional queue to run the trigger in.
`max_action_concurrency`	`int \| None`	Optional maximum number of actions that can run concurrently within a triggered run.
`labels`	`Mapping[str, str] \| None`	Optional labels to attach to the trigger.
`annotations`	`Mapping[str, str] \| None`	Optional annotations to attach to the trigger.
`custom_context`	`Mapping[str, str] \| None`	Optional context metadata propagated to triggered runs.

Returns: Trigger: A trigger that runs every hour, on the hour.

minutely()

        
    
def minutely(
    trigger_time_input_key: str | None = None,
    name: str = 'minutely',
    description: str = 'A trigger that runs every minute',
    auto_activate: bool = True,
    inputs: Dict[str, Any] | None = None,
    env_vars: Dict[str, str] | None = None,
    interruptible: bool | None = None,
    overwrite_cache: bool = False,
    queue: str | None = None,
    max_action_concurrency: int | None = None,
    labels: Mapping[str, str] | None = None,
    annotations: Mapping[str, str] | None = None,
    custom_context: Mapping[str, str] | None = None,
) -> Trigger

Creates a Cron trigger that runs every minute.

Parameter	Type	Description
`trigger_time_input_key`	`str \| None`	The input parameter for the trigger time. If None, no trigger time input is added.
`name`	`str`	The name of the trigger, default is “every_minute”.
`description`	`str`	A description of the trigger.
`auto_activate`	`bool`	Whether the trigger should be automatically activated.
`inputs`	`Dict[str, Any] \| None`	Optional inputs for the trigger.
`env_vars`	`Dict[str, str] \| None`	Optional environment variables.
`interruptible`	`bool \| None`	Whether the trigger is interruptible.
`overwrite_cache`	`bool`	Whether to overwrite the cache.
`queue`	`str \| None`	Optional queue to run the trigger in.
`max_action_concurrency`	`int \| None`	Optional maximum number of actions that can run concurrently within a triggered run.
`labels`	`Mapping[str, str] \| None`	Optional labels to attach to the trigger.
`annotations`	`Mapping[str, str] \| None`	Optional annotations to attach to the trigger.
`custom_context`	`Mapping[str, str] \| None`	Optional context metadata propagated to triggered runs.

Returns: Trigger: A trigger that runs every minute.

monthly()

        
    
def monthly(
    trigger_time_input_key: str | None = None,
    name: str = 'monthly',
    description: str = 'A trigger that runs monthly on the 1st at midnight',
    auto_activate: bool = True,
    inputs: Dict[str, Any] | None = None,
    env_vars: Dict[str, str] | None = None,
    interruptible: bool | None = None,
    overwrite_cache: bool = False,
    queue: str | None = None,
    max_action_concurrency: int | None = None,
    labels: Mapping[str, str] | None = None,
    annotations: Mapping[str, str] | None = None,
    custom_context: Mapping[str, str] | None = None,
) -> Trigger

Creates a Cron trigger that runs monthly on the 1st at midnight.

Parameter	Type	Description
`trigger_time_input_key`	`str \| None`	The input parameter for the trigger time. If None, no trigger time input is added.
`name`	`str`	The name of the trigger, default is “monthly”.
`description`	`str`	A description of the trigger.
`auto_activate`	`bool`	Whether the trigger should be automatically activated.
`inputs`	`Dict[str, Any] \| None`	Optional inputs for the trigger.
`env_vars`	`Dict[str, str] \| None`	Optional environment variables.
`interruptible`	`bool \| None`	Whether the trigger is interruptible.
`overwrite_cache`	`bool`	Whether to overwrite the cache.
`queue`	`str \| None`	Optional queue to run the trigger in.
`max_action_concurrency`	`int \| None`	Optional maximum number of actions that can run concurrently within a triggered run.
`labels`	`Mapping[str, str] \| None`	Optional labels to attach to the trigger.
`annotations`	`Mapping[str, str] \| None`	Optional annotations to attach to the trigger.
`custom_context`	`Mapping[str, str] \| None`	Optional context metadata propagated to triggered runs.

Returns: Trigger: A trigger that runs monthly on the 1st at midnight.

weekly()

        
    
def weekly(
    trigger_time_input_key: str | None = None,
    name: str = 'weekly',
    description: str = 'A trigger that runs weekly on Sundays at midnight',
    auto_activate: bool = True,
    inputs: Dict[str, Any] | None = None,
    env_vars: Dict[str, str] | None = None,
    interruptible: bool | None = None,
    overwrite_cache: bool = False,
    queue: str | None = None,
    max_action_concurrency: int | None = None,
    labels: Mapping[str, str] | None = None,
    annotations: Mapping[str, str] | None = None,
    custom_context: Mapping[str, str] | None = None,
) -> Trigger

Creates a Cron trigger that runs weekly on Sundays at midnight.

Parameter	Type	Description
`trigger_time_input_key`	`str \| None`	The input parameter for the trigger time. If None, no trigger time input is added.
`name`	`str`	The name of the trigger, default is “weekly”.
`description`	`str`	A description of the trigger.
`auto_activate`	`bool`	Whether the trigger should be automatically activated.
`inputs`	`Dict[str, Any] \| None`	Optional inputs for the trigger.
`env_vars`	`Dict[str, str] \| None`	Optional environment variables.
`interruptible`	`bool \| None`	Whether the trigger is interruptible.
`overwrite_cache`	`bool`	Whether to overwrite the cache.
`queue`	`str \| None`	Optional queue to run the trigger in.
`max_action_concurrency`	`int \| None`	Optional maximum number of actions that can run concurrently within a triggered run.
`labels`	`Mapping[str, str] \| None`	Optional labels to attach to the trigger.
`annotations`	`Mapping[str, str] \| None`	Optional annotations to attach to the trigger.
`custom_context`	`Mapping[str, str] \| None`	Optional context metadata propagated to triggered runs.

Returns: Trigger: A trigger that runs weekly on Sundays at midnight.