Launch plans, schedules, and fixed inputs

Launch plans in flytekit provide a mechanism to parameterize workflow executions, allowing you to define default or fixed inputs, set up recurring schedules, and configure execution-time metadata like notifications and resource constraints. While every workflow is automatically registered with a default launch plan, you can create custom launch plans to handle specific operational scenarios, such as daily batch processing or specialized testing environments.

Creating Launch Plans

You create launch plans using the LaunchPlan.get_or_create method in flytekit/core/launch_plan.py. This method ensures that launch plans are cached and reused, preventing redundant entity creation during registration.

Default Launch Plans

A default launch plan uses the workflow's name and inherits all default values defined in the workflow's function signature. It does not include schedules or notifications.

from flytekit import workflow, LaunchPlan

@workflow
def my_workflow(a: int, b: str = "default"):
    ...

# Retrieves or creates the default launch plan for my_workflow
default_lp = LaunchPlan.get_or_create(workflow=my_workflow)

Internally, LaunchPlan.get_default_launch_plan extracts parameters from the workflow's python_interface and populates _saved_inputs with any default values found in the signature.

Custom Launch Plans with Inputs

When you need to override workflow defaults or lock specific inputs, you must provide a unique name for the launch plan.

• Default Inputs: These provide values that can still be overridden at execution time.
• Fixed Inputs: These are immutable at execution time. If a user attempts to provide a different value for a fixed input during launch, Flyte will reject the execution.

# A launch plan with specific defaults and one fixed input
custom_lp = LaunchPlan.get_or_create(
    name="frequent_execution_lp",
    workflow=my_workflow,
    default_inputs={"a": 10},
    fixed_inputs={"b": "fixed_value"}
)

The LaunchPlan.create method handles the translation of these native Python values into Flyte literals using translate_inputs_to_literals. It also ensures that fixed_inputs are removed from the parameters map (which defines the execution-time interface) so they cannot be modified by users.

Scheduling Executions

Flytekit supports two primary ways to schedule launch plans: CronSchedule and FixedRate. These are defined in flytekit/core/schedule.py.

Cron Schedules

CronSchedule supports standard cron expressions or aliases like @daily or @hourly.

from flytekit import CronSchedule, LaunchPlan

daily_lp = LaunchPlan.get_or_create(
    name="daily_report",
    workflow=my_workflow,
    schedule=CronSchedule(schedule="@daily"),
    default_inputs={"a": 1}
)

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

from datetime import datetime

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

scheduled_lp = LaunchPlan.get_or_create(
    name="timed_lp",
    workflow=report_wf,
    schedule=CronSchedule(
        schedule="0 0 * * *",
        kickoff_time_input_arg="kickoff_time"
    )
)

Fixed Rate Schedules

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

from datetime import timedelta
from flytekit import FixedRate, LaunchPlan

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

The FixedRate class validates that the duration is at least one minute. Internally, _translate_duration converts the timedelta into the appropriate FixedRateUnit (MINUTE, HOUR, or DAY) supported by the Flyte IDL.

Triggers and Advanced Configuration

The trigger parameter in LaunchPlan.get_or_create is the modern interface for specifying schedules and other activation mechanisms. You can wrap a schedule in an OnSchedule trigger:

from flytekit import OnSchedule

triggered_lp = LaunchPlan.get_or_create(
    name="triggered_lp",
    workflow=my_workflow,
    trigger=OnSchedule(CronSchedule(schedule="*/5 * * * *"))
)

Execution Metadata

Launch plans also allow you to attach metadata that applies to every execution they trigger:

• Notifications: Send alerts (Email, Slack, PagerDuty) on execution success or failure.
• Labels and Annotations: Attach Kubernetes-style metadata to executions.
• Security Context: Define the IAM role or Kubernetes service account the execution should run as.
• Max Parallelism: Control the maximum number of nodes that can run in parallel across the workflow.

from flytekit.models.common import Labels
from flytekit.models.security import SecurityContext, Identity

secure_lp = LaunchPlan.get_or_create(
    name="secure_lp",
    workflow=my_workflow,
    labels=Labels({"team": "data-science"}),
    security_context=SecurityContext(
        run_as=Identity(iam_role="arn:aws:iam::123456789012:role/flyte-role")
    )
)

Reference Launch Plans

If you need to trigger a launch plan that is already registered on a Flyte cluster from within another workflow, 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="daily_report",
    version="v1",
    inputs={"a": int},
    outputs={"res": str}
)

Alternatively, you can use the @reference_launch_plan decorator to define the interface using a Python function signature, which flytekit uses to validate the connection at registration time.