[Projects] Fix get pipeline when not remote and default to remote (#1293

)

mlrun/__main__.py

@@ -829,7 +829,7 @@ def project(
                 engine=engine,
                 local=local,
             )
-            print(f"run id: {run_result.run_id}, state={run_result.state}")
+            print(f"run id: {run_result.run_id}")
         except Exception as exc:
             print(traceback.format_exc())
             message = f"failed to run pipeline, {exc}"

mlrun/projects/pipelines.py

@@ -303,6 +303,21 @@ def wait_for_completion(run_id, project=None, timeout=None, expected_statuses=No
     def get_state(run_id, project=None):
         return ""
 
+    @staticmethod
+    def _get_handler(workflow_handler, workflow_spec, project, secrets):
+        if not (workflow_handler and callable(workflow_handler)):
+            workflow_file = workflow_spec.get_source_file(project.spec.context)
+            workflow_handler = create_pipeline(
+                project,
+                workflow_file,
+                pipeline_context.functions,
+                secrets,
+                handler=workflow_handler or workflow_spec.handler,
+            )
+        else:
+            builtins.funcs = pipeline_context.functions
+        return workflow_handler
+
 
 class _KFPRunner(_PipelineRunner):
     """Kubeflow pipelines runner"""
@@ -334,17 +349,9 @@ def run(
         namespace=None,
     ) -> _PipelineRunStatus:
         pipeline_context.set(project, workflow_spec)
-        if not workflow_handler or not callable(workflow_handler):
-            workflow_file = workflow_spec.get_source_file(project.spec.context)
-            workflow_handler = create_pipeline(
-                project,
-                workflow_file,
-                pipeline_context.functions,
-                secrets,
-                handler=workflow_handler,
-            )
-        else:
-            builtins.funcs = pipeline_context.functions
+        workflow_handler = _PipelineRunner._get_handler(
+            workflow_handler, workflow_spec, project, secrets
+        )
 
         namespace = namespace or config.namespace
         id = run_pipeline(
@@ -402,17 +409,9 @@ def run(
         namespace=None,
     ) -> _PipelineRunStatus:
         pipeline_context.set(project, workflow_spec)
-        if not workflow_handler or not callable(workflow_handler):
-            workflow_file = workflow_spec.get_source_file(project.spec.context)
-            workflow_handler = create_pipeline(
-                project,
-                workflow_file,
-                pipeline_context.functions,
-                secrets,
-                handler=workflow_handler,
-            )
-        else:
-            builtins.funcs = pipeline_context.functions
+        workflow_handler = _PipelineRunner._get_handler(
+            workflow_handler, workflow_spec, project, secrets
+        )
 
         workflow_id = uuid.uuid4().hex
         pipeline_context.workflow_id = workflow_id

mlrun/projects/project.py

@@ -937,6 +937,7 @@ def set_workflow(
         embed=False,
         engine=None,
         args_schema: typing.List[EntrypointParam] = None,
+        handler=None,
         **args,
     ):
         """add or update a workflow, specify a name and the code path
@@ -946,6 +947,7 @@ def set_workflow(
         :param embed:         add the workflow code into the project.yaml
         :param engine:        workflow processing engine ("kfp" or "local")
         :param args_schema:   list of arg schema definitions (:py:class`~mlrun.model.EntrypointParam`)
+        :param handler:       workflow function handler
         :param args:          argument values (key=value, ..)
         """
         if not workflow_path:
@@ -960,6 +962,8 @@ def set_workflow(
             workflow = {"name": name, "path": workflow_path}
         if args:
             workflow["args"] = args
+        if handler:
+            workflow["handler"] = handler
         if args_schema:
             args_schema = [
                 schema.to_dict() if hasattr(schema, "to_dict") else schema

mlrun/run.py

@@ -38,7 +38,6 @@
 from .datastore import store_manager
 from .db import get_or_set_dburl, get_run_db
 from .execution import MLClientCtx
-from .k8s_utils import get_k8s_helper
 from .model import BaseMetadata, RunObject, RunTemplate
 from .runtimes import (
     DaskCluster,
@@ -1008,7 +1007,7 @@ def get_pipeline_if_completed(run_id, namespace=namespace):
         resp = client.wait_for_run_completion(run_id, timeout)
         if resp:
             resp = resp.to_dict()
-            resp = format_summary_from_kfp_run(resp["run"])
+            resp = format_summary_from_kfp_run(resp)
         show_kfp_run(resp)
 
     status = resp["run"]["status"] if resp else "unknown"
@@ -1037,6 +1036,7 @@ def get_pipeline(
         str, mlrun.api.schemas.PipelinesFormat
     ] = mlrun.api.schemas.PipelinesFormat.summary,
     project: str = None,
+    remote: bool = True,
 ):
     """Get Pipeline status
 
@@ -1046,11 +1046,11 @@ def get_pipeline(
             - ``summary`` (default value) - Return summary of the object data.
             - ``full`` - Return full pipeline object.
     :param project:    the project of the pipeline run
+    :param remote:     read kfp data from mlrun service (default=True)
 
     :return: kfp run dict
     """
     namespace = namespace or mlconf.namespace
-    remote = not get_k8s_helper(silent=True).is_running_inside_kubernetes_cluster()
     if remote:
         mldb = get_run_db()
         if mldb.kind != "http":
@@ -1072,7 +1072,7 @@ def get_pipeline(
                 not format_
                 or format_ == mlrun.api.schemas.PipelinesFormat.summary.value
             ):
-                resp = format_summary_from_kfp_run(resp["run"])
+                resp = format_summary_from_kfp_run(resp)
 
     show_kfp_run(resp)
     return resp

tests/system/projects/test_project.py

@@ -83,7 +83,7 @@ def _create_project(self, project_name, with_repo=False):
             doc="model package/algorithm",
         )
         proj.set_workflow("main", "./kflow.py", args_schema=[arg])
-        proj.set_workflow("newflow", "./newflow.py")
+        proj.set_workflow("newflow", "./newflow.py", handler="newpipe")
         proj.save()
         return proj
 
@@ -218,7 +218,6 @@ def _test_new_pipeline(self, name, engine):
         print(project.to_yaml())
         run = project.run(
             "newflow",
-            workflow_handler="newpipe",
             engine=engine,
             artifact_path=f"v3io:///projects/{name}",
             watch=True,
@@ -248,8 +247,6 @@ def test_local_cli(self):
             name,
             "-r",
             "newflow",
-            "--handler",
-            "newpipe",
             "--engine",
             "local",
             "-w",