kfp.client

The kfp.client module contains the KFP API client.

class kfp.client.Client(host: Optional[str] = None, client_id: Optional[str] = None, namespace: str = 'kubeflow', other_client_id: Optional[str] = None, other_client_secret: Optional[str] = None, existing_token: Optional[str] = None, cookies: Optional[str] = None, proxy: Optional[str] = None, ssl_ca_cert: Optional[str] = None, kube_context: Optional[str] = None, credentials: Optional[str] = None, ui_host: Optional[str] = None, verify_ssl: Optional[bool] = None)[source]

Bases: object

The KFP SDK client for the Kubeflow Pipelines backend API.

Parameters

host – Host name to use to talk to Kubeflow Pipelines. If not set, the in-cluster service DNS name will be used, which only works if the current environment is a pod in the same cluster (such as a Jupyter instance spawned by Kubeflow’s JupyterHub). (More information on connecting.)
client_id – Client ID used by Identity-Aware Proxy.
namespace – Kubernetes namespace to use. Used for multi-user deployments. For single-user deployments, this should be left as None.
other_client_id – Client ID used to obtain the auth codes and refresh tokens (reference).
other_client_secret – Client secret used to obtain the auth codes and refresh tokens.
existing_token – Authentication token to pass in directly. Used in cases where the token is generated from outside the SDK.
cookies – CookieJar object containing cookies that will be passed to the Pipelines API.
proxy – HTTP or HTTPS proxy server.
ssl_ca_cert – Certification for proxy.
kube_context – kubectl context to use. Must be a context listed in the kubeconfig file. Defaults to the current-context set within kubeconfig.
credentials – TokenCredentialsBase object which provides the logic to populate the requests with credentials to authenticate against the API server.
ui_host – Base URL to use to open the Kubeflow Pipelines UI. This is used when running the client from a notebook to generate and print links.
verify_ssl – Whether to verify the server’s TLS certificate.

archive_experiment(experiment_id: str) → dict[source]

Archives an experiment.

Parameters: experiment_id – ID of the experiment.
Returns: Empty dictionary.

archive_run(run_id: str) → dict[source]

Archives a run.

Parameters: run_id – ID of the run.
Returns: Empty dictionary.

create_experiment(name: str, description: Optional[str] = None, namespace: Optional[str] = None) → ApiExperiment[source]

Creates a new experiment.

Parameters

name – Name of the experiment.
description – Description of the experiment.
namespace – Kubernetes namespace to use. Used for multi-user deployments. For single-user deployments, this should be left as None.

Returns

ApiExperiment object.

create_recurring_run(experiment_id: str, job_name: str, description: Optional[str] = None, start_time: Optional[str] = None, end_time: Optional[str] = None, interval_second: Optional[int] = None, cron_expression: Optional[str] = None, max_concurrency: Optional[int] = 1, no_catchup: Optional[bool] = None, params: Optional[dict] = None, pipeline_package_path: Optional[str] = None, pipeline_id: Optional[str] = None, version_id: Optional[str] = None, enabled: bool = True, pipeline_root: Optional[str] = None, enable_caching: Optional[bool] = None, service_account: Optional[str] = None) → ApiJob[source]

Creates a recurring run.

Parameters

experiment_id – ID of the experiment.
job_name – Name of the job.
description – Description of the job.
start_time – RFC3339 time string of the time when to start the job.
end_time – RFC3339 time string of the time when to end the job.
interval_second – Integer indicating the seconds between two recurring runs in for a periodic schedule.
cron_expression – Cron expression representing a set of times, using 6 space-separated fields (e.g., '0 0 9 ? * 2-6'). See cron format.
max_concurrency – Integer indicating how many jobs can be run in parallel.
no_catchup – Whether the recurring run should catch up if behind schedule. For example, if the recurring run is paused for a while and re-enabled afterwards. If no_catchup=False, the scheduler will catch up on (backfill) each missed interval. Otherwise, it only schedules the latest interval if more than one interval is ready to be scheduled. Usually, if your pipeline handles backfill internally, you should turn catchup off to avoid duplicate backfill.
pipeline_package_path – Local path of the pipeline package (the filename should end with one of the following .tar.gz, .tgz, .zip, .json).
params – Arguments to the pipeline function provided as a dict.
pipeline_id – ID of a pipeline.
version_id – ID of a pipeline version. If both pipeline_id and version_id are specified, version_id will take precedence. If only pipeline_id is specified, the default version of this pipeline is used to create the run.
enabled – Whether to enable or disable the recurring run.
pipeline_root – Root path of the pipeline outputs.
enable_caching – Whether or not to enable caching for the run. If not set, defaults to the compile time settings, which is True for all tasks by default, while users may specify different caching options for individual tasks. If set, the setting applies to all tasks in the pipeline (overrides the compile time settings).
service_account – Specifies which Kubernetes service account this recurring run uses.

