Launch plans, schedules, and fixed inputs

Launch plans are the core Flyte construct for parameterizing workflow execution. They let you define default inputs (overridable at launch time), fixed inputs (immutable at launch time), schedules, and notifications. Every workflow automatically gets a default launch plan, but you can create named launch plans with additional configuration using LaunchPlan.get_or_create() or LaunchPlan.create().

Creating launch plans

Use LaunchPlan.get_or_create() for most cases. It caches launch plans by name to prevent duplication and provides a friendly interface.

from flytekit import LaunchPlan, workflow

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

# Default launch plan (no name, no extra args)
default_lp = LaunchPlan.get_or_create(workflow=my_wf)

# Named launch plan with default and fixed inputs
named_lp = LaunchPlan.get_or_create(
    workflow=my_wf,
    name="my_named_lp",
    default_inputs={"a": 42},  # Overridable at launch time
    fixed_inputs={"c": "fixed_value"}  # Cannot be changed at launch time
)

If you omit the name parameter, you cannot specify any other parameters (schedule, notifications, labels, etc.). Default launch plans are automatically named after the workflow and cannot have additional associations.

Named launch plans can have schedules, notifications, labels, annotations, security contexts, and other properties. The get_or_create() method caches launch plans by name and validates that subsequent calls with the same name don't conflict:

# This will return the cached launch plan from above
cached_lp = LaunchPlan.get_or_create(
    workflow=my_wf,
    name="my_named_lp",
    default_inputs={"a": 42},
    fixed_inputs={"c": "fixed_value"}
)

If you try to create a second launch plan with the same name but different parameters, you'll get an AssertionError listing the conflicting fields:

# This raises AssertionError: "Trying to create two launch plans ... but with different values for 'default_inputs'"
LaunchPlan.get_or_create(
    workflow=my_wf,
    name="my_named_lp",
    default_inputs={"a": 99},  # Different default
    fixed_inputs={"c": "fixed_value"}
)

For advanced use cases, you can use LaunchPlan.create() directly, but get_or_create() is recommended because it handles caching and validation.

Default vs fixed inputs

Default inputs are values you provide at launch plan creation time that can be overridden when you execute the launch plan. Fixed inputs are values that cannot be changed at launch time.

lp = LaunchPlan.get_or_create(
    workflow=my_wf,
    name="inputs_example",
    default_inputs={"a": 10},  # Can be overridden: lp(a=20)
    fixed_inputs={"c": "immutable"}  # Cannot be overridden
)

Internally, fixed inputs are removed from the parameter map (see line 338 in launch_plan.py):

parameters = {k: v for k, v in parameters.items() if k not in fixed_inputs}

This ensures fixed inputs cannot be passed as keyword arguments when calling the launch plan. The saved_inputs property returns a copy of both default and fixed inputs for local execution:

# Returns {'a': 10, 'c': 'immutable'} (both default and fixed)
saved = lp.saved_inputs

Schedules

Launch plans can be scheduled to run automatically using CronSchedule or FixedRate.

Cron schedules

Use CronSchedule for cron-based execution. The native scheduler uses the schedule parameter with 5-field cron format or cron aliases:

from flytekit import CronSchedule

# Using cron alias
hourly_schedule = CronSchedule(schedule="hourly")

# Using cron expression (5 fields for native scheduler)
every_10_minutes = CronSchedule(schedule="*/10 * * * *")

# With offset (ISO 8601 duration)
offset_schedule = CronSchedule(schedule="daily", offset="PT1H")  # 1 hour after daily trigger

# With kickoff time input
@workflow
def my_wf(kickoff_time: datetime.datetime):
    ...

kickoff_schedule = CronSchedule(
    schedule="*/1 * * * *",
    kickoff_time_input_arg="kickoff_time"
)

The cron_expression parameter is deprecated and raises AssertionError if used. Use schedule instead.

Fixed rate schedules

Use FixedRate for fixed-interval execution. The duration must be at least one minute; sub-minute granularity raises AssertionError:

from datetime import timedelta
from flytekit import FixedRate

# Every 10 minutes
every_10_min = FixedRate(duration=timedelta(minutes=10))

# Every 2 hours
every_2_hours = FixedRate(duration=timedelta(hours=2))

# Every 3 days
every_3_days = FixedRate(duration=timedelta(days=3))

The _translate_duration() method automatically converts the duration to the appropriate unit (days, hours, or minutes) based on divisibility.

Using schedules with launch plans

Pass the schedule to get_or_create() or create():

lp = LaunchPlan.get_or_create(
    workflow=my_wf,
    name="scheduled_lp",
    schedule=CronSchedule(schedule="daily"),
    notifications=[...]  # Optional notifications
)

The OnSchedule trigger wrapper exists but is not used anywhere in the codebase — no call sites found for trigger=OnSchedule(...).

Reference launch plans

Use reference_launch_plan() to point to an existing launch plan registered on Flyte Admin. This doesn't call Admin at runtime; instead, it uses the interface you provide via the function signature for compilation:

from flytekit import reference_launch_plan

@reference_launch_plan(project="my-project", domain="prod", name="existing_lp", version="v1")
def reference_wf(a: int, c: str) -> str:
    ...

# reference_wf is now a ReferenceLaunchPlan that points to the remote launch plan

If the interface you provide doesn't match the remote launch plan, you'll get a compilation error at registration time.

ArrayNode mapping over launch plans

When mapping over a launch plan with array_node(), fixed inputs are automatically excluded from the mapped inputs to prevent them from being overridden per-mapped-instance:

from flytekit import array_node

@array_node(target=lp, concurrency=5)
def map_over_lp(values: List[int]):
    # lp.fixed_inputs are excluded from mapping
    return lp(a=values[i])  # Can only override non-fixed inputs

Launch plan execution

Calling a launch plan only supports keyword arguments — positional arguments raise AssertionError:

# This works
result = lp(a=5, c="override")

# This raises AssertionError: "Only Keyword Arguments are supported for launch plan executions"
result = lp(5, "override")

When called during compilation (inside a workflow), it creates a node. When called outside compilation (local execution), it forwards to the workflow with saved inputs merged with provided kwargs.