Skip to main content

Launch plans, schedules, and fixed inputs

What a launch plan is​

A launch plan is the executable wrapper Flyte places around a workflow. It lets you pre-bake how a workflow runs: supplying default and fixed input values, attaching a schedule or notification, and decorating executions with labels, annotations, auth roles, and parallelism bounds. Where the workflow definition declares what runs, the launch plan declares how invocations of that workflow are parameterized and triggered.

The core type is LaunchPlan in flytekit/core/launch_plan.py. It stores:

  • _parameters — a ParameterMap describing the workflow's inputs, including any defaults.
  • _fixed_inputs — a LiteralMap of inputs that callers cannot change.
  • _saved_inputs — the same defaults plus fixed values as Python native values, used for local execution and node creation.
  • _schedule, _trigger, _notifications, _labels, _annotations, _max_parallelism, _security_context, _overwrite_cache, _auto_activate — the execution-time knobs.

Because every workflow is registered with a default launch plan (one with no defaults, fixed values, schedules, or notifications), you almost always create launch plans through a single factory entry point.

Creating a launch plan (get_or_create vs create)​

The friendliest way to make a launch plan is LaunchPlan.get_or_create(). With no name it produces the workflow's default launch plan:

from flytekit import workflow, LaunchPlan

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

lp = LaunchPlan.get_or_create(workflow=my_wf)

The moment you add any other property, you must also give the launch plan a unique name. The name, combined with project, domain, and version, forms the launch plan's primary key. get_or_create enforces the default-plan restriction directly — omitting name while passing default_inputs, fixed_inputs, schedule, notifications, or any of the other knobs raises ValueError:

"Only named launchplans can be created that have other properties. Drop the name if you want to create a default launchplan. Default launchplans cannot have any other associations"

A named launch plan layers on defaults, fixed values, and schedules:

lp = LaunchPlan.get_or_create(
    workflow=wf,
    name="your_lp_name_1",
    default_inputs={"a": 3},
    fixed_inputs={"c": "4"},
)

Both get_or_create and the lower-level create populate LaunchPlan.CACHE, a module-level dict keyed by launch plan name. get_or_create returns the cached instance when you call it again with the same name — but it guards against conflicting definitions. If the cached plan belongs to a different workflow, or any of schedule, notifications, default_inputs, labels, annotations, raw_output_data_config, max_parallelism, security_context, overwrite_cache, or auto_activate differs from the cached values, it raises AssertionError and tells you to use a different name. create() is stricter: it raises AssertionError outright if the name is already in the cache.

Internally, get_or_create delegates to LaunchPlan.get_default_launch_plan(ctx, workflow) when no name is given, and to LaunchPlan.create(name, workflow, ...) otherwise. get_default_launch_plan also copies the workflow's default_options labels and annotations onto the plan, so a plan created with no explicit labels still inherits them from workflow.default_options.

Default inputs vs fixed inputs​

The distinction between the two is the heart of launch-plan parameterization, and create() shows it clearly.

Default inputs are expressed as Python native values and become serialized defaults in the plan's ParameterMap. They come from two sources: the workflow function's own signature defaults, and the default_inputs= argument — with the argument taking precedence. create() first turns the whole workflow signature into a ParameterMap via transform_inputs_to_parameters(ctx, workflow.python_interface) (imported from flytekit.core.interface), then builds a temporary interface from just default_inputs and serializes those into parameters too:

from flytekit.core.interface import transform_inputs_to_parameters
from flytekit.core.interface import Interface

# 1) Turn the whole workflow signature into a ParameterMap.
wf_signature_parameters = transform_inputs_to_parameters(ctx, workflow.python_interface)

# 2) Build a temp interface from just default_inputs so they can be
#    serialized into Parameters.
temp_inputs = {}
for k, v in default_inputs.items():
    temp_inputs[k] = (workflow.python_interface.inputs[k], v)
temp_interface = Interface(inputs=temp_inputs, outputs={})
temp_signature = transform_inputs_to_parameters(ctx, temp_interface)

The source then folds the temporary parameters over the signature's via wf_signature_parameters._parameters.update(temp_signature.parameters), so a name present in both resolves to the default_inputs value. A caller can still override a default at launch time — the default is just the fallback value. In the serialized spec it becomes a Parameter with a default literal, as the test for LaunchPlan.create("get_or_create2", wf, default_inputs={"a": 3}) confirms by asserting lp_with_defaults.parameters.parameters["a"].default.scalar.primitive.integer == 3.

Fixed inputs are the opposite. create() converts them to a LiteralMap via translate_inputs_to_literals, and the LaunchPlan.__init__ guts them out of the parameter map entirely:

parameters = {k: v for k, v in parameters.items() if k not in fixed_inputs.literals}
self._parameters = _interface_models.ParameterMap(parameters=parameters)

Because they're gone from the parameter map, they cannot be supplied or overridden at launch time. The fixed value is baked into the launch plan and always fed to the workflow.

