[SC-49206] Add run and stop commands to the cli (#323)

Changes:
Adds run/stop commands which underneath just call the REST API.

Tested:
Ran tox tests.

Author: calving-db

databricks_cli/pipelines/api.py

@@ -63,6 +63,12 @@ def get(self, pipeline_id, headers=None):
     def reset(self, pipeline_id, headers=None):
         self.client.reset(pipeline_id, headers)
 
+    def run(self, pipeline_id, headers=None):
+        self.client.run(pipeline_id, headers)
+
+    def stop(self, pipeline_id, headers=None):
+        self.client.stop(pipeline_id, headers)
+
     def _upload_libraries_and_update_spec(self, spec):
         spec = copy.deepcopy(spec)
         lib_objects = LibraryObject.from_json(spec.get('libraries', []))

databricks_cli/pipelines/cli.py

@@ -187,6 +187,49 @@ def reset_cli(api_client, pipeline_id):
     click.echo("Reset triggered for pipeline {}".format(pipeline_id))
 
 
+@click.command(context_settings=CONTEXT_SETTINGS,
+               short_help='Starts the execution of a delta pipeline run')
+@click.option('--pipeline-id', default=None, type=PipelineIdClickType(),
+              help=PipelineIdClickType.help)
+@debug_option
+@profile_option
+@pipelines_exception_eater
+@provide_api_client
+def run_cli(api_client, pipeline_id):
+    """
+    Starts the execution of a delta pipelines run by starting the cluster and processing data.
+
+    Usage:
+
+    databricks pipelines run --pipeline-id 1234
+    """
+    _validate_pipeline_id(pipeline_id)
+    PipelinesApi(api_client).run(pipeline_id)
+    click.echo("Run triggered for pipeline {}".format(pipeline_id))
+
+
+@click.command(context_settings=CONTEXT_SETTINGS,
+               short_help='Stops the execution of a delta pipeline run')
+@click.option('--pipeline-id', default=None, type=PipelineIdClickType(),
+              help=PipelineIdClickType.help)
+@debug_option
+@profile_option
+@pipelines_exception_eater
+@provide_api_client
+def stop_cli(api_client, pipeline_id):
+    """
+    Stops the execution of a delta pipelines run by terminating the cluster. Processing of data can
+    be resumed by calling `run`.
+
+    Usage:
+
+    databricks pipelines stop --pipeline-id 1234
+    """
+    _validate_pipeline_id(pipeline_id)
+    PipelinesApi(api_client).stop(pipeline_id)
+    click.echo("Stopped pipeline {}".format(pipeline_id))
+
+
 def _read_spec(src):
     """
     Reads the spec at src as a JSON if no file extension is provided, or if in the extension format
@@ -276,3 +319,5 @@ def pipelines_group():  # pragma: no cover
 pipelines_group.add_command(delete_cli, name='delete')
 pipelines_group.add_command(get_cli, name='get')
 pipelines_group.add_command(reset_cli, name='reset')
+pipelines_group.add_command(run_cli, name='run')
+pipelines_group.add_command(stop_cli, name='stop')

databricks_cli/sdk/service.py

@@ -938,3 +938,13 @@ def reset(self, pipeline_id=None, headers=None):
         _data = {}
 
         return self.client.perform_query('POST', '/pipelines/{pipeline_id}/reset'.format(pipeline_id=pipeline_id), data=_data, headers=headers)
+
+    def run(self, pipeline_id=None, headers=None):
+        _data = {}
+
+        return self.client.perform_query('POST', '/pipelines/{pipeline_id}/run'.format(pipeline_id=pipeline_id), data=_data, headers=headers)
+
+    def stop(self, pipeline_id=None, headers=None):
+        _data = {}
+
+        return self.client.perform_query('POST', '/pipelines/{pipeline_id}/stop'.format(pipeline_id=pipeline_id), data=_data, headers=headers)

tests/pipelines/test_api.py

@@ -221,6 +221,22 @@ def test_reset(pipelines_api):
                                    data={}, headers=None)
 
 
+def test_run(pipelines_api):
+    pipelines_api.run(PIPELINE_ID)
+    client_mock = pipelines_api.client.client.perform_query
+    assert client_mock.call_count == 1
+    client_mock.assert_called_with('POST', '/pipelines/{}/run'.format(PIPELINE_ID),
+                                   data={}, headers=None)
+
+
+def test_stop(pipelines_api):
+    pipelines_api.stop(PIPELINE_ID)
+    client_mock = pipelines_api.client.client.perform_query
+    assert client_mock.call_count == 1
+    client_mock.assert_called_with('POST', '/pipelines/{}/stop'.format(PIPELINE_ID),
+                                   data={}, headers=None)
+
+
 def test_partition_local_remote(pipelines_api):
     libraries = [
         # local files

tests/pipelines/test_cli.py

@@ -186,18 +186,20 @@ def test_deploy_update_delete_cli_correct_spec_extensions(pipelines_api_mock, tm
 
 
 @provide_conf
-def test_reset_cli_id(pipelines_api_mock):
-    runner = CliRunner()
-    runner.invoke(cli.reset_cli, ['--pipeline-id', PIPELINE_ID])
-    assert pipelines_api_mock.reset.call_args[0][0] == PIPELINE_ID
+def test_cli_id(pipelines_api_mock):
+    for command in [cli.reset_cli, cli.stop_cli, cli.run_cli]:
+        runner = CliRunner()
+        runner.invoke(command, ['--pipeline-id', PIPELINE_ID])
+        assert pipelines_api_mock.reset.call_args[0][0] == PIPELINE_ID
 
 
 @provide_conf
-def test_reset_cli_no_id(pipelines_api_mock):
-    runner = CliRunner()
-    result = runner.invoke(cli.reset_cli, [])
-    assert result.exit_code == 1
-    assert pipelines_api_mock.reset.call_count == 0
+def test_cli_no_id(pipelines_api_mock):
+    for command in [cli.reset_cli, cli.stop_cli, cli.run_cli]:
+        runner = CliRunner()
+        result = runner.invoke(command, [])
+        assert result.exit_code == 1
+        assert pipelines_api_mock.reset.call_count == 0
 
 
 @provide_conf