Merge pull request #58 from dataiku/feature/project-apps

[WIP] Dataiku Apps API draft

Author: cstenac

dataikuapi/dss/app.py

@@ -0,0 +1,133 @@
+import sys
+import os.path as osp
+from .future import DSSFuture
+from dataikuapi.utils import DataikuException
+import random, string
+
+def random_string(length):
+    return ''.join(random.choice(string.ascii_letters) for _ in range(length))
+
+class DSSApp(object):
+    """
+    A handle to interact with a app on the DSS instance.
+    Do not create this class directly, instead use :meth:`dataikuapi.DSSClient.get_app``
+    """
+    def __init__(self, client, app_id):
+       self.client = client
+       self.app_id = app_id
+
+    ########################################################
+    # Instances
+    ########################################################
+
+    def create_instance(self, instance_key, instance_name, wait=True):
+        """
+        Creates a new instance of this app. Each instance. must have a globally unique
+        instance key, separate from any project key across the whole DSS instance
+
+        :return: 
+        """
+        future_resp = self.client._perform_json(
+            "POST", "/apps/%s/instances" % self.app_id, body={
+            "targetProjectKey" : instance_key,
+            "targetProjectName" : instance_name
+        })
+        future = DSSFuture(self.client, future_resp.get("jobId", None), future_resp)
+        if wait:
+            result = future.wait_for_result()
+            return DSSAppInstance(self.client, instance_key)
+        else:
+            return future
+
+    def make_random_project_key(self):
+        return "%s_tmp_%s" % (self.app_id, random_string(10))
+
+    def create_temporary_instance(self):
+        """
+        Creates a new temporary instance of this app.
+        The return value should be used as a Python context manager. Upon exit, the temporary app
+        instance is deleted
+        :return a :class:`TemporaryDSSAppInstance`
+        """
+        key = self.make_random_project_key()
+        instance = self.create_instance(key, key, True)
+        return TemporaryDSSAppInstance(self.client, key)
+
+    def list_instance_keys(self):
+        """
+        List the existing instances of this app
+
+        :return a list of instance keys, each as a string
+        """
+        return [x["projectKey"] for x in self.list_instances()]
+
+    def list_instances(self):
+        """
+        List the existing instances of this app
+        
+        :rtype: list of dicts
+        :return a list of instances, each as a dict containing at least a "projectKey" field
+        """
+        return self.client._perform_json(
+            "GET", "/apps/%s/instances/" % self.app_id)
+
+    def get_instance(self, instance_key):
+        return DSSAppInstance(self.client, instance_key)
+
+
+    def get_manifest(self):
+        raw_data = self.client._perform_json("GET", "/apps/%s/" % self.app_id)
+        return DSSAppManifest(self.client, raw_data)
+
+
+class DSSAppManifest(object):
+
+    def __init__(self, client, raw_data):
+        """The manifest for an app. Do not create this class directly"""
+        self.client = client
+        self.raw_data = raw_data
+
+    def get_raw(self):
+        return self.raw_data
+
+    def get_all_actions(self):
+        return [x for section in self.raw_data["homepageSections"] for x in section["tiles"]]
+
+    def get_runnable_scenarios(self):
+        """Return the scenario identifiers that are declared as actions for this app"""
+        return [x["scenarioId"] for x in self.get_all_actions() if x["type"] == "SCENARIO_RUN"]
+
+class DSSAppInstance(object):
+
+    def __init__(self, client, project_key):
+       self.client = client
+       self.project_key = project_key
+
+    def get_as_project(self):
+        """
+        Get the :class:`dataikuapi.dss.project DSSProject` corresponding to this app instance
+        """
+        return self.client.get_project(self.project_key)
+
+    def get_manifest(self):
+        """
+        Get the app manifest for this instance, as a :class:`DSSAppManifest`
+        """
+        raw_data = self.client._perform_json("GET", "/projects/%s/app-manifest" % self.project_key)
+        return DSSAppManifest(self.client, raw_data)
+
+class TemporaryDSSAppInstance(DSSAppInstance):
+    """internal class"""
+
+    def __init__(self, client, project_key):
+        DSSAppInstance.__init__(self, client,project_key)
+
+
+    def close(self):
+        self.get_as_project().delete()
+
+    def __enter__(self,):
+        return self
+
+    def __exit__(self, type, value, traceback):
+        self.close()

dataikuapi/dss/dataset.py

@@ -2,6 +2,7 @@
 from ..utils import DataikuUTF8CSVReader
 from ..utils import DataikuStreamedHttpUTF8CSVReader
 import json
+from .future import DSSFuture
 from .metrics import ComputedMetrics
 from .discussion import DSSObjectDiscussions
 from .statistics import DSSStatisticsWorksheet
@@ -153,6 +154,22 @@ def clear(self, partitions=None):
                 "DELETE", "/projects/%s/datasets/%s/data" % (self.project_key, self.dataset_name),
                 params={"partitions" : partitions})
 
