Accept class definition from Python Client directly (#1587)

Author: vishalbollu

dev/generate_python_client_md.sh

@@ -52,7 +52,7 @@ sed -i "s/# cortex.client.Client/# cortex.client.Client\n/g" $ROOT/docs/miscella
 sed -i "s/](#cortex\./](#/g" $ROOT/docs/miscellaneous/python-client.md
 sed -i "s/](#client\.Client\./](#/g" $ROOT/docs/miscellaneous/python-client.md
 
-# indentdation
+# indentation
 sed -i "s/    \* /  \* /g" $ROOT/docs/miscellaneous/python-client.md
 sed -i "s/#### /## /g" $ROOT/docs/miscellaneous/python-client.md
 
@@ -61,12 +61,3 @@ sed -i 's/[[:space:]]*$//' $ROOT/docs/miscellaneous/python-client.md
 truncate -s -1 $ROOT/docs/miscellaneous/python-client.md
 
 pip3 uninstall -y cortex
-
-cat << EOF
-
-#### MANUAL EDITS REQUIRED ####
-
-- Copy the docstring for \`client(env: str)\` in pkg/workloads/cortex/client/__init__.py into the generated docs and unindent
-
-Then check the diff
-EOF

docs/cluster-management/install.md

@@ -56,7 +56,20 @@ import requests
 local_client = cortex.client("local")
 
 # deploy the model as a realtime api and wait for it to become active
-deployments = local_client.deploy("./cortex.yaml", wait=True)
+
+api_spec={
+  "name": "iris-classifier",
+  "kind": "RealtimeAPI",
+  "predictor": {
+    "type": "python",
+    "path": "predictor.py",
+    "config": {
+      "model": "s3://cortex-examples/pytorch/iris-classifier/weights.pth"
+    }
+  }
+}
+
+deployments = local_client.deploy(api_spec, project_dir=".", wait=True)
 
 # get the api's endpoint
 url = deployments[0]["api"]["endpoint"]

docs/miscellaneous/python-client.md

@@ -29,37 +29,6 @@ client(env: str)
 
 Initialize a client based on the specified environment.
 
-To deploy and manage APIs on a new cluster:
-
-1. Spin up a cluster using the CLI command `cortex cluster up`.
-    An environment named "aws" will be created once the cluster is ready.
-2. Initialize your client:
-
-    ```python
-    import cortex
-    c = cortex.client("aws")
-    c.deploy("./cortex.yaml")
-    ```
-
-To deploy and manage APIs on an existing cluster:
-
-1. Use the command `cortex cluster info` to get the Operator Endpoint.
-2. Configure a client to your cluster:
-
-    ```python
-    import cortex
-    c = cortex.cluster_client("aws", operator_endpoint, aws_access_key_id, aws_secret_access_key)
-    c.deploy("./cortex.yaml")
-    ```
-
-To deploy and manage APIs locally:
-
-```python
-import cortex
-c = cortex.client("local")
-c.deploy("./cortex.yaml")
-```
-
 **Arguments**:
 
 - `env` - Name of the environment to use.
@@ -136,14 +105,23 @@ Delete an environment configured on this machine.
 ## deploy
 
 ```python
- | deploy(config_file: str, force: bool = False, wait: bool = False) -> list
+ | deploy(api_spec: dict, predictor=None, pip_dependencies=[], conda_dependencies=[], project_dir: Optional[str] = None, force: bool = False, wait: bool = False) -> list
 ```
 
-Deploy or update APIs specified in the config_file.
+Deploy an API.
 
 **Arguments**:
 
-- `config_file` - Local path to a yaml file defining Cortex APIs.
+- `api_spec` - A dictionary defining a single Cortex API. Schema can be found here:
+  → Realtime API: https://docs.cortex.dev/v/master/deployments/realtime-api/api-configuration
+  → Batch API: https://docs.cortex.dev/v/master/deployments/batch-api/api-configuration
+  → Traffic Splitter: https://docs.cortex.dev/v/master/deployments/realtime-api/traffic-splitter
+- `predictor` - A Cortex Predictor class implementation. Not required when deploying a traffic splitter.
+  → Realtime API: https://docs.cortex.dev/v/master/deployments/realtime-api/predictors
+  → Batch API: https://docs.cortex.dev/v/master/deployments/batch-api/predictors
+- `pip_dependencies` - A list of PyPI dependencies that will be installed before running your predictor class.
+- `conda_dependencies` - A list of Conda dependencies that will be installed before running your predictor class.
+- `project_dir` - Path to a python project.
 - `force` - Override any in-progress api updates.
 - `wait` - Streams logs until the APIs are ready.
 

pkg/workloads/cortex/client/README.md

@@ -51,7 +51,7 @@ import requests
 local_client = cortex.client("local")
 
 # deploy the model as a realtime api and wait for it to become active
-deployments = local_client.deploy("./cortex.yaml", wait=True)
+deployments = local_client.deploy_project(config_file="./cortex.yaml", wait=True)
 
 # get the api's endpoint
 url = deployments[0]["api"]["endpoint"]

pkg/workloads/cortex/client/cortex/__init__.py

@@ -24,37 +24,6 @@ def client(env: str):
     """
     Initialize a client based on the specified environment.
 
-    To deploy and manage APIs on a new cluster:
-
-    1. Spin up a cluster using the CLI command `cortex cluster up`.
-        An environment named "aws" will be created once the cluster is ready.
-    2. Initialize your client:
-
-        ```python
-        import cortex
-        c = cortex.client("aws")
-        c.deploy("./cortex.yaml")
-        ```
-
-    To deploy and manage APIs on an existing cluster:
-
-    1. Use the command `cortex cluster info` to get the Operator Endpoint.
-    2. Configure a client to your cluster:
-
-        ```python
-        import cortex
-        c = cortex.cluster_client("aws", operator_endpoint, aws_access_key_id, aws_secret_access_key)
-        c.deploy("./cortex.yaml")
-        ```
-
-    To deploy and manage APIs locally:
-
-    ```python
-    import cortex
-    c = cortex.client("local")
-    c.deploy("./cortex.yaml")
-    ```
-
     Args:
         env: Name of the environment to use.
 

pkg/workloads/cortex/client/cortex/client.py

@@ -18,9 +18,15 @@
 import sys
 import subprocess
 import threading
+import yaml
+import uuid
+import dill
+import inspect
+from pathlib import Path
 
 from typing import List, Dict, Optional, Tuple, Callable, Union
 from cortex.binary import run_cli, get_cli_path
+from cortex import util
 
 
 class Client:
@@ -33,7 +39,107 @@ def __init__(self, env: str):
         """
         self.env = env
 
+    # CORTEX_VERSION_MINOR x5
     def deploy(
+        self,
+        api_spec: dict,
+        predictor=None,
+        pip_dependencies=[],
+        conda_dependencies=[],
+        project_dir: Optional[str] = None,
+        force: bool = False,
+        wait: bool = False,
+    ) -> list:
+        """
+        Deploy an API.
+
+        Args:
+            api_spec: A dictionary defining a single Cortex API. Schema can be found here:
+                → Realtime API: https://docs.cortex.dev/v/master/deployments/realtime-api/api-configuration
+                → Batch API: https://docs.cortex.dev/v/master/deployments/batch-api/api-configuration
+                → Traffic Splitter: https://docs.cortex.dev/v/master/deployments/realtime-api/traffic-splitter
+            predictor: A Cortex Predictor class implementation. Not required when deploying a traffic splitter.
+                → Realtime API: https://docs.cortex.dev/v/master/deployments/realtime-api/predictors
+                → Batch API: https://docs.cortex.dev/v/master/deployments/batch-api/predictors
+            pip_dependencies: A list of PyPI dependencies that will be installed before the predictor class implementation is invoked.
+            conda_dependencies: A list of Conda dependencies that will be installed before the predictor class implementation is invoked.
+            project_dir: Path to a python project.
+            force: Override any in-progress api updates.
+            wait: Streams logs until the APIs are ready.
+
+        Returns:
+            Deployment status, API specification, and endpoint for each API.
+        """
+
+        if project_dir is not None and predictor is not None:
+            raise ValueError(
+                "`predictor` and `project_dir` parameters cannot be specified at the same time, please choose one"
+            )
+
+        if project_dir is not None:
+            cortex_yaml_path = os.path.join(project_dir, f".cortex-{uuid.uuid4()}.yaml")
+
+            with util.open_temporarily(cortex_yaml_path, "w") as f:
+                yaml.dump([api_spec], f)  # write a list
+                return self._deploy(cortex_yaml_path, force, wait)
+
+        project_dir = Path.home() / ".cortex" / "deployments" / str(uuid.uuid4())
+        with util.open_tempdir(str(project_dir)):
+            cortex_yaml_path = os.path.join(project_dir, "cortex.yaml")
+
+            if predictor is None:
+                # for deploying a traffic splitter
+                with open(cortex_yaml_path, "w") as f:
+                    yaml.dump([api_spec], f)  # write a list
+                    return self._deploy(cortex_yaml_path, force=force, wait=wait)
+
+            # Change if PYTHONVERSION changes
+            expected_version = "3.6"
+            actual_version = f"{sys.version_info.major}.{sys.version_info.minor}"
+            if actual_version < expected_version:
+                raise Exception("cortex is only supported for python versions >= 3.6")  # unexpected
+            if actual_version > expected_version:
+                is_python_set = any(
+                    conda_dep.startswith("python=") or "::python=" in conda_dep
+                    for conda_dep in conda_dependencies
+                )
+
+                if not is_python_set:
+                    conda_dependencies = [
+                        f"conda-forge::python={sys.version_info.major}.{sys.version_info.minor}.{sys.version_info.micro}"
+                    ] + conda_dependencies
+
+            if len(pip_dependencies) > 0:
+                with open(project_dir / "requirements.txt", "w") as requirements_file:
+                    requirements_file.write("\n".join(pip_dependencies))
+
+            if len(conda_dependencies) > 0:
+                with open(project_dir / "conda-packages.txt", "w") as conda_file:
+                    conda_file.write("\n".join(conda_dependencies))
+
+            if not inspect.isclass(predictor):
+                raise ValueError("predictor parameter must be a class definition")
+
+            with open(project_dir / "predictor.pickle", "wb") as pickle_file:
+                dill.dump(predictor, pickle_file)
+                if api_spec.get("predictor") is None:
+                    api_spec["predictor"] = {}
+
+                if predictor.__name__ == "PythonPredictor":
+                    predictor_type = "python"
+                if predictor.__name__ == "TensorFlowPredictor":
+                    predictor_type = "tensorflow"
+                if predictor.__name__ == "ONNXPredictor":
+                    predictor_type = "onnx"
+
+                api_spec["predictor"]["path"] = "predictor.pickle"
+                api_spec["predictor"]["type"] = predictor_type
+
+            with open(cortex_yaml_path, "w") as f:
+                yaml.dump([api_spec], f)  # write a list
+                return self._deploy(cortex_yaml_path, force=force, wait=wait)
+
+    def _deploy(
         self,
         config_file: str,
         force: bool = False,

pkg/workloads/cortex/client/cortex/util.py

@@ -0,0 +1,39 @@
+# Copyright 2020 Cortex Labs, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from contextlib import contextmanager
+import os
+from pathlib import Path
+import shutil
+
+
+@contextmanager
+def open_temporarily(path, mode):
+    file = open(path, mode)
+
+    try:
+        yield file
+    finally:
+        file.close()
+        os.remove(path)
+
+
+@contextmanager
+def open_tempdir(dir_path):
+    Path(dir_path).mkdir(parents=True)
+
+    try:
+        yield dir_path
+    finally:
+        shutil.rmtree(dir_path)

pkg/workloads/cortex/client/setup.py

@@ -94,7 +94,13 @@ def run(self):
             "cortex = cortex.binary:run",
         ],
     },
-    install_requires=(["importlib-resources; python_version < '3.7'"]),
+    install_requires=(
+        [
+            "importlib-resources; python_version < '3.7'",
+            "pyyaml>=5.3.0",
+            "dill==0.3.2",  # lines up with dill package version used in cortex serving code
+        ]
+    ),
     python_requires=">=3.6",
     cmdclass={
         "install": InstallBinary,

pkg/workloads/cortex/lib/api/predictor.py

@@ -246,25 +246,16 @@ def class_impl(self, project_dir):
 
         try:
             with FileLock("/run/init_stagger.lock"):
-                impl = self._load_module("cortex_predictor", os.path.join(project_dir, self.path))
+                predictor_class = self._get_class_impl(
+                    "cortex_predictor", os.path.join(project_dir, self.path), target_class_name
+                )
         except CortexException as e:
             e.wrap("error in " + self.path)
             raise
         finally:
             refresh_logger()
 
         try:
-            classes = inspect.getmembers(impl, inspect.isclass)
-            predictor_class = None
-            for class_df in classes:
-                if class_df[0] == target_class_name:
-                    if predictor_class is not None:
-                        raise UserException(
-                            f"multiple definitions for {target_class_name} class found; please check your imports and class definitions and ensure that there is only one Predictor class definition"
-                        )
-                    predictor_class = class_df[1]
-            if predictor_class is None:
-                raise UserException(f"{target_class_name} class is not defined")
             _validate_impl(predictor_class, validations, self.api_spec)
             if self.type == PythonPredictorType:
                 _validate_python_predictor_with_models(predictor_class, self.api_spec)
@@ -273,24 +264,32 @@ def class_impl(self, project_dir):
             raise
         return predictor_class
 
-    def _load_module(self, module_name, impl_path):
+    def _get_class_impl(self, module_name, impl_path, target_class_name):
         if impl_path.endswith(".pickle"):
             try:
-                impl = imp.new_module(module_name)
-
                 with open(impl_path, "rb") as pickle_file:
-                    pickled_dict = dill.load(pickle_file)
-                    for key in pickled_dict:
-                        setattr(impl, key, pickled_dict[key])
+                    return dill.load(pickle_file)
             except Exception as e:
                 raise UserException("unable to load pickle", str(e)) from e
-        else:
-            try:
-                impl = imp.load_source(module_name, impl_path)
-            except Exception as e:
-                raise UserException(str(e)) from e
 
-        return impl
+        try:
+            impl = imp.load_source(module_name, impl_path)
+        except Exception as e:
+            raise UserException(str(e)) from e
+
+        classes = inspect.getmembers(impl, inspect.isclass)
+        predictor_class = None
+        for class_df in classes:
+            if class_df[0] == target_class_name:
+                if predictor_class is not None:
+                    raise UserException(
+                        f"multiple definitions for {target_class_name} class found; please check your imports and class definitions and ensure that there is only one Predictor class definition"
+                    )
+                predictor_class = class_df[1]
+        if predictor_class is None:
+            raise UserException(f"{target_class_name} class is not defined")
+
+        return predictor_class
 
     def _is_model_caching_enabled(self) -> bool:
         """