Repository: getindata/kedro-kubeflow
Pull request #101 (2022-02-04 08:33:51)

Add pipeline name when uploading
#### Description

Pipeline name was missing in Kubeflow interface, so we couldn't differentiate two pipelines from one project. It's fixed by using the template `[kedro_project_name] pipeline_name`.

Also, KFP client is bumped to 1.8

##### PR Checklist
- [x] Tests added
- [x] [Changelog](CHANGELOG.md) updated 



Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index f528e89..80f8f83 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -2,6 +2,8 @@
 
 ## [Unreleased]
 
+-   Kedro pipeline name is now added into Kubeflow pipeline name during upload
+
 ## [0.5.1] - 2022-01-28
 
 -   Possibility to run custom Kedro pipeline as on-exit-handler
diff --git a/kedro_kubeflow/cli.py b/kedro_kubeflow/cli.py
index 1c0d961..f9a9c9b 100644
--- a/kedro_kubeflow/cli.py
+++ b/kedro_kubeflow/cli.py
@@ -165,7 +165,7 @@ def upload_pipeline(ctx, image, pipeline) -> None:
     config = context_helper.config.run_config
 
     context_helper.kfp_client.upload(
-        pipeline=pipeline,
+        pipeline_name=pipeline,
         image=image if image else config.image,
         image_pull_policy=config.image_pull_policy,
     )
diff --git a/kedro_kubeflow/kfpclient.py b/kedro_kubeflow/kfpclient.py
index d6f5881..fd55262 100644
--- a/kedro_kubeflow/kfpclient.py
+++ b/kedro_kubeflow/kfpclient.py
@@ -1,4 +1,3 @@
-import json
 import logging
 import uuid
 from tempfile import NamedTemporaryFile
@@ -95,17 +94,23 @@ class KubeflowClient(object):
         )
         self.log.info("Generated pipeline definition was saved to %s" % output)
 
-    def upload(self, pipeline, image, image_pull_policy="IfNotPresent"):
+    def get_full_pipeline_name(self, pipeline_name):
+        return f"[{self.project_name}] {pipeline_name}"
+
+    def upload(self, pipeline_name, image, image_pull_policy="IfNotPresent"):
         pipeline = self.generator.generate_pipeline(
-            pipeline, image, image_pull_policy
+            pipeline_name, image, image_pull_policy
         )
 
-        if self._pipeline_exists(self.project_name):
-            pipeline_id = self._get_pipeline_id(self.project_name)
+        full_pipeline_name = self.get_full_pipeline_name(pipeline_name)
+        if self._pipeline_exists(full_pipeline_name):
+            pipeline_id = self.client.get_pipeline_id(full_pipeline_name)
             version_id = self._upload_pipeline_version(pipeline, pipeline_id)
             self.log.info("New version of pipeline created: %s", version_id)
         else:
-            (pipeline_id, version_id) = self._upload_pipeline(pipeline)
+            (pipeline_id, version_id) = self._upload_pipeline(
+                pipeline, full_pipeline_name
+            )
             self.log.info("Pipeline created")
 
         self.log.info(
@@ -115,25 +120,7 @@ class KubeflowClient(object):
         )
 
     def _pipeline_exists(self, pipeline_name):
-        return self._get_pipeline_id(pipeline_name) is not None
-
-    def _get_pipeline_id(self, pipeline_name):
-        pipelines = self.client.pipelines.list_pipelines(
-            filter=json.dumps(
-                {
-                    "predicates": [
-                        {
-                            "key": "name",
-                            "op": 1,
-                            "string_value": pipeline_name,
-                        }
-                    ]
-                }
-            )
-        ).pipelines
-
-        if pipelines:
-            return pipelines[0].id
+        return self.client.get_pipeline_id(pipeline_name) is not None
 
     def _upload_pipeline_version(self, pipeline_func, pipeline_id):
         version_name = f"{clean_name(self.project_name)}-{uuid.uuid4()}"[:100]
@@ -146,12 +133,12 @@ class KubeflowClient(object):
                 _request_timeout=10000,
             ).id
 
-    def _upload_pipeline(self, pipeline_func):
+    def _upload_pipeline(self, pipeline_func, pipeline_name):
         with NamedTemporaryFile(suffix=".yaml") as f:
             Compiler().compile(pipeline_func, f.name)
             pipeline = self.client.pipeline_uploads.upload_pipeline(
                 f.name,
-                name=self.project_name,
+                name=pipeline_name,
                 description=self.pipeline_description,
                 _request_timeout=10000,
             )
@@ -187,7 +174,9 @@ class KubeflowClient(object):
         experiment_id = self._ensure_experiment_exists(
             experiment_name, experiment_namespace
         )