+    def copy_to(self, target, sync_schema=True, write_mode="OVERWRITE"):
+        """
+        Copies the data of this dataset to another dataset
+
+        :param target Dataset: a :class:`dataikuapi.dss.dataset.DSSDataset` representing the target of this copy
+        :returns: a DSSFuture representing the operation
+        """
+        dqr = {
+             "targetProjectKey" : target.project_key,
+             "targetDatasetName": target.dataset_name,
+             "syncSchema": sync_schema,
+             "writeMode" : write_mode
+        }
+        future_resp = self.client._perform_json("POST", "/projects/%s/datasets/%s/actions/copyTo" % (self.project_key, self.dataset_name), body=dqr)
+        return DSSFuture(self.client, future_resp.get("jobId", None), future_resp)
+
     ########################################################
     # Dataset actions
     ########################################################
@@ -205,6 +222,23 @@ def run_checks(self, partition='', checks=None):
                     "POST" , "/projects/%s/datasets/%s/actions/runChecks" %(self.project_key, self.dataset_name),
                     params={'partition':partition}, body=checks)
 
+    def uploaded_add_file(self, fp, filename):
+        """
+        Adds a file to an "uploaded files" dataset
+
+        :param file fp: A file-like object that represents the file to upload
+        :param str filename: The filename for the file to upload 
+        """
+        self.client._perform_empty("POST", "/projects/%s/datasets/%s/uploaded/files" % (self.project_key, self.dataset_name),
+         files={"file":(filename, fp)})
+
+    def uploaded_list_files(self):
+        """
+        List the files in an "uploaded files" dataset
+        """
+        return self.client._perform_json("GET", "/projects/%s/datasets/%s/uploaded/files" % (self.project_key, self.dataset_name))
+
+
     ########################################################
     # Statistics worksheets
     ########################################################

dataikuapi/dss/flow.py

