I want to create the default launch plan from flytekit.models.admin.workflow.WorkflowSpec. is there any example about this? I have searched and just found some example in creating Default Launchplan from (flytekit.core.python_fucnction_task.PythonFuctionTask, flytekit.core.workflow.PythonFunctionWorkflow), But didn't found any case with flytekit.models.admin.workflow.WorkflowSpec.

did you try

LaunchPlan.get_or_create

@Kevin Su I'm trying to write my own script, Seems like i have to convert flytekit.models.admin.workflow.Workflow to flytekit.core.workflow.WorkflowBase. The interface can be get from flytekit.models.admin.workflow.Workflow too because this is the compiled workflow(the workflow get from flytekit.client.friencly).

flytekit remote should be updated to support workflowSpec

⭐ Create a new Flyte Core Feature issue: https://github.com/flyteorg/flyte/issues/new?assignees=&labels=enhancement%2Cuntriaged&template=feature_request.yaml&title=%5BCore+feature%5D+

@Kevin Su For Sure, I will.

tyty

@Kevin Su I have found that Flytekit.client can create the LaunchPLan simplely.

import json
from flytekit.configuration import PlatformConfig
from flytekit.clients import raw, friendly
from typing import Optional
from flyteidl.admin.common_pb2 import ObjectGetRequest
from flytekit.models.core.identifier import Identifier
from flytekit.models.core.identifier import ResourceType
from flytekit.models.core.identifier import Identifier, ResourceType
# from flytekit.core.launch_plan import LaunchPlan

from flyteidl.admin.launch_plan_pb2 import LaunchPlanCreateRequest, LaunchPlanSpe
from flyteidl.admin.workflow_pb2 import Workflow as _workflow
from flyteidl.core import identifier_pb2 as _identifier_pb2
import flyteidl.admin.launch_plan_pb2 as _launch_plan_pb2

# Set Up Client
sandbox_cfg = PlatformConfig.for_endpoint(endpoint='localhost:30080', insecure=True)
raw_client = raw.RawSynchronousFlyteClient(cfg=sandbox_cfg, insecure=True)
friendly_client =  friendly.SynchronousFlyteClient(cfg=sandbox_cfg)

# Get WorkflowSpec from Client
def get_workflow(client=raw_client,
                 project:str='flytesnacks',
                 domain:str='development',
                 name: str='pb2_wf', 
                 version: Optional[str]=None) -> _workflow:
    obj_req = ObjectGetRequest(
        id=Identifier(
            resource_type=ResourceType.WORKFLOW,
            project=project,
            domain=domain,
            name=name,
            version=version
        ).to_flyte_idl()
    )
    _wf = client.get_workflow(get_object_request=obj_req)
    return _wf

# Create default LaunchPlan with WorkflowSpec
def create_default_launch_plan(wf: _workflow,
                        client=raw_client,
                        lp_name: str='name_lp',
                        lp_version: str='lp_default_name'):
    launch_plan = _launch_plan_pb2.LaunchPlan(
        id=_identifier_pb2.Identifier(
            resource_type=_identifier_pb2.LAUNCH_PLAN,
            project=wf.id.project,
            domain=wf.id.domain,
            name=lp_name,
            version=lp_version,
        ),
        spec=_launch_plan_pb2.LaunchPlanSpec(
            workflow_id=_identifier_pb2.Identifier(
                resource_type=_identifier_pb2.WORKFLOW,
                project=wf.id.project,
                domain=wf.id.domain,
                name=wf.id.name,
                version=wf.id.version,
            )
        ),
    )
    lp_create_request = LaunchPlanCreateRequest(
        id = launch_plan.id,
        spec = launch_plan.spec
    )
    try:
        client.create_launch_plan(launch_plan_create_request=lp_create_request)
    except Exception as e:
        print('a')
        print(e)
        
def create_schedule_launch_plan():
    pass

if __name__ == '__main__':
    wf = get_workflow(name = 'aaa', version='v3')
    create_default_launch_plan(wf, lp_name=wf.id.name, lp_version=wf.id.version)

nice, could you help us update the flytesnacks example