-        pipeline_id = self._get_pipeline_id(self.project_name)
+        pipeline_id = self.client.get_pipeline_id(
+            self.get_full_pipeline_name(pipeline)
+        )
         formatted_run_name = run_name.format(**parameters)
         self._disable_runs(experiment_id, formatted_run_name)
         self.client.create_recurring_run(
diff --git a/setup.py b/setup.py
index 8fe336a..4c46333 100644
--- a/setup.py
+++ b/setup.py
@@ -8,7 +8,7 @@ with open("README.md") as f:
 INSTALL_REQUIRES = [
     "kedro>=0.16,<=0.18",
     "click<8.0",
-    "kfp~=1.6.0",
+    "kfp~=1.8.0",
     "tabulate>=0.8.7",
     "semver~=2.10",
     "google-auth<2.0dev",
@@ -48,7 +48,7 @@ setup(
         "Programming Language :: Python :: 3.8",
     ],
     keywords="kedro kubeflow plugin",
-    author=u"Mateusz Pytel, Mariusz Strzelecki",
+    author="Mateusz Pytel, Mariusz Strzelecki",
     author_email="mateusz@getindata.com",
     url="https://github.com/getindata/kedro-kubeflow/",
     packages=find_packages(exclude=["ez_setup", "examples", "tests", "docs"]),


Test patch:
diff --git a/tests/test_cli.py b/tests/test_cli.py
index 32ca2b9..76782e0 100644
--- a/tests/test_cli.py
+++ b/tests/test_cli.py
@@ -127,7 +127,7 @@ class TestPluginCLI(unittest.TestCase):
 
         assert result.exit_code == 0
         context_helper.kfp_client.upload.assert_called_with(
-            image="img", image_pull_policy="Always", pipeline="pipe"
+            image="img", image_pull_policy="Always", pipeline_name="pipe"
         )
 
     def test_schedule(self):
diff --git a/tests/test_kfpclient.py b/tests/test_kfpclient.py
index 01948a3..50eb087 100644
--- a/tests/test_kfpclient.py
+++ b/tests/test_kfpclient.py
@@ -221,10 +221,7 @@ class TestKubeflowClient(unittest.TestCase):
         self.kfp_client_mock.get_experiment.return_value = (
             self.create_experiment()
         )
-        self.kfp_client_mock.pipelines = unittest.mock.MagicMock()
-        self.kfp_client_mock.pipelines.list_pipelines.return_value = (
-            self.create_pipelines_list()
-        )
+        self.kfp_client_mock.get_pipeline_id.return_value = "someid"
 
         # when
         self.client_under_test.schedule(
@@ -254,10 +251,7 @@ class TestKubeflowClient(unittest.TestCase):
         self.kfp_client_mock.create_experiment.return_value = (
             self.create_experiment()
         )
-        self.kfp_client_mock.pipelines = unittest.mock.MagicMock()
-        self.kfp_client_mock.pipelines.list_pipelines.return_value = (
-            self.create_pipelines_list()
-        )
+        self.kfp_client_mock.get_pipeline_id.return_value = "someid"
 
         # when
         self.client_under_test.schedule(
@@ -284,10 +278,7 @@ class TestKubeflowClient(unittest.TestCase):
         self.kfp_client_mock.get_experiment.return_value = (
             self.create_experiment()
         )
-        self.kfp_client_mock.pipelines = unittest.mock.MagicMock()
-        self.kfp_client_mock.pipelines.list_pipelines.return_value = (
-            self.create_pipelines_list()
-        )
+        self.kfp_client_mock.get_pipeline_id.return_value = "someid"
         self.kfp_client_mock.list_recurring_runs.return_value = (
             self.create_recurring_jobs_list("scheduled run for region ABC")
         )
@@ -317,14 +308,11 @@ class TestKubeflowClient(unittest.TestCase):
     def test_should_upload_new_pipeline(self):
         # given
         self.create_client({"description": "Very Important Pipeline"})
-        self.kfp_client_mock.pipelines = unittest.mock.MagicMock()
-        self.kfp_client_mock.pipelines.list_pipelines.return_value = (
-            self.create_empty_pipelines_list()
-        )
+        self.kfp_client_mock.get_pipeline_id.return_value = None
 
         # when
         self.client_under_test.upload(
-            pipeline="pipeline",
+            pipeline_name="pipeline_name",
             image="unittest-image",
             image_pull_policy="Always",
         )
@@ -336,20 +324,16 @@ class TestKubeflowClient(unittest.TestCase):
             args,
             kwargs,
         ) = self.kfp_client_mock.pipeline_uploads.upload_pipeline.call_args
-        assert kwargs["name"] == "my-awesome-project"
+        assert kwargs["name"] == "[my-awesome-project] pipeline_name"
         assert kwargs["description"] == "Very Important Pipeline"
 
     def test_should_upload_new_version_of_existing_pipeline(self):
         # given
-        self.kfp_client_mock.pipelines = unittest.mock.MagicMock()
-        self.kfp_client_mock.pipelines = unittest.mock.MagicMock()
-        self.kfp_client_mock.pipelines.list_pipelines.return_value = (
-            self.create_pipelines_list()
-        )
+        self.kfp_client_mock.get_pipeline_id.return_value = "123"
 
         # when
         self.client_under_test.upload(
-            pipeline="pipeline",
+            pipeline_name="pipeline",
             image="unittest-image",
             image_pull_policy="Always",
         )