Returns

ApiJob object.

create_run_from_pipeline_func(pipeline_func: Union[Callable[[...], Any], BaseComponent], arguments: Optional[Dict[str, Any]] = None, run_name: Optional[str] = None, experiment_name: Optional[str] = None, namespace: Optional[str] = None, pipeline_root: Optional[str] = None, enable_caching: Optional[bool] = None, service_account: Optional[str] = None) → RunPipelineResult[source]

Runs pipeline on KFP-enabled Kubernetes cluster.

This command compiles the pipeline function, creates or gets an experiment, then submits the pipeline for execution.

Parameters

pipeline_func – Pipeline function constructed with @kfp.dsl.pipeline decorator.
arguments – Arguments to the pipeline function provided as a dict.
run_name – Name of the run to be shown in the UI.
experiment_name – Name of the experiment to add the run to.
namespace – Kubernetes namespace to use. Used for multi-user deployments. For single-user deployments, this should be left as None.
pipeline_root – Root path of the pipeline outputs.
enable_caching – Whether or not to enable caching for the run. If not set, defaults to the compile time settings, which is True for all tasks by default, while users may specify different caching options for individual tasks. If set, the setting applies to all tasks in the pipeline (overrides the compile time settings).
service_account – Specifies which Kubernetes service account to use for this run.

Returns

RunPipelineResult object containing information about the pipeline run.

create_run_from_pipeline_package(pipeline_file: str, arguments: Optional[Dict[str, str]] = None, run_name: Optional[str] = None, experiment_name: Optional[str] = None, namespace: Optional[str] = None, pipeline_root: Optional[str] = None, enable_caching: Optional[bool] = None, service_account: Optional[str] = None) → RunPipelineResult[source]

Runs pipeline on KFP-enabled Kubernetes cluster.

This command takes a local pipeline package, creates or gets an experiment, then submits the pipeline for execution.

Parameters

pipeline_file – A compiled pipeline package file.
arguments – Arguments to the pipeline function provided as a dict.
run_name – Name of the run to be shown in the UI.
experiment_name – Name of the experiment to add the run to.
namespace – Kubernetes namespace to use. Used for multi-user deployments. For single-user deployments, this should be left as None.
pipeline_root – Root path of the pipeline outputs.
enable_caching – Whether or not to enable caching for the run. If not set, defaults to the compile time settings, which is True for all tasks by default, while users may specify different caching options for individual tasks. If set, the setting applies to all tasks in the pipeline (overrides the compile time settings).
service_account – Specifies which Kubernetes service account to use for this run.

Returns

RunPipelineResult object containing information about the pipeline run.

delete_experiment(experiment_id: str) → dict[source]

Delete experiment.

Parameters: experiment_id – ID of the experiment.
Returns: Empty dictionary.

delete_job(job_id: str) → dict[source]

Deletes a job (recurring run).

Parameters: job_id – ID of the job.
Returns: Empty dictionary.

delete_pipeline(pipeline_id: str) → dict[source]

Deletes a pipeline.

Parameters: pipeline_id – ID of the pipeline.
Returns: Empty dictionary.

delete_pipeline_version(version_id: str) → dict[source]

Deletes a pipeline version.

Parameters: version_id – ID of the pipeline version.
Returns: Empty dictionary.

delete_run(run_id: str) → dict[source]

Deletes a run.

Parameters: run_id – ID of the run.
Returns: Empty dictionary.

disable_job(job_id: str) → dict[source]

Disables a job (recurring run).

Parameters: job_id – ID of the job.
Returns: Empty dictionary.

enable_job(job_id: str) → dict[source]

Enables a job (recurring run).

Parameters: job_id – ID of the job.
Returns: Empty dictionary.

get_experiment(experiment_id=None, experiment_name=None, namespace=None) → ApiExperiment[source]

Gets details of an experiment.

Either experiment_id or experiment_name is required.

Parameters