For convenience, create() also stores both as native Python values in _saved_inputs by merging default_inputs.update(fixed_inputs). The comment explains why: the literal forms exist for protobuf serialization, but keeping the original Python values avoids translating them back every time a launch plan is invoked locally or linked into a node. The saved_inputs property returns a copy of that dict so a caller who mutates it doesn't corrupt the plan.

Calling a launch plan inside a workflow​

Once you have a launch plan, you invoke it the same way you call a task or subworkflow — with keyword arguments:

@workflow
def my_wf(a: int) -> typing.Tuple[int, int]:
    t = t1(a=a)
    w = lp(a=a)   # lp is a LaunchPlan for my_sub_wf
    return t, w

Calling lp(...) goes through LaunchPlan.__call__. It only accepts keyword arguments — passing any positional argument raises AssertionError("Only Keyword Arguments are supported for launch plan executions"). What happens next depends on the current context:

  • During compilation (a workflow is being built into a graph), it merges self.saved_inputs — the defaults and fixed values as Python values — with the call-time kwargs, then hands everything to create_and_link_node(ctx, entity=self, **inputs) in flytekit/core/promise.py.
  • During local execution it does the same merge and forwards straight to the wrapped workflow: self.workflow(*args, **inputs).

create_and_link_node iterates over the entity's python_interface.inputs, assigning a binding for each; the defaults and fixed values ride along as literal bindings on the node rather than as workflow-input bindings. That's why a workflow function's own default parameters still flow through — the test test_lp_default_paramaters_work_when_called_from_another_workflow declares def my_sub_wf(a: int, b: int = 5) and then asserts my_wf(a=8) == 13, with b satisfied by the default.

The result is a node whose flyte_entity is the launch plan. The serialization test asserts this shape directly — the node's workflow_node.launchplan_ref.resource_type equals identifier_models.ResourceType.LAUNCH_PLAN and its launchplan_ref.name == "my_sub_wf_lp1". Multi-output launch plans produce multiple Promises, returned as a named tuple, mirroring tasks and subworkflows.

The same wiring is available imperatively through WorkflowBase: my_wf.add_launch_plan(lp, **kwargs) forwards to add_entity, which opens a compilation state and calls create_node(entity=lp, **kwargs) in flytekit/core/node_creation.py. create_node rejects positional args with the same "Only keyword args are supported" guard, then attaches each output promise as a named attribute on the returned Node.

Launch plans in dynamic tasks​

There's one wrinkle for launch plans used inside @dynamic tasks. Because a dynamic task's structure is only discovered at runtime, flytekit can't automatically see that the launch plan must be registered on flyteadmin before the workflow can run. The @dynamic decorator's node_dependency_hints parameter exists for this:

@workflow
def workflow0():
    ...

launchplan0 = LaunchPlan.get_or_create(workflow0)

# Specify node_dependency_hints so that launchplan0 will be registered on flyteadmin, despite this being a
# dynamic task.
@dynamic(node_dependency_hints=[launchplan0])
def launch_dynamically():
    # To run a sub-launchplan it must have previously been registered on flyteadmin.
    return [launchplan0]*10