@@ -0,0 +1,125 @@
+from .utils import AnyLoc
+from .recipe import DSSRecipeDefinitionAndPayload
+from .future import DSSFuture
+import logging
+
+class DSSProjectFlow(object):
+    def __init__(self, client, project):
+        self.client = client
+        self.project = project
+
+
+    def replace_input_computable(self, current_ref, new_ref, type="DATASET"):
+        """
+        This method replaces all references to a "computable" (Dataset, Managed Folder or Saved Model)
+        as input of recipes in the whole Flow by a reference to another computable.
+
+        No specific checks are performed. It is your responsibility to ensure that the schema
+        of the new dataset will be compatible with the previous one (in the case of datasets).
+
+        If `new_ref` references an object in a foreign project, this method will automatically
+        ensure that `new_ref` is exposed to the current project
+
+        :param current_ref str: Either a "simple" object name (dataset name, model id, managed folder id)
+                    or a foreign object reference in the form "FOREIGN_PROJECT_KEY.local_id")
+        :param new_ref str: Either a "simple" object name (dataset name, model id, managed folder id)
+                    or a foreign object reference in the form "FOREIGN_PROJECT_KEY.local_id")
+        :param type str: The type of object being replaced (DATASET, SAVED_MODEL or MANAGED_FOLDER)
+        """
+
+        new_loc = AnyLoc.from_ref(self.project.project_key, new_ref)
+
+        if new_loc.project_key != self.project.project_key:
+            logging.info("New ref is in project %s, exposing it to project %s" % (new_loc.project_key, self.project.project_key))
+            new_ref_src_project = self.client.get_project(new_loc.project_key)
+            settings = new_ref_src_project.get_settings()
+            settings.add_exposed_object(type, new_loc.object_id, self.project.project_key)
+            settings.save()
+
+        for recipe in self.project.list_recipes():
+            fake_rap = DSSRecipeDefinitionAndPayload({"recipe" : recipe})
+            if fake_rap.has_input(current_ref):
+                logging.info("Recipe %s has %s as input, performing the replacement by %s"% \
+                    (recipe["name"], current_ref, new_ref))
+                recipe_obj = self.project.get_recipe(recipe["name"])
+                dap = recipe_obj.get_definition_and_payload()
+                dap.replace_input(current_ref, new_ref)
+                recipe_obj.set_definition_and_payload(dap)
+
+    ########################################################
+    # Flow tools
+    ########################################################
+
+    def start_tool(self, type, data={}):
+        """
+        Start a tool or open a view in the flow
+
+        :param type str: one of {COPY, CHECK_CONSISTENCY, PROPAGATE_SCHEMA} (tools) or {TAGS, CUSTOM_FIELDS, CONNECTIONS, COUNT_OF_RECORDS, FILESIZE, FILEFORMATS, RECIPES_ENGINES, RECIPES_CODE_ENVS, IMPALA_WRITE_MODE, HIVE_MODE, SPARK_ENGINE, SPARK_CONFIG, SPARK_PIPELINES, SQL_PIPELINES, PARTITIONING, PARTITIONS, SCENARIOS, CREATION, LAST_MODIFICATION, LAST_BUILD, RECENT_ACTIVITY, WATCH}  (views)
+        :param data dict: initial data for the tool (optional)
+    
+        :returns: a :class:`.flow.DSSFlowTool` handle to interact with the newly-created tool or view
+        """
+        tool_id = self.client._perform_text("POST", "/projects/%s/flow/tools/" % self.project.project_key, params={'type':type}, body=data)
+        return DSSFlowTool(self.client, self.project.project_key, tool_id)
+
+    def propagate_schema(self, dataset_name, rebuild=False, recipe_update_options={}, excluded_recipes=[], mark_as_ok_recipes=[], partition_by_dim={}, partition_by_computable={}):
+        """
+        Start an automatic schema propagate from a dataset
+
+        :param dataset_name str: name of a dataset to start propagating from
+        :param rebuild bool: whether to automatically rebuild datasets if needed
+        :param recipe_update_options str: pre-recipe or per-recipe-type update options to apply on recipes
+        :param excluded_recipes list: list of recipes where propagation is to stop
+        :param mark_as_ok_recipes list: list of recipes to consider as ok
+        :param partition_by_dim dict: partition value to use for each dimension
+        :param partition_by_computable dict: partition spec to use for a given dataset or recipe (overrides partition_by_dim)
+    
+        :returns: dict of the messages collected during the update
+        """
+        data = {'recipeUpdateOptions':recipe_update_options, 'partitionByDim':partition_by_dim, 'partitionByComputable':partition_by_computable, 'excludedRecipes':excluded_recipes, 'markAsOkRecipes':mark_as_ok_recipes}
+        update_future = self.client._perform_json("POST", "/projects/%s/flow/tools/propagate-schema/%s/" % (self.project.project_key, dataset_name), params={'rebuild':rebuild}, body=data)
+        return DSSFuture(self.client,update_future.get('jobId', None), update_future)
+
+
+class DSSFlowTool(object):
+    """
+    Handle to interact with a flow tool
+    """
+    def __init__(self, client, project_key, tool_id):
+        self.client = client
+        self.project_key = project_key
+        self.tool_id = tool_id
+
+    def stop(self):
+        """
+        Stops the tool and releases the resources held by it
+        """
+        return self.client._perform_json("POST", "/projects/%s/flow/tools/%s/stop" % (self.project_key, self.tool_id))
+
+    def get_state(self, options={}):
+        """
+        Get the current state of the tool or view
+
+        :returns: the state, as a dict
+        """
+        return self.client._perform_json("GET", "/projects/%s/flow/tools/%s/state" % (self.project_key, self.tool_id), body=options)
+ 
+    def do(self, action):
+        """
+        Perform a manual user action on the tool
+
+        :returns: the current state, as a dict
+        """
+        return self.client._perform_json("PUT", "/projects/%s/flow/tools/%s/action" % (self.project_key, self.tool_id), body=action)
+ 
+    def update(self, options={}):
+        """
+        (for tools only) Start updating the tool state
+
+        :params options dict: options for the update (optional)
+
+        :returns: a :class:`.future.DSSFuture` handle to interact with task of performing the update
+        """
+        update_future = self.client._perform_json("POST", "/projects/%s/flow/tools/%s/update" % (self.project_key, self.tool_id), body=options)
+        return DSSFuture(self.client,update_future.get('jobId', None), update_future)
+

dataikuapi/dss/managedfolder.py

@@ -3,6 +3,7 @@
 from ..utils import DataikuStreamedHttpUTF8CSVReader
 import json
 from .metrics import ComputedMetrics
+from .future import DSSFuture
 from .discussion import DSSObjectDiscussions
 
 class DSSManagedFolder(object):
@@ -173,3 +174,22 @@ def get_object_discussions(self):
         :rtype: :class:`dataikuapi.discussion.DSSObjectDiscussions`
         """
         return DSSObjectDiscussions(self.client, self.project_key, "MANAGED_FOLDER", self.odb_id)
+
+    ########################################################
+    # utilities
+    ########################################################
+    def copy_to(self, target, write_mode="OVERWRITE"):
+        """
+        Copies the data of this folder to another folder
+
+        :param target Folder: a :class:`dataikuapi.dss.managedfolder.DSSManagedFolder` representing the target of this copy
+        :returns: a DSSFuture representing the operation
+        """
+        dqr = {
+             "targetProjectKey" : target.project_key,
+             "targetFolderId": target.odb_id,
+             "writeMode" : write_mode
+        }
+        future_resp = self.client._perform_json("POST", "/projects/%s/managedfolders/%s/actions/copyTo" % (self.project_key, self.odb_id), body=dqr)
+        return DSSFuture(self.client, future_resp.get("jobId", None), future_resp)
+