Launch plans, schedules, and fixed inputs

Launch plans as parameterized workflow executions

If you need to register one workflow with a reusable execution configuration—such as a set of defaults, inputs that callers must not change, or a schedule—use a named LaunchPlan. A workflow also has a default launch plan, but that plan is deliberately minimal: it has the workflow’s interface and signature defaults, without launch-plan-specific defaults, fixed inputs, schedules, or notifications.

For a workflow such as the one shown in the LaunchPlan source docstring, retrieve that default plan by omitting name:

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

LaunchPlan.get_or_create(workflow=my_wf)

The docstring’s workflow is named wf, while its example call uses my_wf; when using the pattern, pass the actual workflow object you defined. LaunchPlan.get_or_create(workflow=...) caches the default under the workflow name. Calling it again for that workflow returns the cached plan rather than creating another default.

To add launch-plan-specific behavior, provide a unique name and pass the configuration to get_or_create (or use the lower-level LaunchPlan.create). For example, the public signatures support this combination:

from datetime import timedelta

schedule = FixedRate(duration=timedelta(minutes=10))

launch_plan = LaunchPlan.get_or_create(
    workflow=my_wf,
    name="my-wf-every-ten-minutes",
    default_inputs={"a": 7},
    fixed_inputs={"c": "production"},
    schedule=schedule,
)

Here a receives a launch-plan default and c is fixed. The schedule is optional; the same named-plan factory also accepts notifications, labels, annotations, raw_output_data_config, max_parallelism, security_context, auth_role, trigger, overwrite_cache, and auto_activate.

How defaults and fixed inputs are represented

LaunchPlan.create starts by transforming the workflow’s Python interface into a parameter map. It then constructs a temporary Interface from the explicit default_inputs and updates the workflow-derived parameters with those values. Consequently, an explicit launch-plan default takes precedence over a default declared in the workflow signature.

Fixed inputs follow a different path. translate_inputs_to_literals converts the Python values using both workflow.interface.inputs and workflow.python_interface.inputs, and the result is stored in a model LiteralMap. During LaunchPlan.__init__, names present in that literal map are removed from the ParameterMap. The source performs this filtering with parameters = {k: v for k, v in parameters.parameters.items() if k not in fixed_inputs.literals}, then stores the result in _interface_models.ParameterMap(parameters=parameters) and stores the literal map in _fixed_inputs.

The resulting object therefore exposes ordinary parameters separately from fixed values:

launch_plan.parameters
launch_plan.fixed_inputs
launch_plan.saved_inputs

create also retains the original native values in _saved_inputs. It updates the saved dictionary with fixed_inputs, so saved_inputs contains both explicit defaults and fixed values. The saved_inputs property returns a copy. This native dictionary is used when the plan is called locally or while a workflow is being compiled; it does not make fixed inputs mutable launch-time parameters. The lower-level node-binding path rejects an attempt to supply a fixed-input name with Fixed inputs cannot be specified.

A default launch plan takes its saved defaults from the workflow’s Python signature:

default_inputs = {
    name: default
    for name, (type, default) in workflow.python_interface.inputs_with_defaults.items()
}
lp._saved_inputs = default_inputs

This is why the unnamed default plan can apply defaults declared in the workflow function, whereas a named plan can add or override defaults through default_inputs.

Calling a launch plan in a workflow

Invoke a launch plan with keyword arguments only. The saved defaults and fixed values are copied, then call-site keywords are applied:

result = launch_plan(a=11)

LaunchPlan.__call__ rejects positional arguments with AssertionError("Only Keyword Arguments are supported for launch plan executions"). Its behavior then depends on the current FlyteContext:

• During compilation, it calls create_and_link_node(ctx, entity=self, **inputs), producing a node for the launch plan.
• Outside compilation, it forwards the merged values to the wrapped workflow with self.workflow(*args, **inputs).

WorkflowBase.add_launch_plan is the workflow-level integration point. Its implementation delegates to the general entity-adding path, which enters compilation and creates the corresponding node. The same node creation machinery recognizes launch plans alongside other callable Flyte entities.

A launch plan is also appended to FlyteEntities.entities in its constructor. That collection is used by registration and serialization discovery, so constructing a plan makes it available to those flows. LaunchPlan.construct_node_metadata() delegates directly to the wrapped workflow’s construct_node_metadata().

Scheduling a launch plan

Pass a CronSchedule or FixedRate to the named plan’s schedule argument when you want the plan to carry the older/direct schedule field:

from datetime import timedelta

launch_plan = LaunchPlan.get_or_create(
    workflow=my_wf,
    name="my-wf-hourly",
    schedule=FixedRate(duration=timedelta(hours=1)),
)

CronSchedule is the native-scheduler option. Its schedule argument accepts a five-field cron expression accepted by croniter or one of the aliases defined by flytekit, including hourly, daily, weekly, monthly, annually, their plural forms, and the corresponding @... forms:

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

Do not pass the deprecated cron_expression argument. CronSchedule.__init__ rejects it immediately with an assertion and instructs callers to use schedule instead. The separate legacy expression validator in the class expects six AWS/CloudWatch-style fields and requires ? in either the day-of-month or day-of-week position; native schedule uses the five-field form. An offset, when supplied, is checked against flytekit’s ISO-8601-duration regular expression.

