Skip to main content

Launch plans, schedules, and fixed inputs

Launch plans in flytekit provide a way to parameterize workflow executions, apply fixed or default inputs, and define schedules for recurring runs. While every workflow is registered with a default launch plan, you can create custom ones to lock down specific configurations or automate execution.

Creating Launch Plans​

The LaunchPlan class is the primary interface for defining these execution templates. You typically use LaunchPlan.get_or_create to define a launch plan. If you don't provide a name, flytekit returns the default launch plan for the workflow.

from flytekit import workflow, LaunchPlan

@workflow
def my_wf(a: int, b: str) -> str:
    return f"{b}: {a}"

# Get the default launch plan
default_lp = LaunchPlan.get_or_create(workflow=my_wf)

# Create a named launch plan with specific settings
custom_lp = LaunchPlan.get_or_create(
    name="my_custom_launch_plan",
    workflow=my_wf,
    default_inputs={"a": 10},
    fixed_inputs={"b": "fixed_value"}
)

Internally, LaunchPlan.get_or_create manages a cache (LaunchPlan.CACHE) to ensure that multiple calls for the same named launch plan return the same object. If you attempt to create two launch plans with the same name but different configurations, flytekit raises an AssertionError.

Parameterizing Inputs​

Launch plans distinguish between default_inputs and fixed_inputs:

  • Default Inputs: These provide values that are used if the execution doesn't specify them. They can be overridden at launch time.
  • Fixed Inputs: These values are "locked" into the launch plan. They cannot be changed when triggering an execution from this specific launch plan.

When you define fixed_inputs, flytekit's LaunchPlan.create method uses translate_inputs_to_literals to convert Python native types into Flyte's internal LiteralMap representation. These fixed values are then removed from the ParameterMap that defines the launch plan's external interface, effectively hiding them from the user at execution time.

Scheduling Executions​

To automate workflow runs, you can attach a schedule to a launch plan. flytekit supports two primary schedule types: CronSchedule and FixedRate.

Cron Schedules​

Use CronSchedule when you need complex, calendar-based timing. It supports standard cron aliases (like @daily or @hourly) or full cron expressions.

from flytekit import LaunchPlan
from flytekit.core.schedule import CronSchedule

daily_lp = LaunchPlan.get_or_create(
    name="daily_execution",
    workflow=my_wf,
    schedule=CronSchedule(
        schedule="0 0 * * *",  # Runs every day at midnight
        kickoff_time_input_arg="kickoff_time"
    ),
    default_inputs={"a": 1, "b": "daily"}
)

The kickoff_time_input_arg parameter allows you to pass the scheduled time into the workflow as a datetime object. This is useful for workflows that process data based on the time they were triggered.

Fixed Rate Schedules​

Use FixedRate for simple intervals, such as running every 10 minutes.

from datetime import timedelta
from flytekit import LaunchPlan
from flytekit.core.schedule import FixedRate

interval_lp = LaunchPlan.get_or_create(
    name="interval_execution",
    workflow=my_wf,
    schedule=FixedRate(duration=timedelta(minutes=10)),
    default_inputs={"a": 5, "b": "interval"}
)

Note that FixedRate enforces a minimum granularity of one minute. The _translate_duration method in FixedRate automatically converts the timedelta into the appropriate FixedRateUnit (MINUTE, HOUR, or DAY) required by the Flyte backend.

Triggers and Notifications​

Beyond schedules, launch plans support advanced triggers and notifications. The trigger parameter (currently in alpha) uses the LaunchPlanTriggerBase protocol to define how a launch plan is activated.

You can also configure notifications to alert users when a workflow execution reaches a specific state (e.g., success or failure). These are passed as a list of Notification models:

from flytekit.models.common import Notification
from flytekit.models.core.execution import WorkflowExecutionPhase

# Example notification configuration (requires model imports)
notification = Notification(
    phases=[WorkflowExecutionPhase.SUCCEEDED],
    email_config=...
)

Reference Launch Plans​

If you need to trigger a launch plan that is already registered on a Flyte cluster from within another workflow, use ReferenceLaunchPlan. This acts as a pointer and does not require the full workflow definition, only the expected interface.

from flytekit import reference_launch_plan

@reference_launch_plan(
    project="flytesnacks",
    domain="development",
    name="my_existing_lp",
    version="v1"
)
def existing_lp(a: int, b: str) -> str:
    ...

The reference_launch_plan decorator uses transform_function_to_interface to validate that the local Python signature matches the interface of the remote launch plan during compilation.