kfp-pipeline-spec

import kfp_pipeline_spec.pipeline_spec_pb2 as pb from google.protobuf import json_format # Create a minimal KFP PipelineSpec object directly using the protobuf definitions pipeline_spec = pb.PipelineSpec( pipeline_info=pb.PipelineInfo(name="my-direct-pipeline"), root=pb.PipelineTaskSpec( executor_label="my-executor", inputs=pb.TaskInputsSpec(), outputs=pb.TaskOutputsSpec(), ), components={ "my-executor": pb.ComponentSpec( executor_label="my-executor", input_definitions=pb.ComponentInputsSpec(), output_definitions=pb.ComponentOutputsSpec(), implementation=pb.ComponentSpec.ContainerImplementation( container=pb.ContainerSpec( image="python:3.9-slim", command=["python", "-c", "print('Hello from kfp-pipeline-spec!')"] ) ) ) } ) # Serialize the PipelineSpec to JSON (as it would be in a compiled YAML file) json_output = json_format.MessageToJson(pipeline_spec, indent=2) print(json_output) # To verify it's a valid spec object assert isinstance(pipeline_spec, pb.PipelineSpec) print(f"Pipeline name: {pipeline_spec.pipeline_info.name}")