Supplying the scheduled kickoff time

Both schedule classes accept kickoff_time_input_arg. Set it to the name of a workflow input when the workflow needs the scheduled kickoff time:

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

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

The CronSchedule source documents this as a convenient workflow input for the run’s kickoff time and cautions that the actual start can differ from the nominal schedule by a few seconds. The value is therefore not described as an atomic clock reading by flytekit.

Fixed-rate conversion and granularity

FixedRate accepts a datetime.timedelta and converts it into the model’s fixed-rate representation. The conversion chooses days first, then hours, and otherwise minutes. For example, flytekit’s source documents:

from datetime import timedelta

FixedRate(duration=timedelta(minutes=10))

The implementation rejects durations with microseconds or with leftover seconds below whole-minute granularity. Use an exact whole-minute interval. A duration that is an exact multiple of a day becomes a day-based model value; an exact multiple of an hour becomes an hour-based value; all other supported intervals become minute-based values.

The trigger adapter

LaunchPlan accepts trigger in addition to the direct schedule argument. LaunchPlanTriggerBase is a protocol requiring to_flyte_idl, and OnSchedule implements that protocol by wrapping a CronSchedule or FixedRate and delegating to the wrapped object:

on_schedule = OnSchedule(
    CronSchedule(schedule="*/1 * * * *")
)

launch_plan = LaunchPlan.get_or_create(
    workflow=my_wf,
    name="my-wf-triggered",
    trigger=on_schedule,
)

The trigger form is marked [alpha] in the get_or_create documentation. OnSchedule.to_flyte_idl() returns the schedule’s protobuf representation; it does not add another scheduling algorithm around the wrapped schedule.

Naming, caching, and reuse

An omitted name means “the default launch plan for this workflow.” get_or_create raises ValueError if an unnamed call also supplies any launch-plan-specific property, including default_inputs, fixed_inputs, schedule, notifications, metadata, security settings, max_parallelism, trigger, or overwrite_cache. Name the plan when you need any of those settings.

Named plans are cached in the process-global LaunchPlan.CACHE. Reusing a name returns the cached object only when the workflow and relevant configuration match. A different workflow raises an assertion, and differences in schedule, notifications, saved inputs, labels, annotations, output configuration, parallelism, security context, cache behavior, or activation behavior also raise an assertion. LaunchPlan.create independently rejects a duplicate name.

Two implementation details matter when reusing plans:

• In the cached-name path, get_or_create normalizes dictionaries and calls default_inputs.update(fixed_inputs). The caller’s dictionaries can therefore be mutated; do not rely on them remaining untouched.
• clone_with uses truthiness-based fallbacks such as schedule or self.schedule and max_parallelism or self.max_parallelism. It cannot intentionally replace an existing value with a falsey value through that helper. Its trigger argument is passed directly rather than falling back to the original trigger.

Integration with dynamic and mapped execution

A dynamically invoked sub-launch plan must already be registered on Flyte Admin. The task API documents node_dependency_hints for this ordering relationship:

@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

array_node also accepts local or remote launch plans as targets. For a local launch plan, it records the names in target.fixed_inputs.literals as excluded inputs, so those values are not treated as mapped inputs. For launch-plan targets, the array node is configured with SINGLE_INPUT_FILE data mode and FULL_STATE execution mode.

Referencing an existing remote plan

Use ReferenceLaunchPlan when the launch plan is already registered and you need a typed pointer to it rather than a locally constructed plan. Its constructor requires the remote project, domain, name, and version plus explicit input and output type mappings:

ref_launch_plan = ReferenceLaunchPlan(
    project="project",
    domain="dev",
    name="my.launch.plan",
    version="abc123",
    inputs={"a": str},
    outputs={},
)

ReferenceLaunchPlan inherits ReferenceEntity. It does not contact Admin to discover the remote interface, so the declared interface is the contract used during compilation. The reference_launch_plan(project, domain, name, version) decorator is the convenience form that derives that interface from an annotated function; reference.get_reference_entity is the programmatic alternative for a ResourceType.LAUNCH_PLAN reference. Reference entities cannot execute locally: their shared execute method raises NotImplementedError, so local execution requires mocking the remote entity.

Common edge cases

Situation	flytekit behavior
You pass launch-plan configuration without name	get_or_create raises ValueError; unnamed plans are reserved for the default plan.
You call a launch plan positionally	LaunchPlan.__call__ raises AssertionError; use keyword arguments.
You supply a fixed input at launch time	Fixed names are removed from the parameter map, and node binding rejects them with Fixed inputs cannot be specified.
You pass cron_expression to CronSchedule	The constructor rejects the deprecated argument; use five-field native schedule.
Your FixedRate has seconds or microseconds below a minute	_translate_duration raises AssertionError; use an exact whole-minute duration.
You need the scheduled kickoff time	Set kickoff_time_input_arg, but account for the documented few-second timing difference.
You reuse a cached name with different configuration	get_or_create raises an assertion instead of creating a second plan.
You expect clone_with to clear a falsey setting	Truthiness-based fallbacks preserve the old value.
You rely on this source snapshot for tests or Markdown examples	The searched snapshot contains no test_*.py, *_test.py, or Markdown documentation files; the strongest inline examples are the source docstrings and API comments.