mammoth-parrot-74806
02/28/2025, 11:52 AMregister_script
of Flyte Remote object? I'm currently doing this:
def register(
self,
pipeline: Workflow,
version: str,
schedule_config: Optional[ScheduleConfig] = None,
labels: Optional[Dict[str, str]] = None
) -> Tuple[FlyteWorkflow, FlyteLaunchPlan]:
"""
Register a Flyte pipeline and create a launch plan for scheduling.
Args:
pipeline (Workflow): Flyte pipeline to be registered
schedule_config (Optional[ScheduleConfig]): configuration for scheduling the pipeline
labels (Optional[Dict[str, str]]): labels to be added to the pipeline
Returns:
Tuple[FlyteWorkflow, FlyteLaunchPlan]: Tuple containing the registered workflow and launch plan
"""
# Register workflow
registered_workflow = self._remote.register_script(
entity=pipeline,
version=version,
source_path='.',
default_launch_plan=False,
fast_package_options=FastPackageOptions([], copy_style='copy', show_files='show_files')
)
# Create Launch Plan
lp = LaunchPlan.get_or_create(
name=schedule_config.name,
workflow=pipeline,
schedule=schedule_config.schedule,
default_inputs=schedule_config.default_inputs
)
# Register launch plan
registered_launch_plan = self._remote.register_launch_plan(
entity=lp,
version=version
)
# Activate launch plan
flytelp = self._remote.fetch_launch_plan(name=lp.name, version=version)
self._remote.activate_launchplan(flytelp.id)
return registered_workflow, registered_launch_plan
But if I try to do this sequentially, each workflow needs to have a different version because tasks are shared (the code of them is exactly the same) 🤔average-finland-92144
02/28/2025, 4:54 PM@dynamic
inputs at execution time rather than creating separate workflows for each parameter set.mammoth-parrot-74806
02/28/2025, 5:02 PM