A launch plan links a partial or complete list of inputs to a workflow, along with optional run-time overrides like notifications and schedules. Launch plans serve several purposes:
- Run the same workflow multiple times with different predefined inputs
- Attach schedules or notifications to a workflow without modifying it
- Share a workflow with fixed inputs that cannot be overridden
- Let another user trigger a workflow while allowing them to override specific inputs
Launch plans are the only mechanism for invoking workflow executions. When a workflow is registered, Flyte automatically creates a default launch plan for it.
Create a default launch plan
Import LaunchPlan and the workflow you want to wrap:
from flytekit import LaunchPlan
from workflow import simple_wf
Create a default launch plan with no pre-bound inputs:
default_lp = LaunchPlan.get_or_create(workflow=simple_wf)
Run it locally by calling it like a function — you must supply all required inputs:
default_lp(x=[-3, 0, 3], y=[7, 4, -2])
Bind default input values that are used when the caller does not provide them:
lp_with_defaults = LaunchPlan.get_or_create(
workflow=simple_wf,
name="simple_wf_with_defaults",
default_inputs={"x": [-3, 0, 3], "y": [7, 4, -2]},
)
Trigger the launch plan without specifying inputs:
Override the defaults at call time:
lp_with_defaults(x=[1, 2, 3], y=[4, 5, 6])
Lock specific inputs so they cannot be changed at execution time:
lp_with_fixed = LaunchPlan.get_or_create(
workflow=simple_wf,
name="simple_wf_with_fixed_inputs",
fixed_inputs={"x": [-3, 0, 3]},
)
Attempting to override a fixed input raises an error:
# This will raise an error — x is fixed
lp_with_fixed(x=[1, 2, 3], y=[7, 4, -2])
You can combine default_inputs and fixed_inputs in the same launch plan. Fixed inputs are locked; default inputs can be overridden.
Scheduling
Launch plans can trigger workflows on a schedule. Flyte supports cron-based schedules and fixed-rate intervals.
Cron schedule
from flytekit import LaunchPlan, CronSchedule
from workflow import simple_wf
scheduled_lp = LaunchPlan.get_or_create(
workflow=simple_wf,
name="scheduled_simple_wf",
default_inputs={"x": [-3, 0, 3], "y": [7, 4, -2]},
schedule=CronSchedule(
schedule="0 9 * * MON-FRI", # 9 AM on weekdays
kickoff_time_input_arg=None,
),
)
Fixed rate
from datetime import timedelta
from flytekit import LaunchPlan, FixedRate
from workflow import simple_wf
hourly_lp = LaunchPlan.get_or_create(
workflow=simple_wf,
name="hourly_simple_wf",
default_inputs={"x": [-3, 0, 3], "y": [7, 4, -2]},
schedule=FixedRate(duration=timedelta(hours=1)),
)
Schedules only execute when the launch plan is active. A newly created launch plan is inactive by default. Activate it using flytectl or the Flyte UI.
Activate and deactivate a launch plan
Use flytectl to manage the active state of a launch plan:
# Activate a launch plan
flytectl update launchplan -p my_project -d development \
my_project.workflows.scheduled_simple_wf \
--version v1 --activate
# Deactivate a launch plan
flytectl update launchplan -p my_project -d development \
my_project.workflows.scheduled_simple_wf \
--version v1 --deactivate
Complete example
from flytekit import LaunchPlan, CronSchedule
from workflow import simple_wf
# Default launch plan — no pre-bound inputs
default_lp = LaunchPlan.get_or_create(workflow=simple_wf)
# Launch plan with default inputs
lp_with_defaults = LaunchPlan.get_or_create(
workflow=simple_wf,
name="simple_wf_with_defaults",
default_inputs={"x": [-3, 0, 3], "y": [7, 4, -2]},
)
# Launch plan with fixed inputs
lp_with_fixed = LaunchPlan.get_or_create(
workflow=simple_wf,
name="simple_wf_with_fixed_inputs",
fixed_inputs={"x": [-3, 0, 3]},
)
# Scheduled launch plan
scheduled_lp = LaunchPlan.get_or_create(
workflow=simple_wf,
name="scheduled_simple_wf",
default_inputs={"x": [-3, 0, 3], "y": [7, 4, -2]},
schedule=CronSchedule(schedule="0 9 * * MON-FRI"),
)
if __name__ == "__main__":
# Run locally
print(lp_with_defaults())
print(lp_with_defaults(x=[1, 2, 3], y=[4, 5, 6]))