Launch plans, schedules, and fixed inputs

Launch plans in flytekit allow you to parameterize workflow executions, set fixed inputs that cannot be changed at runtime, and define schedules for automated runs. While every workflow is registered with a default launch plan, creating custom launch plans enables you to reuse the same workflow logic with different configurations.

Creating Launch Plans

You create launch plans using the LaunchPlan.get_or_create method. If you only need the default launch plan (which uses the workflow's default parameters and has no schedule), you can omit the name. However, any launch plan with custom inputs or schedules must be named.

from flytekit import workflow, LaunchPlan

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

# Default launch plan (unnamed)
default_lp = LaunchPlan.get_or_create(workflow=my_wf)

# Custom named launch plan
custom_lp = LaunchPlan.get_or_create(
    name="my_custom_lp",
    workflow=my_wf,
    default_inputs={"a": 10}
)

Internally, LaunchPlan.get_or_create (found in launch_plan.py) manages a cache of launch plans to ensure that multiple calls for the same workflow/name combination return the same object. If you provide a name and parameters that differ from a cached version, flytekit raises an AssertionError to prevent configuration conflicts.

Parameterizing Inputs

Launch plans support two types of input parameterization: default_inputs and fixed_inputs.

Default Inputs

default_inputs provide values that are used if the caller does not specify them. They can be overridden at execution time.

lp = LaunchPlan.get_or_create(
    name="overridable_lp",
    workflow=my_wf,
    default_inputs={"a": 42}
)

Fixed Inputs

fixed_inputs are "locked" values that cannot be changed when the launch plan is invoked. If you try to provide a value for a fixed input during execution, Flyte will reject the request.

lp = LaunchPlan.get_or_create(
    name="locked_lp",
    workflow=my_wf,
    fixed_inputs={"a": 100}
)

When you define fixed_inputs, flytekit's LaunchPlan.create method uses translate_inputs_to_literals to convert your Python values into Flyte's internal LiteralMap format. It also removes these keys from the ParameterMap so they are no longer visible as overridable inputs in the Flyte UI or CLI.

Scheduling Executions

To run a workflow automatically at specific intervals, you attach a schedule to a launch plan. flytekit provides two primary schedule types in schedule.py: CronSchedule and FixedRate.

Cron Schedules

CronSchedule uses standard cron expressions or aliases (like @daily or @hourly).

from flytekit import CronSchedule

daily_lp = LaunchPlan.get_or_create(
    name="daily_run",
    workflow=my_wf,
    schedule=CronSchedule(
        schedule="0 0 * * *",  # Runs every day at midnight
        kickoff_time_input_arg="kickoff_time" # Optional: maps trigger time to a workflow input
    ),
    default_inputs={"a": 1}
)

Fixed Rate Schedules

FixedRate schedules run at a consistent frequency defined by a timedelta.

from datetime import timedelta
from flytekit import FixedRate

frequent_lp = LaunchPlan.get_or_create(
    name="every_ten_minutes",
    workflow=my_wf,
    schedule=FixedRate(duration=timedelta(minutes=10)),
    default_inputs={"a": 5}
)

The FixedRate class validates that your duration is at least one minute, as Flyte does not support sub-minute granularity. It automatically translates the timedelta into the appropriate FixedRateUnit (MINUTE, HOUR, or DAY) during initialization.

Advanced Configuration

Launch plans also serve as a container for execution-time metadata and security settings:

• Notifications: Use the notifications parameter to send alerts (Email, Slack, PagerDuty) on execution events.
• Security Context: Use security_context to define the IAM role or Kubernetes service account the execution should assume. Note that the older auth_role parameter is deprecated in favor of security_context.
• Labels and Annotations: Attach labels and annotations to the resulting executions for tracking and organization.
• Auto-Activation: Set auto_activate=True to ensure the schedule is active immediately upon registration.

Referencing Existing Launch Plans

If you need to trigger a launch plan that is already registered on a Flyte cluster without re-defining it in your current code, use ReferenceLaunchPlan or the @reference_launch_plan decorator.

from flytekit import reference_launch_plan

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

This creates a pointer to the remote entity. flytekit will validate that the interface you define in the decorated function matches the interface of the remote launch plan during registration.