Configure a trigger

Say a refresh_dashboard task serves two teams. The US team wants it every morning over the last day of data, and finance wants a 30-day run on the first of each month. You don’t need two copies of the task or a wrapper script. Attach two triggers, each with its own schedule and inputs:

        
    
import flyte

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

morning_us = flyte.Trigger(
    "morning-us",
    flyte.Cron("0 7 * * *", timezone="America/New_York"),
    inputs={"region": "us", "days": 1},
)

monthly_all = flyte.Trigger(
    "monthly-all",
    flyte.Cron("0 0 1 * *"),
    inputs={"region": "all", "days": 30},
)


@env.task(triggers=[morning_us, monthly_all])
async def refresh_dashboard(region: str, days: int = 7) -> str:
    ...

A trigger is a named, pre-bound way to run a task. It carries the inputs and run settings, and optionally an automation that fires it. This page covers how to define one. See Schedule triggers for the schedule options.

Triggers are set in the task decorator

A trigger is created by setting the triggers parameter in the task decorator to a flyte.Trigger object or a list of such objects (triggers are not settable at the TaskEnvironment definition or task.override levels).

Here is a simple example:

triggers.py
                
import flyte
from datetime import datetime, timezone

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

@env.task(triggers=flyte.Trigger.hourly())  # Every hour
def hourly_task(trigger_time: datetime, x: int = 1) -> str:
    return f"Hourly example executed at {trigger_time.isoformat()} with x={x}"

Here we use a predefined schedule trigger to run the hourly_task every hour. Other predefined triggers can be used similarly (see Predefined schedule triggers).

If you want full control over the trigger behavior, you can define a trigger using the flyte.Trigger class directly.

`flyte.Trigger`

For complete parameter documentation, see the Trigger, Cron, and FixedRate API references.

The Trigger class allows you to define custom triggers with full control over scheduling and execution behavior. It has the following signature:

triggers.py
                
flyte.Trigger(
    name,
    automation=None,
    description="",
    auto_activate=True,
    inputs=None,
    env_vars=None,
    interruptible=None,
    overwrite_cache=False,
    queue=None,
    labels=None,
    annotations=None
)

Only name is required. The automation parameter decides what fires the trigger: flyte.Cron or flyte.FixedRate for a schedule, flyte.OnArtifact for a new artifact version, or None (the default) for a trigger that is only fired on demand.

Everything else describes the run that the trigger creates.

Here’s a comprehensive example showing all parameters:

triggers.py
                
                    
                
            
                
            

                
                    
                
            
comprehensive_trigger = flyte.Trigger(
    name="monthly_financial_report",
    automation=flyte.Cron("0 6 1 * *", timezone="America/New_York"),
    description="Monthly financial report generation for executive team",
    auto_activate=True,
    inputs={
        "report_date": flyte.TriggerTime,
        "report_type": "executive_summary",
        "include_forecasts": True
    },
    env_vars={
        "REPORT_OUTPUT_FORMAT": "PDF",
        "EMAIL_NOTIFICATIONS": "true"
    },
    interruptible=False,  # Critical report, use dedicated resources
    overwrite_cache=True,  # Always fresh data
    queue="financial-reports",
    labels={
        "team": "finance",
        "criticality": "high",
        "automation": "scheduled"
    },
    annotations={
        "compliance.company.com/sox-required": "true",
        "backup.company.com/retain-days": "2555"  # 7 years
    }
)

The `inputs` parameter

The inputs parameter allows you to provide default values for your task’s parameters when the trigger fires. This is essential for parameterizing your automated executions and passing trigger-specific data to your tasks.

Basic usage

triggers.py
                
trigger_with_inputs = flyte.Trigger(
    "data_processing",
    flyte.Cron("0 6 * * *"),  # Daily at 6 AM
    inputs={
        "batch_size": 1000,
        "environment": "production",
        "debug_mode": False
    }
)

@env.task(triggers=trigger_with_inputs)
def process_data(batch_size: int, environment: str, debug_mode: bool = True) -> str:
    return f"Processing {batch_size} items in {environment} mode"

Using `flyte.TriggerTime`

The special flyte.TriggerTime value is used in the inputs to indicate the task parameter into which Flyte will inject the trigger execution timestamp. It is only available on schedule triggers (flyte.Cron or flyte.FixedRate), since a trigger without a schedule has no fire time:

triggers.py
                
timestamp_trigger = flyte.Trigger(
    "daily_report",
    flyte.Cron("0 0 * * *"),  # Daily at midnight
    inputs={
        "report_date": flyte.TriggerTime,  # Receives trigger execution time
        "report_type": "daily_summary"
    }
)

@env.task(triggers=timestamp_trigger)
def generate_report(report_date: datetime, report_type: str) -> str:
    return f"Generated {report_type} for {report_date.strftime('%Y-%m-%d')}"

Required vs optional parameters

If your task has parameters without default values, you must provide values for them in the trigger inputs, otherwise the trigger will fail to execute.

        
    
# ❌ This will fail - missing required parameter 'data_source'
bad_trigger = flyte.Trigger(
    "bad_trigger",
    flyte.Cron("0 0 * * *")
    # Missing inputs for required parameter 'data_source'
)

@env.task(triggers=bad_trigger)
def bad_trigger_taska(data_source: str, batch_size: int = 100) -> str:
    return f"Processing from {data_source} with batch size {batch_size}"

# ✅ This works - all required parameters provided
good_trigger = flyte.Trigger(
    "good_trigger",
    flyte.Cron("0 0 * * *"),
    inputs={
        "data_source": "prod_database",  # Required parameter
        "batch_size": 500  # Override default
    }
)

@env.task(triggers=good_trigger)
def good_trigger_task(data_source: str, batch_size: int = 100) -> str:
    return f"Processing from {data_source} with batch size {batch_size}"

Complex input types

You can pass various data types through trigger inputs:

triggers.py
                
                    
                
            
                
            

                
                    
                
            
complex_trigger = flyte.Trigger(
    "ml_training",
    flyte.Cron("0 2 * * 1"),  # Weekly on Monday at 2 AM
    inputs={
        "model_config": {
            "learning_rate": 0.01,
            "batch_size": 32,
            "epochs": 100
        },
        "feature_columns": ["age", "income", "location"],
        "validation_split": 0.2,
        "training_date": flyte.TriggerTime
    }
)

@env.task(triggers=complex_trigger)
def train_model(
    model_config: dict,
    feature_columns: list[str],
    validation_split: float,
    training_date: datetime
) -> str:
    return f"Training model with {len(feature_columns)} features on {training_date}"

Multiple triggers per task

You can attach multiple triggers to a single task by providing a list of triggers. This allows you to run the same task on different schedules or with different configurations:

triggers.py
                
@env.task(triggers=[
    flyte.Trigger.hourly(),  # Predefined trigger
    flyte.Trigger.daily(),   # Another predefined trigger
    flyte.Trigger("custom", flyte.Cron("0 */6 * * *"))  # Custom trigger every 6 hours
])
def multi_trigger_task(trigger_time: datetime = flyte.TriggerTime) -> str:
    # Different logic based on execution timing
    if trigger_time.hour == 0:  # Daily run at midnight
        return f"Daily comprehensive processing at {trigger_time}"
    else:  # Hourly or custom runs
        return f"Regular processing at {trigger_time.strftime('%H:%M')}"

You can mix and match trigger types, combining predefined triggers with those that use flyte.Cron, and flyte.FixedRate automations (see Schedule triggers).