How to use the flyteidl.admin.workflow_pb2 function in flyteidl

n6 = my_list_task(a=n5.outputs.b)

    nodes = [n1, n2, n3, n4, n5, n6]

    wf_out = [
        workflow.Output(
            'nested_out',
            [n5.outputs.b, n6.outputs.b, [n1.outputs.b, n2.outputs.b]],
            sdk_type=[[primitives.Integer]]
        ),
        workflow.Output('scalar_out', n1.outputs.b, sdk_type=primitives.Integer)
    ]

    w = workflow.SdkWorkflow(inputs=input_list, outputs=wf_out, nodes=nodes)
    serialized = w.serialize()
    assert isinstance(serialized, _workflow_pb2.WorkflowSpec)
    assert len(serialized.template.nodes) == 6
    assert len(serialized.template.interface.inputs.variables.keys()) == 2
    assert len(serialized.template.interface.outputs.variables.keys()) == 2

def to_flyte_idl(self):
        """
        :rtype: flyteidl.admin.workflow_pb2.WorkflowClosure
        """
        return _admin_workflow.WorkflowClosure(
            compiled_workflow=self.compiled_workflow.to_flyte_idl()
        )

def to_flyte_idl(self):
        """
        :rtype: flyteidl.admin.workflow_pb2.WorkflowSpec
        """
        return _admin_workflow.WorkflowSpec(
            template=self._template.to_flyte_idl(),
            sub_workflows=[s.to_flyte_idl() for s in self._sub_workflows],
        )

the database must match the existing definition exactly. Furthermore, as long as the request
            remains identical, calling this method multiple times will result in success.

        :param: flytekit.models.core.identifier.Identifier workflow_identifier: The identifier for this workflow.
        :param: Text version: The version identifier of this workflow. Used to distinguish between different iterations
            of tasks with the same name. If any aspect of the underlying workflow definition changes, then the version
            must also change to be accepted by the Flyte Admin Service.
        :param: flytekit.models.admin.workflow.WorkflowSpec workflow_spec: This is the actual definition of the workflow
            that should be created.
        :raises flytekit.common.exceptions.user.FlyteEntityAlreadyExistsException: If an identical version of the
            workflow is found, this exception is raised.  The client might choose to ignore this exception because the
            identical workflow is already registered.
        :raises grpc.RpcError:
        """
        super(SynchronousFlyteClient, self).create_workflow(
            _workflow_pb2.WorkflowCreateRequest(
                id=workflow_identifier.to_flyte_idl(),
                spec=workflow_spec.to_flyte_idl()
            )

def to_flyte_idl(self):
        """
        :rtype: flyteidl.admin.workflow_pb2.Workflow
        """
        return _admin_workflow.Workflow(
            id=self.id.to_flyte_idl(),
            closure=self.closure.to_flyte_idl()
        )

def _extract_pair(identifier_file, object_file):
    """
    :param Text identifier_file:
    :param Text object_file:
    :rtype: (flyteidl.core.identifier_pb2.Identifier, T)
    """
    resource_map = {
        _identifier_pb2.LAUNCH_PLAN: _launch_plan_pb2.LaunchPlanSpec,
        _identifier_pb2.WORKFLOW: _workflow_pb2.WorkflowSpec,
        _identifier_pb2.TASK: _task_pb2.TaskSpec
    }
    id = _load_proto_from_file(_identifier_pb2.Identifier, identifier_file)
    if not id.resource_type in resource_map:
        raise _user_exceptions.FlyteAssertion(f"Resource type found in identifier {id.resource_type} invalid, must be launch plan, "
                              f"task, or workflow")
    entity = _load_proto_from_file(resource_map[id.resource_type], object_file)
    return id, entity

client = _friendly_client.SynchronousFlyteClient(host, insecure=insecure)
    files = list(files)
    files.sort()
    _click.secho("Parsing files...", fg='green', bold=True)
    for f in files:
        _click.echo(f"  {f}")

    flyte_entities_list = _extract_files(files)
    for id, flyte_entity in flyte_entities_list:
        try:
            if id.resource_type == _identifier_pb2.LAUNCH_PLAN:
                client.raw.create_launch_plan(_launch_plan_pb2.LaunchPlanCreateRequest(id=id, spec=flyte_entity))
            elif id.resource_type == _identifier_pb2.TASK:
                client.raw.create_task(_task_pb2.TaskCreateRequest(id=id, spec=flyte_entity))
            elif id.resource_type == _identifier_pb2.WORKFLOW:
                client.raw.create_workflow(_workflow_pb2.WorkflowCreateRequest(id=id, spec=flyte_entity))
            else:
                raise _user_exceptions.FlyteAssertion(f"Only tasks, launch plans, and workflows can be called with this function, "
                                      f"resource type {id.resource_type} was passed")
            _click.secho(f"Registered {id}", fg='green')
        except _user_exceptions.FlyteEntityAlreadyExistsException:
            _click.secho(f"Skipping because already registered {id}", fg='cyan')

    _click.echo(f"Finished scanning {len(flyte_entities_list)} files")