experiment_id – ID of the experiment.
experiment_name – Name of the experiment.
namespace – Kubernetes namespace to use. Used for multi-user deployments. For single-user deployments, this should be left as None.

Returns

ApiExperiment object.

get_kfp_healthz() → ApiGetHealthzResponse[source]

Gets healthz info for KFP deployment.

Returns: JSON response from the healtz endpoint.

get_pipeline(pipeline_id: str) → ApiPipeline[source]

Gets pipeline details.

Parameters: pipeline_id – ID of the pipeline.
Returns: ApiPipeline object.

get_pipeline_id(name: str) → Optional[str][source]

Gets the ID of a pipeline by its name.

Parameters: name – Pipeline name.
Returns: The pipeline ID if a pipeline with the name exists.

get_pipeline_version(version_id: str) → ApiPipelineVersion[source]

Gets a pipeline version.

Parameters: version_id – ID of the pipeline version.
Returns: ApiPipelineVersion object.

get_recurring_run(job_id: str) → ApiJob[source]

Gets recurring run (job) details.

Parameters: job_id – ID of the recurring run (job).
Returns: ApiJob object.

get_run(run_id: str) → ApiRun[source]

Gets run details.

Parameters: run_id – ID of the run.
Returns: ApiRun object.

get_user_namespace() → str[source]

Gets user namespace in context config.

Returns: Kubernetes namespace from the local context file or empty if it wasn’t set.

list_experiments(page_token: str = '', page_size: int = 10, sort_by: str = '', namespace: Optional[str] = None, filter: Optional[str] = None) → ApiListExperimentsResponse[source]

Lists experiments.

Parameters

page_token – Page token for obtaining page from paginated response.
page_size – Size of the page.
sort_by – Sort string of format '[field_name]', '[field_name] desc'. For example, 'name desc'.
namespace – Kubernetes namespace to use. Used for multi-user deployments. For single-user deployments, this should be left as None.

filter –

A url-encoded, JSON-serialized Filter protocol buffer (see filter.proto message). For a list of all filter operations 'op', see here. Example:

json.dumps({
    "predicates": [{
        "op": _FILTER_OPERATIONS["EQUALS"],
        "key": "name",
        "stringValue": "my-name",
    }]
})

Returns

ApiListExperimentsResponse object.

list_pipeline_versions(pipeline_id: str, page_token: str = '', page_size: int = 10, sort_by: str = '', filter: Optional[str] = None) → ApiListPipelineVersionsResponse[source]

Lists pipeline versions.

Parameters

pipeline_id – ID of the pipeline for which to list versions.
page_token – Page token for obtaining page from paginated response.
page_size – Size of the page.
sort_by – Sort string of format '[field_name]', '[field_name] desc'. For example, 'name desc'.

filter –

A url-encoded, JSON-serialized Filter protocol buffer (see filter.proto message). For a list of all filter operations 'op', see here. Example:

json.dumps({
    "predicates": [{
        "op": _FILTER_OPERATIONS["EQUALS"],
        "key": "name",
        "stringValue": "my-name",
    }]
})

Returns

ApiListPipelineVersionsResponse object.

list_pipelines(page_token: str = '', page_size: int = 10, sort_by: str = '', filter: Optional[str] = None) → ApiListPipelinesResponse[source]

Lists pipelines.

Parameters

page_token – Page token for obtaining page from paginated response.
page_size – Size of the page.
sort_by – Sort string of format '[field_name]', '[field_name] desc'. For example, 'name desc'.

filter –

A url-encoded, JSON-serialized Filter protocol buffer (see filter.proto message). For a list of all filter operations 'op', see here. Example:

json.dumps({
    "predicates": [{
        "op": _FILTER_OPERATIONS["EQUALS"],
        "key": "name",
        "stringValue": "my-name",
    }]
})

Returns

ApiListPipelinesResponse object.

list_recurring_runs(page_token: str = '', page_size: int = 10, sort_by: str = '', experiment_id: Optional[str] = None, filter: Optional[str] = None) → ApiListJobsResponse[source]

Lists recurring runs.

Parameters

page_token – Page token for obtaining page from paginated response.
page_size – Size of the page.
sort_by – Sort string of format '[field_name]', '[field_name] desc'. For example, 'name desc'.
experiment_id – Experiment ID to filter upon.

filter –

A url-encoded, JSON-serialized Filter protocol buffer (see filter.proto message). For a list of all filter operations 'op', see here. Example:

