Chapter 5. Using the AI pipelines REST API with Kubeflow pipelines SDK

versions = client.list_pipeline_versions(<_pipeline_id_>)

for version in versions.pipeline_versions:
    print(f"Version: {version.display_name} (ID: {version.pipeline_version_id})")

version = client.get_pipeline_version(
    pipeline_id=<_pipeline_id_>,
    pipeline_version_id=<_pipeline_version_id_>
)

experiment = client.create_experiment(
    name="my-experiment",
    description="Experiment for testing pipelines",
)

# List all experiments
experiments = client.list_experiments()

for exp in experiments.experiments:
    print(f"Experiment: {exp.display_name} (ID: {exp.experiment_id})")

# List with pagination and sorting
experiments = client.list_experiments(
    page_size=20,
    sort_by='created_at desc'
)

export PIPELINE_HOST=<your_pipeline_server_url>

#!/usr/bin/env python3
"""
Complete example of using AI Pipelines REST API
"""

import kfp
from kfp import Client
import time
import logging
import os

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def main():
    # Initialize client
    pipeline_host = os.getenv('PIPELINE_HOST')
    if not pipeline_host:
        raise ValueError("PIPELINE_HOST environment variable not set")
    client = Client(host=pipeline_host)

    try:
        # 1. Create or get experiment
        experiment_name = 'rest-api-example'
        try:
            experiment = client.get_experiment(experiment_name=experiment_name)
            logger.info(f"Using existing experiment: {experiment.experiment_id}")
        except:
            experiment = client.create_experiment(
                name=experiment_name,
                description='Example experiment using REST API'
            )
            logger.info(f"Created new experiment: {experiment.experiment_id}")

        # 2. Upload pipeline
        # Replace '<path_to_pipeline_yaml>' with your compiled pipeline file path
        pipeline = client.upload_pipeline(
            pipeline_package_path='<path_to_pipeline_yaml>',
            pipeline_name='example-pipeline',
            description='Example pipeline for REST API demo'
        )
        logger.info(f"Uploaded pipeline: {pipeline.pipeline_id}")

        # 3. Submit run
        run = client.run_pipeline(
            experiment_id=experiment.experiment_id,
            job_name=f'example-run-{int(time.time())}',
            pipeline_id=pipeline.pipeline_id,
            params={
                'input_data': 'gs://your-bucket/data.csv',
                'model_name': 'example-model',
                'epochs': 10
            }
        )
        logger.info(f"Submitted run: {run.run_id}")

        # 4. Monitor run
        logger.info("Monitoring run progress...")
        client.wait_for_run_completion(run.run_id, timeout=3600)

        # 5. Get final results
        run_detail = client.get_run(run.run_id)
        final_status = run_detail.state
        logger.info(f"Run completed with status: {final_status}")

        if final_status.lower() == 'succeeded':
            logger.info("Pipeline executed successfully!")
        else:
            logger.error(f"Pipeline failed with status: {final_status}")

    except Exception as e:
        logger.error(f"Error in pipeline execution: {str(e)}")
        raise

if __name__ == '__main__':
    main()

def get_all_pipelines(client):
    """Efficiently retrieve all pipelines using pagination."""
    all_pipelines = []
    page_token = None

    while True:
        response = client.list_pipelines(
            page_size=100,
            page_token=page_token
        )

        all_pipelines.extend(response.pipelines)

        if not response.next_page_token:
            break

        page_token = response.next_page_token

    return all_pipelines