adding executeAPI (#165) (#179)

* adding executeAPI

Signed-off-by: Alibi Zhenis <alibizhenis4@gmail.com>

* updating documentation

Signed-off-by: Alibi Zhenis <alibizhenis4@gmail.com>

* changing code and test

Signed-off-by: Alibi Zhenis <alibizhenis4@gmail.com>

* changing code

Signed-off-by: Alibi Zhenis <alibizhenis4@gmail.com>

* fixing tests

Signed-off-by: Alibi Zhenis <alibizhenis4@gmail.com>

* changing tests

Signed-off-by: Alibi Zhenis <alibizhenis4@gmail.com>

* checking output value in integration tests

Signed-off-by: Alibi Zhenis <alibizhenis4@gmail.com>

---------

Signed-off-by: Alibi Zhenis <alibizhenis4@gmail.com>
(cherry picked from commit 8eb26eb)

Co-authored-by: Alibi Zhenis <92104549+AlibiZhenis@users.noreply.github.com>

docs/source/reference/api/ml_commons_execute_api.rst

@@ -0,0 +1,6 @@
+Execute
+==================
+
+.. currentmodule:: opensearch_py_ml
+
+.. autofunction:: opensearch_py_ml.ml_commons.MLCommonClient.execute

docs/source/reference/mlcommons.rst

@@ -96,3 +96,9 @@ Delete Task
 
    api/ml_commons_delete_task_api
 
+Execute
+~~~~~~~~~~~~~~~~~
+.. toctree::
+   :maxdepth: 2
+
+   api/ml_commons_execute_api

opensearch_py_ml/ml_commons/ml_commons_client.py

@@ -13,6 +13,7 @@
 from opensearchpy import OpenSearch
 
 from opensearch_py_ml.ml_commons.ml_common_utils import ML_BASE_URI, TIMEOUT
+from opensearch_py_ml.ml_commons.model_execute import ModelExecute
 from opensearch_py_ml.ml_commons.model_uploader import ModelUploader
 
 
@@ -25,6 +26,22 @@ class MLCommonClient:
     def __init__(self, os_client: OpenSearch):
         self._client = os_client
         self._model_uploader = ModelUploader(os_client)
+        self._model_execute = ModelExecute(os_client)
+
+    def execute(self, algorithm_name: str, input_json: dict) -> dict:
+        """
+        This method executes ML algorithms that can be only executed directly (i.e. do not support train and
+        predict APIs), like anomaly localization and metrics correlation. The algorithm has to be supported by ML Commons.
+        Refer to https://opensearch.org/docs/2.7/ml-commons-plugin/api/#execute
+
+        :param algorithm_name: Name of the algorithm
+        :type algorithm_name: string
+        :param input_json: Dictionary of parameters
+        :type input_json: dict
+        :return: returns the API response
+        :rtype: dict
+        """
+        return self._model_execute._execute(algorithm_name, input_json)
 
     @deprecated(
         reason="Since OpenSearch 2.7.0, you can use register_model instead",

opensearch_py_ml/ml_commons/model_execute.py

@@ -0,0 +1,48 @@
+# SPDX-License-Identifier: Apache-2.0
+# The OpenSearch Contributors require contributions made to
+# this file be licensed under the Apache-2.0 license or a
+# compatible open source license.
+# Any modifications Copyright OpenSearch Contributors. See
+# GitHub history for details.
+
+import json
+
+from opensearchpy import OpenSearch
+
+from opensearch_py_ml.ml_commons.ml_common_utils import ML_BASE_URI
+
+
+class ModelExecute:
+    """
+    Class for executing algorithms using ML Commons execute API.
+    """
+
+    API_ENDPOINT = "models/_execute"
+
+    def __init__(self, os_client: OpenSearch):
+        self._client = os_client
+
+    def _execute(self, algorithm_name: str, input_json: dict) -> dict:
+        """
+        This method executes ML algorithms that can be only executed directly (i.e. do not support train and
+        predict APIs), like anomaly localization and metrics correlation.
+        The input json must be a dictionary or a deserializable Python object.
+        The algorithm has to be supported by ML Commons.
+        Refer to https://opensearch.org/docs/2.7/ml-commons-plugin/api/#execute
+
+        :param algorithm_name: Name of the algorithm
+        :type algorithm_name: string
+        :param input_json: Dictionary of parameters or a deserializable string, byte, or bytearray
+        :type input_json: dict
+        :return: returns the API response
+        :rtype: dict
+        """
+
+        if not isinstance(input_json, dict):
+            input_json = json.loads(input_json)
+
+        return self._client.transport.perform_request(
+            method="POST",
+            url=f"{ML_BASE_URI}/_execute/{algorithm_name}",
+            body=input_json,
+        )

tests/ml_commons/test_ml_commons_client.py

@@ -72,6 +72,34 @@ def test_init():
     assert type(ml_client._model_uploader) == ModelUploader
 
 
+def test_execute():
+    raised = False
+    try:
+        input_json = {"operation": "max", "input_data": [1.0, 2.0, 3.0]}
+        result = ml_client.execute(
+            algorithm_name="local_sample_calculator", input_json=input_json
+        )
+        assert result["output"]["result"] == 3
+    except:  # noqa: E722
+        raised = True
+    assert (
+        raised == False
+    ), "Raised Exception during execute API testing with dictionary"
+
+    raised = False
+    try:
+        input_json = '{"operation": "max", "input_data": [1.0, 2.0, 3.0]}'
+        result = ml_client.execute(
+            algorithm_name="local_sample_calculator", input_json=input_json
+        )
+        assert result["output"]["result"] == 3
+    except:  # noqa: E722
+        raised = True
+    assert (
+        raised == False
+    ), "Raised Exception during execute API testing with JSON string"
+
+
 def test_DEPRECATED_integration_pretrained_model_upload_unload_delete():
     raised = False
     try:
@@ -381,3 +409,6 @@ def test_integration_model_train_register_full_cycle():
                 except:  # noqa: E722
                     raised = True
                 assert raised == False, "Raised Exception in deleting model"
+
+
+test_integration_model_train_register_full_cycle()