python.py

from learning_orchestra_client.observe.observe import Observer
from learning_orchestra_client._util._response_treat import ResponseTreat
from learning_orchestra_client._util._entity_reader import EntityReader
import requests
from typing import Union


class FunctionPython:
    __CODE_FIELD = "function"
    __PARAMETERS_FIELD = "functionParameters"
    __NAME_FIELD = "name"
    __DESCRIPTION_FIELD = "description"

    def __init__(self, cluster_ip: str):
        self.__api_path = "/api/learningOrchestra/v1/function/python"
        self.__service_url = f'{cluster_ip}{self.__api_path}'
        self.__response_treat = ResponseTreat()
        self.__cluster_ip = cluster_ip
        self.__entity_reader = EntityReader(self.__service_url)
        self.__observer = Observer(self.__cluster_ip)

    def run_function_sync(self,
                          name: str,
                          parameters: dict,
                          code: str,
                          description: str = "",
                          pretty_response: bool = False) -> Union[dict, str]:
        """
        description: This method runs a python 3 code in sync mode, so it
        represents a wildcard for the data scientist. It can be used when
        train, predict, tune, explore or any other pipe must be customized. The
        function is also useful for new pipes. pretty_response: If true it
        returns a string, otherwise a dictionary.

        name: Is the name of the object stored in Learning Orchestra storage
        system (volume or mongoDB).
        url: Url to CSV file.

        return: A JSON object with an error or warning message or the correct
        operation result.
        """
        request_body = {
            self.__NAME_FIELD: name,
            self.__PARAMETERS_FIELD: parameters,
            self.__CODE_FIELD: code,
            self.__DESCRIPTION_FIELD: description}

        request_url = self.__service_url
        response = requests.post(url=request_url, json=request_body)
        self.__observer.wait(name)

        return self.__response_treat.treatment(response, pretty_response)

    def run_function_async(self,
                           name: str,
                           parameters: dict,
                           code: str,
                           description: str = "",
                           pretty_response: bool = False) -> Union[dict, str]:
        """
        description: This method runs a python 3 code in async mode, so it
        represents a wildcard for the data scientist. It does not lock the
        caller, so a wait method must be used. It can be used when train,
        predict, tune, explore or any other pipe must be customized. The
        function is also useful for new pipes.

        pretty_response: If true it returns a string, otherwise a dictionary.
        name: Is the name of the function to be called
        code: the Python code
        parameters: the parameters of the function being called

        return: A JSON object with an error or warning message or the correct
        operation result.
        """
        request_body = {
            self.__NAME_FIELD: name,
            self.__PARAMETERS_FIELD: parameters,
            self.__CODE_FIELD: code,
            self.__DESCRIPTION_FIELD: description}

        request_url = self.__service_url

        response = requests.post(url=request_url, json=request_body)
        return self.__response_treat.treatment(response, pretty_response)

    def search_all_executions(self, pretty_response: bool = False) \
            -> Union[dict, str]:
        """
        description: This method retrieves all created functions metadata,
        i.e., it does not retrieve the function result content.

        pretty_response: If true it returns a string, otherwise a dictionary.

        return: All function executions metadata stored in Learning Orchestra
        or an empty result.
        """
        response = self.__entity_reader.read_all_instances_from_entity()
        return self.__response_treat.treatment(response, pretty_response)

    def delete_execution(self, name: str, pretty_response=False) \
            -> Union[dict, str]:
        """
        description: This method is responsible for deleting the function.
        This delete operation is asynchronous, so it does not lock the caller
         until the deletion finished. Instead, it returns a JSON object with a
         URL for a future use. The caller uses the URL for delete checks.

        pretty_response: If true it returns a string, otherwise a dictionary.
        name: Represents the function name.

        return: JSON object with an error message, a warning message or a
        correct delete message
        """

        request_url = f'{self.__service_url}/{name}'

        response = requests.delete(request_url)
        return self.__response_treat.treatment(response, pretty_response)

    def search_execution_content(self,
                                 name: str,
                                 query: dict = {},
                                 limit: int = 10,
                                 skip: int = 0,
                                 pretty_response: bool = False) \
            -> Union[dict, str]:
        """
        description:  This method is responsible for retrieving the function
        results, including metadata. A function is executed many times, using
        different parameters,
        thus many results are stored
        in Learning Orchestra.

        pretty_response: If true it returns a string, otherwise a dictionary.
        name: Is the name of the function.
        query: Query to make in MongoDB(default: empty query)
        limit: Number of rows to return in pagination(default: 10) (maximum is
        set at 20 rows per request)
        skip: Number of rows to skip in pagination(default: 0)

        return:
         A page with some function results inside or an error if there
        is no such function. The current page is also returned to be used in
        future content requests.
        """

        response = self.__entity_reader.read_entity_content(
            name, query, limit, skip)

        return self.__response_treat.treatment(response, pretty_response)

    def wait(self, dataset_name: str, timeout: int = None) -> dict:
        """
           description: This method is responsible to create a synchronization
           barrier for the run_function_async method or delete_function method.

           name: Represents the function name.
           timeout: Represents the time in seconds to wait for a function to
           finish its run.

           return: JSON object with an error message, a warning message or a
           correct function result
        """
        return self.__observer.wait(dataset_name, timeout)