Skip to main content

Launch plans, schedules, and fixed inputs

Launch plans in flytekit provide a mechanism to parameterize workflow executions, enforce specific input constraints, and automate runs via schedules. While every workflow is registered with a default launch plan, you can create custom launch plans to define specialized execution configurations.

Creating Launch Plans​

The primary way to define a launch plan is through LaunchPlan.get_or_create. If you do not provide a name, flytekit returns the default launch plan for the workflow. If you provide a name, you can specify additional attributes like default inputs, fixed inputs, and schedules.

from flytekit import workflow, LaunchPlan

@workflow
def my_wf(a: int, b: str = "default") -> str:
    ...

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

# Create a named launch plan with custom defaults
named_lp = LaunchPlan.get_or_create(
    workflow=my_wf,
    name="my_custom_lp",
    default_inputs={"a": 10}
)

Internally, LaunchPlan.get_or_create caches instances in LaunchPlan.CACHE to prevent duplicate definitions. If you attempt to create a launch plan with the same name but different parameters, flytekit raises an AssertionError.

Parameterizing Inputs​

Launch plans allow you to control how inputs are handled at execution time using default_inputs and fixed_inputs.

Default Inputs​

default_inputs provide values that are used if the user does not specify them at launch time. These values can be overridden. In LaunchPlan.create, flytekit merges these with the workflow's signature parameters, giving precedence to the values defined in the launch plan.

Fixed Inputs​

fixed_inputs are values that cannot be changed at launch time. When you define a fixed input, flytekit removes that parameter from the ParameterMap (the set of inputs a user can provide) and stores it in a LiteralMap within the launch plan's _fixed_inputs attribute.

# 'a' is locked to 42 and cannot be changed during execution
fixed_lp = LaunchPlan.get_or_create(
    workflow=my_wf,
    name="fixed_input_lp",
    fixed_inputs={"a": 42}
)

Scheduling Executions​

You can automate workflow runs by attaching a schedule to a launch plan. flytekit supports two primary schedule types: CronSchedule and FixedRate.

Cron Schedules​

CronSchedule uses standard cron expressions or aliases (like @hourly, @daily). Note that the cron_expression parameter is deprecated; you should use the schedule parameter instead.

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

hourly_lp = LaunchPlan.get_or_create(
    workflow=my_wf,
    name="hourly_lp",
    schedule=CronSchedule(schedule="0 * * * *"),
    default_inputs={"a": 1, "b": "scheduled"}
)

Fixed Rate Schedules​

FixedRate schedules execute at a regular interval defined by a datetime.timedelta. The minimum supported granularity is one minute.

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

every_ten_minutes_lp = LaunchPlan.get_or_create(
    workflow=my_wf,
    name="ten_min_lp",
    schedule=FixedRate(duration=timedelta(minutes=10))
)

Kickoff Time Input​

If your workflow needs to know exactly when a scheduled execution was triggered, use the kickoff_time_input_arg. This passes the schedule's trigger time into a specific workflow input.

from datetime import datetime
from flytekit import workflow, LaunchPlan
from flytekit.core.schedule import CronSchedule

@workflow
def process_data(kickoff_time: datetime):
    ...

scheduled_lp = LaunchPlan.get_or_create(
    workflow=process_data,
    name="kickoff_aware_lp",
    schedule=CronSchedule(
        schedule="@daily",
        kickoff_time_input_arg="kickoff_time"
    )
)

Referencing Existing Launch Plans​

When you need to trigger a launch plan that is already registered on a Flyte cluster (potentially in a different project or domain), use ReferenceLaunchPlan or the @reference_launch_plan decorator. These objects act as pointers and do not require the full workflow implementation, only the interface definition.

from flytekit import reference_launch_plan

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

The ReferenceLaunchPlan constructor requires the explicit project, domain, name, and version, along with the expected inputs and outputs type maps. This allows flytekit to perform type checking during compilation without fetching the entity from the Flyte Admin service.