Launch plans, schedules, and fixed inputs

Launch plans in flytekit are the primary mechanism for parameterizing workflow executions. While a workflow defines a template of logic, a launch plan provides the specific configuration needed to run that logic—including default or fixed inputs, schedules, and notifications.

Every workflow registered in flytekit automatically receives a default launch plan. This default plan uses the workflow's function signature to determine inputs and has no additional scheduling or notification logic.

Creating Launch Plans

You create launch plans using the LaunchPlan.get_or_create method. If you do not provide a name, flytekit assumes you want the default launch plan for that workflow.

from flytekit import workflow, LaunchPlan

@workflow
def my_wf(a: int, c: str) -> str:
    ...

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

If you need to customize inputs or add a schedule, you must provide a unique name. The LaunchPlan class maintains an internal CACHE to ensure that multiple calls for the same named plan return the same object, preventing conflicting definitions.

Default and Fixed Inputs

Launch plans allow you to pre-configure workflow inputs in two ways:

1. Default Inputs: These provide values that the user can still override at execution time.
2. Fixed Inputs: These are "locked" values that cannot be changed when the launch plan is invoked.

# A launch plan with both default and fixed inputs
custom_lp = LaunchPlan.get_or_create(
    name="frequent_execution_plan",
    workflow=my_wf,
    default_inputs={"a": 10},
    fixed_inputs={"c": "constant_value"}
)

Internally, LaunchPlan.create processes these inputs by transforming them into Flyte's internal literal types. It uses transform_inputs_to_parameters for defaults and translate_inputs_to_literals for fixed values. If a key exists in both default_inputs and fixed_inputs, the fixed value takes precedence and the key is removed from the parameter map available for user input.

Scheduling Executions

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

Cron Schedules

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

from flytekit import LaunchPlan, CronSchedule

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

The CronSchedule class validates the expression using the croniter library. Note that the cron_expression parameter is deprecated in favor of schedule.

Fixed Rate Schedules

FixedRate schedules execute at a consistent interval defined by a datetime.timedelta.

from datetime import timedelta
from flytekit import LaunchPlan, FixedRate

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

The FixedRate implementation in schedule.py enforces a minimum granularity of one minute. If you provide a duration with seconds or microseconds that do not align with a full minute, flytekit will raise an AssertionError.

Capturing Kickoff Time

If your workflow logic needs to know exactly when it was triggered by a schedule, use the kickoff_time_input_arg parameter. This maps the schedule's trigger time to a specific input in your workflow.

from datetime import datetime
from flytekit import workflow, LaunchPlan, CronSchedule

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

scheduled_lp = LaunchPlan.get_or_create(
    name="time_aware_plan",
    workflow=report_wf,
    schedule=CronSchedule(
        schedule="@hourly",
        kickoff_time_input_arg="kickoff_time"
    )
)

Advanced Configuration

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

• Security Context: Use the security_context parameter to define the IAM role or Kubernetes service account the execution should run as. This replaces the deprecated auth_role.
• Notifications: You can pass a list of Notification objects to trigger alerts (e.g., email or Slack) on execution success or failure.
• Labels and Annotations: These allow you to attach metadata to the underlying Kubernetes pods or Flyte execution objects.

Reference Launch Plans

When you need to trigger a launch plan that is already registered on a Flyte cluster from a different project or codebase, use ReferenceLaunchPlan. This allows you to reference the entity by its project, domain, name, and version without needing the original source code.

from flytekit import ReferenceLaunchPlan

existing_lp = ReferenceLaunchPlan(
    project="flytesnacks",
    domain="development",
    name="my_wf_lp",
    version="v1",
    inputs={"a": int, "c": str},
    outputs={"o0": str}
)

This creates a pointer that flytekit can use during compilation and registration to link nodes to the existing remote entity.