node_dependency_hints accepts tasks, launch plans, and workflows, and is validated to only be set on dynamic tasks (python_function_task.py raises if it's used on a static task).

Schedules: CronSchedule and FixedRate​

Flyte schedules a launch plan into periodic executions. Both schedule classes live in flytekit/core/schedule.py and subclass flytekit.models.schedule.Schedule.

CronSchedule​

CronSchedule runs the plan on a cron expression or a convenience alias:

from flytekit.core.schedule import CronSchedule

minutely = CronSchedule(schedule="*/1 * * * *")
hourly   = CronSchedule(schedule="hourly")

The schedule value is validated by _validate_schedule: it's checked case-insensitively against _VALID_CRON_ALIASES ("hourly", "daily", "weekly", "monthly", "yearly", and their @-prefixed variants), and anything else must parse through croniter.croniter. A bad value raises ValueError.

There are two things to watch out for. First, the older cron_expression argument is deprecated — passing it raises AssertionError with "cron_expression is deprecated and should not be used. Use schedule instead." This trip also catches positional usage like CronSchedule("* * ? * * *"), which lands in the cron_expression parameter. Second, an optional offset is accepted as an ISO 8601 duration (for example "P1D") and validated by _validate_offset against the regex pattern ([-+]?)P([-+0-9YMWD]+)?(T([-+0-9HMS.,]+)?)?; anything that doesn't fully match raises ValueError.

FixedRate​

FixedRate runs the plan every timedelta. The constructor converts the duration into a Schedule.FixedRate(value, unit) where the unit is chosen by divisibility — DAY for whole multiples of 24 hours, HOUR for whole multiples of 60 minutes, otherwise MINUTE:

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

FixedRate(duration=timedelta(minutes=10))
FixedRate(duration=timedelta(hours=12))    # unit=HOUR, value=12
FixedRate(duration=timedelta(hours=24))    # unit=DAY, value=1

Sub-minute granularity is rejected: if the duration has microseconds or a seconds remainder that isn't a whole multiple of 60, _translate_duration raises AssertionError ("Granularity of less than a minute is not supported for FixedRate schedules").

Attaching a schedule to a plan​

Pass either object as schedule= when creating the plan. The schedule is stored as the plan's _schedule and, on serialization, becomes part of the admin LaunchPlanSpec. The test test_schedule_with_lp shows the round trip:

lp = LaunchPlan.create(
    "schedule_test",
    quadruple,
    schedule=FixedRate(datetime.timedelta(hours=12), "kickoff_input"),
)
assert lp.schedule == _schedule_models.Schedule(
    "kickoff_input", rate=_schedule_models.Schedule.FixedRate(12, _schedule_models.Schedule.FixedRateUnit.HOUR)
)

Knowing the kickoff time​

Both CronSchedule and FixedRate accept kickoff_time_input_arg, the name of a workflow input parameter that should receive the time a scheduled run was kicked off. This is how your code learns when it's running. The docstring's example wires it to a datetime workflow input:

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

schedule = CronSchedule(
    schedule="*/1 * * * *",
    kickoff_time_input_arg="kickoff_time",
)

The value is forwarded to the underlying model Schedule constructor. Because Flyte has no atomic clock, the kickoff time may be a few seconds off the scheduled minute.

Triggers (alpha): OnSchedule​

Alongside the established schedule= argument there's a newer, alpha trigger= syntax. At the top of schedule.py, LaunchPlanTriggerBase is a typing.Protocol requiring a single method, to_flyte_idl(*args, **kwargs) -> google_message.Message. The trigger parameter on LaunchPlan.create/get_or_create is typed against it and documented as "[alpha] This is a new syntax for specifying schedules."

OnSchedule is the concrete implementation. You wrap a CronSchedule or FixedRate and hand it to trigger=:

from flytekit.core.schedule import CronSchedule, OnSchedule

lp = LaunchPlan.create(
    "your_lp_name",
    my_wf,
    trigger=OnSchedule(CronSchedule(schedule="*/1 * * * *")),
)

OnSchedule.to_flyte_idl() simply delegates to the wrapped schedule, producing a schedule_pb2.Schedule:

def to_flyte_idl(self) -> schedule_pb2.Schedule:
    return self._schedule.to_flyte_idl()

Reference launch plans​

If a launch plan already exists on your Flyte installation, you don't have to redefine it. reference_launch_plan(project, domain, name, version) builds a ReferenceLaunchPlan — a pointer that makes no network call to Admin. The decorated function's body is never executed; only its type signature is read to establish the expected interface:

from flytekit.core.launch_plan import reference_launch_plan

@reference_launch_plan(
    project="flytesnacks",
    domain="development",
    name="my.registered.launch.plan",
    version="abc123",
)
def ref_lp(a: int) -> str:
    # The body is ignored; only the signature matters.
    return "hello"

Internally, the wrapper in reference_launch_plan calls transform_function_to_interface(fn, is_reference_entity=True) to pull the inputs/outputs off the signature and constructs ReferenceLaunchPlan(project, domain, name, version, interface.inputs, interface.outputs). The body is skipped during interface extraction because is_reference_entity=True disables the return-statement checks in transform_function_to_interface. If the interface you declare doesn't match the remote plan, compilation of any workflow that calls it fails at registration time.

ReferenceLaunchPlan inherits from both ReferenceEntity and LaunchPlan. The same object is produced programmatically by get_reference_entity(ResourceType.LAUNCH_PLAN, project, domain, name, version, inputs, outputs) in flytekit/core/reference.py, which returns ReferenceLaunchPlan for the LAUNCH_PLAN resource type (and ReferenceTask / ReferenceWorkflow for the others).

Configuration and gotchas​

Several behaviors are easy to trip over:

  • Default plans can't have associations. get_or_create(workflow=wf) without a name rejects every other argument with ValueError.
  • Names must be unique. create() raises AssertionError if the name is already cached; get_or_create() raises if the cached plan disagrees on workflow or any of its properties.
  • __call__ and create_node are keyword-only. Positional arguments raise AssertionError, matching the behavior of tasks and workflows.
  • Fixed inputs disappear from the parameter map. Supplying a fixed input at call time is impossible — it isn't in _parameters, and the fixed literal is always fed through _saved_inputs.
  • Mutable defaults are rejected in node creation. create_and_link_node (promise.py) raises FlyteAssertion if a workflow input default is a list, dict, or set, since those are Python anti-patterns as defaults.
  • auto_activate defaults to False. When True, the plan is activated automatically on registration; the value is exposed through should_auto_activate.
  • Fixed inputs are excluded from array-node interfaces. ArrayNode (array_node.py) excludes target.fixed_inputs.literals from the per-element interface — a fixed input can't be passed per mapped element — but skips this exclusion for ReferenceLaunchPlan targets since their fixed inputs aren't known without a network call.
  • Every plan registers itself globally. LaunchPlan.__init__ appends the instance to FlyteEntities.entities, the global list the serializer walks to find and emit all declared entities.
  • Arguments are normalized before caching. get_or_create coerces notifications to [] and merges fixed_inputs into default_inputs before the deep-equality comparison, so calling it twice with the same values returns the cached plan rather than raising.