Launch plans, schedules, and fixed inputs

Launch plans in flytekit allow you to parameterize workflow executions, apply default or fixed inputs, and define schedules or triggers. While every workflow is registered with a default launch plan, creating custom launch plans enables you to reuse the same workflow logic with different configurations or automated execution patterns.

Creating Launch Plans

You create a launch plan by associating it with a workflow. If you do not provide a name, flytekit assumes you are requesting the default launch plan for that workflow. If you provide additional parameters like inputs or schedules, you must provide a unique name.

The primary entry point is LaunchPlan.get_or_create.

from flytekit import workflow, LaunchPlan

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

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

# Create a named launch plan with specific inputs
named_lp = LaunchPlan.get_or_create(
    workflow=my_wf,
    name="my_named_lp",
    default_inputs={"a": 10, "b": "hello"}
)

Internally, LaunchPlan.get_or_create (found in flytekit/core/launch_plan.py) manages a cache of launch plans to prevent duplicate creation. If a name is provided, it calls LaunchPlan.create, which handles the translation of Python native types into Flyte literals using translate_inputs_to_literals.

Parameterizing Inputs

Launch plans support two types of input overrides:

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

lp_with_fixed = LaunchPlan.get_or_create(
    workflow=my_wf,
    name="fixed_input_lp",
    fixed_inputs={"a": 42},
    default_inputs={"b": "default value"}
)

When you define fixed_inputs, the LaunchPlan constructor removes these keys from the parameters map (the set of inputs available for user override) and stores them in _fixed_inputs. This ensures that the Flyte engine enforces the immutability of these values during execution.

Scheduling Executions

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

Cron Schedules

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

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

daily_lp = LaunchPlan.get_or_create(
    workflow=my_wf,
    name="daily_execution",
    schedule=CronSchedule(
        schedule="0 0 * * *",  # Runs every day at midnight
    )
)

The CronSchedule class (in flytekit/core/schedule.py) validates the schedule string using croniter. It also supports a kickoff_time_input_arg, which allows you to pass the scheduled time into a specific workflow input.

Fixed Rate Schedules

FixedRate is used for intervals defined by a datetime.timedelta.

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

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

FixedRate internally translates the timedelta into a FixedRateUnit (MINUTE, HOUR, or DAY). Note that flytekit enforces a minimum granularity of one minute; providing a timedelta with seconds or microseconds will raise an AssertionError in FixedRate._translate_duration.

Triggers and Notifications

Beyond schedules, launch plans can be configured with triggers and notifications.

• Triggers: The trigger argument accepts a LaunchPlanTriggerBase. While OnSchedule is the standard wrapper for schedules, this interface is designed to support future trigger types.
• Notifications: You can pass a list of Notification models to alert users on execution success or failure.

from flytekit.models.common import Notification
from flyteidl.admin.common_pb2 import EmailNotification

email_notification = Notification(
    phases=[2, 3],  # Notify on SUCCEEDED and FAILED
    email=EmailNotification(recipients_list=["user@example.com"])
)

lp_with_notify = LaunchPlan.get_or_create(
    workflow=my_wf,
    name="notifying_lp",
    notifications=[email_notification]
)

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 or the @reference_launch_plan decorator. This allows you to reference the entity by its project, domain, name, and version without needing the original source code.

from flytekit import reference_launch_plan

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

The ReferenceLaunchPlan class (in flytekit/core/launch_plan.py) acts as a pointer and does not initiate network calls during instantiation, relying instead on the provided interface for compilation.