Partial merge of hsfs job/execution

Author: aversey

Diff for: python/auto_doc.py

@@ -57,11 +57,11 @@
     },
     "api/jobs.md": {
         "job_api_handle": ["hopsworks.project.Project.get_jobs_api"],
-        "job_create": ["hopsworks.core.job_api.JobsApi.create_job"],
-        "job_get": ["hopsworks.core.job_api.JobsApi.get_job"],
-        "job_get_all": ["hopsworks.core.job_api.JobsApi.get_jobs"],
+        "job_create": ["hopsworks.core.job_api.JobApi.create_job"],
+        "job_get": ["hopsworks.core.job_api.JobApi.get_job"],
+        "job_get_all": ["hopsworks.core.job_api.JobApi.get_jobs"],
         "job_properties": keras_autodoc.get_properties("hopsworks.job.Job"),
-        "job_config": ["hopsworks.core.job_api.JobsApi.get_configuration"],
+        "job_config": ["hopsworks.core.job_api.JobApi.get_configuration"],
         "job_methods": keras_autodoc.get_methods(
             "hopsworks.job.Job", exclude=["from_response_json", "json"]
         ),

Diff for: python/hopsworks/core/execution_api.py

@@ -15,10 +15,10 @@
 #
 
 from hopsworks_common.core.execution_api import (
-    ExecutionsApi,
+    ExecutionApi,
 )
 
 
 __all__ = [
-    ExecutionsApi,
+    ExecutionApi,
 ]

Diff for: python/hopsworks/core/job_api.py

@@ -15,10 +15,10 @@
 #
 
 from hopsworks_common.core.job_api import (
-    JobsApi,
+    JobApi,
 )
 
 
 __all__ = [
-    JobsApi,
+    JobApi,
 ]

Diff for: python/hopsworks/project.py

@@ -62,7 +62,7 @@ def __init__(
 
         self._opensearch_api = opensearch_api.OpenSearchApi(project_id, project_name)
         self._kafka_api = kafka_api.KafkaApi(project_id, project_name)
-        self._jobs_api = job_api.JobsApi(project_id, project_name)
+        self._jobs_api = job_api.JobApi(project_id, project_name)
         self._flink_cluster_api = flink_cluster_api.FlinkClusterApi(
             project_id, project_name
         )
@@ -230,16 +230,16 @@ def get_opensearch_api(self):
             _client.download_certs(self.name)
         return self._opensearch_api
 
-    def get_jobs_api(self):
-        """Get the jobs api for the project.
+    def get_job_api(self):
+        """Get the job API for the project.
 
         # Returns
-            `JobsApi`: The Jobs Api handle
+            `JobApi`: The Job Api handle
         """
         return self._jobs_api
 
     def get_flink_cluster_api(self):
-        """Get the flink cluster api for the project.
+        """Get the flink cluster API for the project.
 
         # Returns
             `FlinkClusterApi`: The Flink Cluster Api handle

Diff for: python/hopsworks_common/core/execution_api.py

@@ -17,7 +17,7 @@
 from hopsworks_common import client, execution
 
 
-class ExecutionsApi:
+class ExecutionApi:
     def __init__(
         self,
         project_id,

Diff for: python/hopsworks_common/core/flink_cluster_api.py

@@ -30,7 +30,7 @@ def __init__(
     ):
         self._project_id = project_id
         self._project_name = project_name
-        self._job_api = job_api.JobsApi(project_id, project_name)
+        self._job_api = job_api.JobApi(project_id, project_name)
 
     def get_configuration(self):
         """Get configuration for the Flink cluster.

Diff for: python/hopsworks_common/core/job_api.py

@@ -20,7 +20,7 @@
 from hopsworks_common.client.exceptions import RestAPIError
 
 
-class JobsApi:
+class JobApi:
     def __init__(
         self,
         project_id,

Diff for: python/hopsworks_common/engine/execution_engine.py

@@ -26,7 +26,7 @@
 class ExecutionEngine:
     def __init__(self, project_id=None):
         self._dataset_api = dataset_api.DatasetApi(project_id)
-        self._execution_api = execution_api.ExecutionsApi(project_id)
+        self._execution_api = execution_api.ExecutionApi(project_id)
         self._log = logging.getLogger(__name__)
 
     def download_logs(self, execution, path=None):

Diff for: python/hopsworks_common/execution.py

@@ -63,7 +63,7 @@ def __init__(
         self._project_id = project_id
 
         self._execution_engine = execution_engine.ExecutionEngine(project_id)
-        self._execution_api = execution_api.ExecutionsApi(project_id)
+        self._execution_api = execution_api.ExecutionApi(project_id)
 
     @classmethod
     def from_response_json(cls, json_dict, project_id=None, job=None):

Diff for: python/hopsworks_common/flink_cluster.py

@@ -35,7 +35,7 @@ def __init__(
         self._execution_id = None
 
         self._execution_engine = execution_engine.ExecutionEngine(project_id)
-        self._execution_api = execution_api.ExecutionsApi(project_id)
+        self._execution_api = execution_api.ExecutionApi(project_id)
         self._flink_cluster_api = flink_cluster_api.FlinkClusterApi(
             project_id, project_name
         )

Diff for: python/hopsworks_common/job.py

@@ -21,7 +21,7 @@
 
 import humps
 from hopsworks_common import util
-from hopsworks_common.client.exceptions import JobExecutionException
+from hopsworks_common.client.exceptions import JobException
 from hopsworks_common.core import execution_api, job_api
 from hopsworks_common.engine import execution_engine
 from hopsworks_common.job_schedule import JobSchedule
@@ -63,13 +63,12 @@ def __init__(
         )
 
         self._execution_engine = execution_engine.ExecutionEngine(project_id)
-        self._execution_api = execution_api.ExecutionsApi(project_id)
+        self._execution_api = execution_api.ExecutionApi(project_id)
         self._execution_engine = execution_engine.ExecutionEngine(project_id)
-        self._job_api = job_api.JobsApi(project_id, project_name)
+        self._job_api = job_api.JobApi(project_id, project_name)
 
     @classmethod
     def from_response_json(cls, json_dict, project_id=None, project_name=None):
-        print(json_dict)
         if "items" in json_dict:
             jobs = []
             for job in json_dict["items"]:
@@ -203,7 +202,7 @@ def get_state(self):
         """
         last_execution = self._job_api.last_execution(self)
         if len(last_execution) != 1:
-            raise JobExecutionException("No executions found for job")
+            raise JobException("No executions found for job")
 
         return last_execution[0].state
 
@@ -218,7 +217,7 @@ def get_final_state(self):
         """
         last_execution = self._job_api.last_execution(self)
         if len(last_execution) != 1:
-            raise JobExecutionException("No executions found for job")
+            raise JobException("No executions found for job")
 
         return last_execution[0].final_status
 
@@ -303,7 +302,7 @@ def unschedule(self):
     def resume_schedule(self):
         """Resumes the schedule of a Job execution"""
         if self._job_schedule is None:
-            raise JobExecutionException("No schedule found for job")
+            raise JobException("No schedule found for job")
 
         job_schedule = JobSchedule(
             id=self._job_schedule.id,
@@ -317,7 +316,7 @@ def resume_schedule(self):
     def pause_schedule(self):
         """Pauses the schedule of a Job execution"""
         if self._job_schedule is None:
-            raise JobExecutionException("No schedule found for job")
+            raise JobException("No schedule found for job")
 
         job_schedule = JobSchedule(
             id=self._job_schedule.id,

Diff for: python/tests/core/test_job.py

@@ -49,84 +49,84 @@ def test_from_response_json_empty(self, backend_fixtures):
 
     def test_wait_for_job(self, mocker, backend_fixtures):
         # Arrange
-        mock_job_api = mocker.patch("hsfs.core.job_api.JobApi")
+        mocker.patch("hopsworks_common.client.get_instance")
+        mock_execution_api = mocker.patch(
+            "hopsworks_common.core.execution_api.ExecutionApi",
+        )
 
         json = backend_fixtures["job"]["get"]["response"]
-        j = job.Job.from_response_json(json)
+        x = job.Job.from_response_json(json).run(await_termination=False)
 
         # Act
-        j._wait_for_job()
+        x.await_termination()
 
         # Assert
-        assert mock_job_api.return_value.last_execution.call_count == 1
+        assert mock_execution_api.return_value._get.call_count == 1
 
     def test_wait_for_job_wait_for_job_false(self, mocker, backend_fixtures):
         # Arrange
-        mock_job_api = mocker.patch("hsfs.core.job_api.JobApi")
+        mock_job_api = mocker.patch("hopsworks_common.core.execution_api.ExecutionApi")
 
         json = backend_fixtures["job"]["get"]["response"]
-        j = job.Job.from_response_json(json)
-
-        # Act
-        j._wait_for_job(False)
+        job.Job.from_response_json(json).run(await_termination=False)
 
         # Assert
-        assert mock_job_api.return_value.last_execution.call_count == 0
+        assert mock_job_api.return_value._get.call_count == 0
 
     def test_wait_for_job_final_status_succeeded(self, mocker, backend_fixtures):
         # Arrange
-        mock_job_api = mocker.patch("hsfs.core.job_api.JobApi")
+        mock_job_api = mocker.patch("hopsworks_common.core.execution_api.ExecutionApi")
 
         json = backend_fixtures["job"]["get"]["response"]
-        j = job.Job.from_response_json(json)
+        x = job.Job.from_response_json(json).run(await_termination=False)
 
-        mock_job_api.return_value.last_execution.return_value = [
+        mock_job_api.return_value._get.return_value = [
             execution.Execution(id=1, state=None, final_status="succeeded")
         ]
 
         # Act
-        j._wait_for_job()
+        x.await_termination()
 
         # Assert
-        assert mock_job_api.return_value.last_execution.call_count == 1
+        assert mock_job_api.return_value._get.call_count == 1
 
     def test_wait_for_job_final_status_failed(self, mocker, backend_fixtures):
         # Arrange
-        mock_job_api = mocker.patch("hsfs.core.job_api.JobApi")
+        mock_job_api = mocker.patch("hopsworks_common.core.execution_api.ExecutionApi")
 
         json = backend_fixtures["job"]["get"]["response"]
-        j = job.Job.from_response_json(json)
+        x = job.Job.from_response_json(json).run(await_termination=False)
 
-        mock_job_api.return_value.last_execution.return_value = [
+        mock_job_api.return_value._get.return_value = [
             execution.Execution(id=1, state=None, final_status="failed")
         ]
 
         # Act
-        with pytest.raises(exceptions.FeatureStoreException) as e_info:
-            j._wait_for_job()
+        with pytest.raises(exceptions.JobException) as e_info:
+            x.await_termination()
 
         # Assert
-        assert mock_job_api.return_value.last_execution.call_count == 1
+        assert mock_job_api.return_value._get.call_count == 1
         assert (
             str(e_info.value)
             == "The Hopsworks Job failed, use the Hopsworks UI to access the job logs"
         )
 
     def test_wait_for_job_final_status_killed(self, mocker, backend_fixtures):
         # Arrange
-        mock_job_api = mocker.patch("hsfs.core.job_api.JobApi")
+        mock_job_api = mocker.patch("hopsworks_common.core.execution_api.ExecutionApi")
 
         json = backend_fixtures["job"]["get"]["response"]
-        j = job.Job.from_response_json(json)
+        x = job.Job.from_response_json(json).run(await_termination=False)
 
-        mock_job_api.return_value.last_execution.return_value = [
+        mock_job_api.return_value._get.return_value = [
             execution.Execution(id=1, state=None, final_status="killed")
         ]
 
         # Act
-        with pytest.raises(exceptions.FeatureStoreException) as e_info:
-            j._wait_for_job()
+        with pytest.raises(exceptions.JobException) as e_info:
+            x.await_termination()
 
         # Assert
-        assert mock_job_api.return_value.last_execution.call_count == 1
+        assert mock_job_api.return_value._get.call_count == 1
         assert str(e_info.value) == "The Hopsworks Job was stopped"

Diff for: python/tests/engine/test_python.py

@@ -2354,7 +2354,7 @@ def test_write_training_dataset(self, mocker):
         mock_td_api = mocker.patch("hsfs.core.training_dataset_api.TrainingDatasetApi")
         mocker.patch("hsfs.util.get_job_url")
         mock_python_engine_wait_for_job = mocker.patch(
-            "hsfs.core.job.Job._wait_for_job"
+            "hopsworks_common.engine.execution_engine.ExecutionEngine.wait_until_finished"
         )
 
         python_engine = python.Engine()