json.dumps({
    "predicates": [{
        "op": _FILTER_OPERATIONS["EQUALS"],
        "key": "name",
        "stringValue": "my-name",
    }]
})

Returns

ApiListJobsResponse object.

list_runs(page_token: str = '', page_size: int = 10, sort_by: str = '', experiment_id: Optional[str] = None, namespace: Optional[str] = None, filter: Optional[str] = None) → ApiListRunsResponse[source]

List runs.

Parameters

page_token – Page token for obtaining page from paginated response.
page_size – Size of the page.
sort_by – Sort string of format '[field_name]', '[field_name] desc'. For example, 'name desc'.
experiment_id – Experiment ID to filter upon
namespace – Kubernetes namespace to use. Used for multi-user deployments. For single-user deployments, this should be left as None.

filter –

A url-encoded, JSON-serialized Filter protocol buffer

(see filter.proto message). For a list of all filter operations 'op', see here. Example:

json.dumps({
    "predicates": [{
        "op": _FILTER_OPERATIONS["EQUALS"],
        "key": "name",
        "stringValue": "my-name",
    }]
})

Returns:

ApiListRunsResponse object.

run_pipeline(experiment_id: str, job_name: str, pipeline_package_path: Optional[str] = None, params: Optional[Dict[str, Any]] = None, pipeline_id: Optional[str] = None, version_id: Optional[str] = None, pipeline_root: Optional[str] = None, enable_caching: Optional[bool] = None, service_account: Optional[str] = None) → ApiRun[source]

Runs a specified pipeline.

Parameters

experiment_id – ID of an experiment.
job_name – Name of the job.
pipeline_package_path – Local path of the pipeline package (the filename should end with one of the following .tar.gz, .tgz, .zip, .json).
params – Arguments to the pipeline function provided as a dict.
pipeline_id – ID of the pipeline.
version_id – ID of the pipeline version to run. If both pipeline_id and version_id are specified, version_id will take precendence. If only pipeline_id is specified, the default version of this pipeline is used to create the run.
pipeline_root – Root path of the pipeline outputs.
enable_caching – Whether or not to enable caching for the run. If not set, defaults to the compile-time settings, which is True for all tasks by default. If set, the setting applies to all tasks in the pipeline (overrides the compile time settings).
service_account – Specifies which Kubernetes service account to use for this run.

Returns

ApiRun object.

set_user_namespace(namespace: str) → None[source]

Sets the namespace in the Kuberenetes cluster to use.

This function should only be used when Kubeflow Pipelines is in the multi-user mode.

Parameters: namespace – Namespace to use within the Kubernetes cluster (namespace containing the Kubeflow Pipelines deployment).

terminate_run(run_id: str) → dict[source]

Terminates a run.

Parameters: run_id – ID of the run.
Returns: Empty dictionary.

unarchive_experiment(experiment_id: str) → dict[source]

Unarchives an experiment.

Parameters: experiment_id – ID of the experiment.
Returns: Empty dictionary.

unarchive_run(run_id: str) → dict[source]

Restores an archived run.

Parameters: run_id – ID of the run.
Returns: Empty dictionary.

upload_pipeline(pipeline_package_path: Optional[str] = None, pipeline_name: Optional[str] = None, description: Optional[str] = None) → ApiPipeline[source]

Uploads a pipeline.

Parameters

pipeline_package_path – Local path to the pipeline package.
pipeline_name – Name of the pipeline to be shown in the UI.
description – Description of the pipeline to be shown in the UI.

Returns

ApiPipeline object.

upload_pipeline_version(pipeline_package_path: str, pipeline_version_name: str, pipeline_id: Optional[str] = None, pipeline_name: Optional[str] = None, description: Optional[str] = None) → ApiPipelineVersion[source]

Uploads a new version of the pipeline.

Parameters

pipeline_package_path – Local path to the pipeline package.
pipeline_version_name – Name of the pipeline version to be shown in the UI.
pipeline_id – ID of the pipeline.
pipeline_name – Name of the pipeline.
description – Description of the pipeline version to show in the UI.

Returns

ApiPipelineVersion object.

wait_for_run_completion(run_id: str, timeout: int) → ApiRun[source]

Waits for a run to complete.

Parameters

run_id – ID of the run.
timeout – Timeout after which the client should stop waiting for run completion (seconds).

Returns

ApiRun object.