[HWORKS-1048] Support multiple modularized project environments (logicalclocks#213)

Author: robzor92

python/hopsworks/core/environment_api.py

@@ -14,6 +14,9 @@
 #   limitations under the License.
 #
 
+import json
+from typing import List, Optional
+
 from hopsworks import client, environment
 from hopsworks.engine import environment_engine
 
@@ -29,7 +32,7 @@ def __init__(
 
         self._environment_engine = environment_engine.EnvironmentEngine(project_id)
 
-    def create_environment(self, await_creation=True):
+    def create_environment(self, name: str, description: Optional[str] = None, base_environment_name: Optional[str] = "python-feature-pipeline", await_creation: Optional[bool] = True) -> environment.Environment:
         """Create Python environment for the project
 
         ```python
@@ -40,10 +43,13 @@ def create_environment(self, await_creation=True):
 
         env_api = project.get_environment_api()
 
-        env = env_api.create_environment()
+        new_env = env_api.create_environment("my_custom_environment", base_environment_name="python-feature-pipeline")
+
 
         ```
         # Arguments
+            name: name of the environment
+            base_environment_name: the name of the environment to clone from
             await_creation: bool. If True the method returns only when the creation is finished. Default True
         # Returns
             `Environment`: The Environment object
@@ -57,21 +63,26 @@ def create_environment(self, await_creation=True):
             self._project_id,
             "python",
             "environments",
-            client.get_python_version(),
+            name,
         ]
         headers = {"content-type": "application/json"}
+        data = {"name": name,
+                "baseImage": {
+                    "name": base_environment_name,
+                    "description": description
+                }}
         env = environment.Environment.from_response_json(
-            _client._send_request("POST", path_params, headers=headers),
+            _client._send_request("POST", path_params, headers=headers, data=json.dumps(data)),
             self._project_id,
             self._project_name,
         )
 
         if await_creation:
-            self._environment_engine.await_environment_command()
+            self._environment_engine.await_environment_command(name)
 
         return env
 
-    def _get_environments(self):
+    def get_environments(self) -> List[environment.Environment]:
         """
         Get all available python environments in the project
         """
@@ -88,7 +99,7 @@ def _get_environments(self):
             self._project_name,
         )
 
-    def get_environment(self):
+    def get_environment(self, name: str) -> environment.Environment:
         """Get handle for the Python environment for the project
 
         ```python
@@ -99,30 +110,42 @@ def get_environment(self):
 
         env_api = project.get_environment_api()
 
-        env = env_api.get_environment()
+        env = env_api.get_environment("my_custom_environment")
 
         ```
+        # Arguments
+            name: name of the environment
         # Returns
             `Environment`: The Environment object
         # Raises
             `RestAPIError`: If unable to get the environment
         """
-        project_envs = self._get_environments()
-        if len(project_envs) == 0:
-            return None
-        elif len(project_envs) > 0:
-            return project_envs[0]
-
-    def _delete(self, python_version):
-        """Delete the project Python environment"""
+        _client = client.get_instance()
+
+        path_params = ["project", self._project_id, "python", "environments", name]
+        query_params = {"expand": ["libraries", "commands"]}
+        headers = {"content-type": "application/json"}
+        return environment.Environment.from_response_json(
+            _client._send_request(
+                "GET", path_params, query_params=query_params, headers=headers
+            ),
+            self._project_id,
+            self._project_name,
+        )
+
+    def _delete(self, name):
+        """Delete the Python environment.
+        :param name: name of environment to delete
+        :type environment: Environment
+        """
         _client = client.get_instance()
 
         path_params = [
             "project",
             self._project_id,
             "python",
             "environments",
-            python_version,
+            name,
         ]
         headers = {"content-type": "application/json"}
         _client._send_request("DELETE", path_params, headers=headers),

python/hopsworks/core/library_api.py

@@ -28,16 +28,27 @@ def __init__(
         self._project_id = project_id
         self._project_name = project_name
 
-    def install(self, library_name: str, python_version: str, library_spec: dict):
-        """Create Python environment for the project"""
+    def _install(self, library_name: str, name: str, library_spec: dict):
+        """Install a library in the environment
+
+        # Arguments
+            library_name: Name of the library.
+            name: Name of the environment.
+            library_spec: installation payload
+        # Returns
+            `Library`: The library object
+        # Raises
+            `RestAPIError`: If unable to install library
+        """
+
         _client = client.get_instance()
 
         path_params = [
             "project",
             self._project_id,
             "python",
             "environments",
-            python_version,
+            name,
             "libraries",
             library_name,
         ]

python/hopsworks/engine/environment_engine.py

@@ -16,29 +16,29 @@
 
 import time
 
-from hopsworks import client, library, environment, command
-from hopsworks.client.exceptions import RestAPIError, EnvironmentException
+from hopsworks import client, command, environment, library
+from hopsworks.client.exceptions import EnvironmentException, RestAPIError
 
 
 class EnvironmentEngine:
     def __init__(self, project_id):
         self._project_id = project_id
 
-    def await_library_command(self, library_name=None):
+    def await_library_command(self, environment_name, library_name):
         commands = [command.Command(status="ONGOING")]
         while len(commands) > 0 and not self._is_final_status(commands[0]):
             time.sleep(5)
-            library = self._poll_commands_library(library_name)
+            library = self._poll_commands_library(environment_name, library_name)
             if library is None:
                 commands = []
             else:
                 commands = library._commands
 
-    def await_environment_command(self):
+    def await_environment_command(self, environment_name):
         commands = [command.Command(status="ONGOING")]
         while len(commands) > 0 and not self._is_final_status(commands[0]):
             time.sleep(5)
-            environment = self._poll_commands_environment()
+            environment = self._poll_commands_environment(environment_name)
             if environment is None:
                 commands = []
             else:
@@ -54,15 +54,15 @@ def _is_final_status(self, command):
         else:
             return False
 
-    def _poll_commands_library(self, library_name):
+    def _poll_commands_library(self, environment_name, library_name):
         _client = client.get_instance()
 
         path_params = [
             "project",
             self._project_id,
             "python",
             "environments",
-            client.get_python_version(),
+            environment_name,
             "libraries",
             library_name,
         ]
@@ -85,15 +85,15 @@ def _poll_commands_library(self, library_name):
             ):
                 return None
 
-    def _poll_commands_environment(self):
+    def _poll_commands_environment(self, environment_name):
         _client = client.get_instance()
 
         path_params = [
             "project",
             self._project_id,
             "python",
             "environments",
-            client.get_python_version(),
+            environment_name,
         ]
 
         query_params = {"expand": "commands"}

python/hopsworks/environment.py

@@ -14,9 +14,10 @@
 #   limitations under the License.
 #
 
-import humps
 import os
+from typing import Optional
 
+import humps
 from hopsworks import command, util
 from hopsworks.core import environment_api, library_api
 from hopsworks.engine import environment_engine
@@ -25,9 +26,11 @@
 class Environment:
     def __init__(
         self,
-        python_version,
-        python_conflicts,
-        pip_search_enabled,
+        name=None,
+        description=None,
+        python_version=None,
+        python_conflicts=None,
+        pip_search_enabled=None,
         conflicts=None,
         conda_channel=None,
         libraries=None,
@@ -38,6 +41,8 @@ def __init__(
         project_name=None,
         **kwargs,
     ):
+        self._name = name
+        self._description = description
         self._python_version = python_version
         self._python_conflicts = python_conflicts
         self._pip_search_enabled = pip_search_enabled
@@ -77,7 +82,17 @@ def python_version(self):
         """Python version of the environment"""
         return self._python_version
 
-    def install_wheel(self, path, await_installation=True):
+    @property
+    def name(self):
+        """Name of the environment"""
+        return self._name
+
+    @property
+    def description(self):
+        """Description of the environment"""
+        return self._description
+
+    def install_wheel(self, path: str, await_installation: Optional[bool] = True):
         """Install a python library packaged in a wheel file
 
         ```python
@@ -92,18 +107,21 @@ def install_wheel(self, path, await_installation=True):
 
         # Install
         env_api = project.get_environment_api()
-        env = env_api.get_environment()
+        env = env_api.get_environment("my_custom_environment")
+
         env.install_wheel(whl_path)
 
         ```
 
         # Arguments
             path: str. The path on Hopsworks where the wheel file is located
             await_installation: bool. If True the method returns only when the installation finishes. Default True
+        # Returns
+            `Library`: The library object
         """
 
         # Wait for any ongoing environment operations
-        self._environment_engine.await_environment_command()
+        self._environment_engine.await_environment_command(self.name)
 
         library_name = os.path.basename(path)
 
@@ -115,16 +133,16 @@ def install_wheel(self, path, await_installation=True):
             "packageSource": "WHEEL",
         }
 
-        library_rest = self._library_api.install(
-            library_name, self.python_version, library_spec
+        library_rest = self._library_api._install(
+            library_name, self.name, library_spec
         )
 
         if await_installation:
-            return self._environment_engine.await_library_command(library_name)
+            return self._environment_engine.await_library_command(self.name, library_name)
 
         return library_rest
 
-    def install_requirements(self, path, await_installation=True):
+    def install_requirements(self, path: str, await_installation: Optional[bool] = True):
         """Install libraries specified in a requirements.txt file
 
         ```python
@@ -139,18 +157,22 @@ def install_requirements(self, path, await_installation=True):
 
         # Install
         env_api = project.get_environment_api()
-        env = env_api.get_environment()
+        env = env_api.get_environment("my_custom_environment")
+
+
         env.install_requirements(requirements_path)
 
         ```
 
         # Arguments
             path: str. The path on Hopsworks where the requirements.txt file is located
             await_installation: bool. If True the method returns only when the installation is finished. Default True
+        # Returns
+            `Library`: The library object
         """
 
         # Wait for any ongoing environment operations
-        self._environment_engine.await_environment_command()
+        self._environment_engine.await_environment_command(self.name)
 
         library_name = os.path.basename(path)
 
@@ -162,12 +184,12 @@ def install_requirements(self, path, await_installation=True):
             "packageSource": "REQUIREMENTS_TXT",
         }
 
-        library_rest = self._library_api.install(
-            library_name, self.python_version, library_spec
+        library_rest = self._library_api._install(
+            library_name, self.name, library_spec
         )
 
         if await_installation:
-            return self._environment_engine.await_library_command(library_name)
+            return self._environment_engine.await_library_command(self.name, library_name)
 
         return library_rest
 
@@ -178,7 +200,7 @@ def delete(self):
         # Raises
             `RestAPIError`.
         """
-        self._environment_api._delete(self.python_version)
+        self._environment_api._delete(self.name)
 
     def __repr__(self):
-        return f"Environment({self._python_version!r})"
+        return f"Environment({self.name!r})"