Update DF service functions for uploads, etc. (#48)

* Add private method to scrub user payload during call operation, refactored from main call() function
* Add private method to handle paging during main call() function
* Add private method to handle error CdpError parsing in main call() function
* Add new helper method to encode values for HTTP submission
* Add new helper method to expand file paths in local user OS
* Add private method to handle new alternative call() process which handles functions that the CDP Control Plane redirects to another URL to complete a file upload
* Move private method to handle standard call()s with unchanged functionality
* Simplify existing call() method to use these refactored methods to improve readability given new branching behavior requirements
* Modify DF import custom flow and custom flow version to use the new redirect call and cleanly upload files to the control plane
* Add Squelch for 409 error with uploads where the file already exists
* Prevent exceptions when errors are squelched.
* Add dataviz functions to the dw module.

Signed-off-by: Andre Araujo <[email protected]>
Signed-off-by: Daniel Chaffelson <[email protected]>
Co-authored-by: Daniel Chaffelson <[email protected]>

Author: asdaraujo

src/cdpy/common.py

@@ -16,6 +16,8 @@
 from urllib3.exceptions import InsecureRequestWarning
 from json import JSONDecodeError
 from typing import Union
+import urllib.parse as urlparse
+from os import path
 
 import cdpcli
 from cdpcli import VERSION as CDPCLI_VERSION
@@ -457,6 +459,20 @@ def _get_path(obj, path):
                 return None
         return value
 
+    @staticmethod
+    def encode_value(value):
+        if value:
+            return urlparse.quote(value)
+        return None
+
+    def expand_file_path(self, file_path):
+        if path.exists(file_path):
+            return path.expandvars(path.expanduser(file_path))
+        else:
+            self.throw_error(
+                CdpError('Path [{}] not found'.format(file_path))
+            )
+
     def wait_for_state(self, describe_func, params: dict, field: Union[str, None, list] = 'status',
                        state: Union[list, str, None] = None, delay: int = 15, timeout: int = 3600,
                        ignore_failures: bool = False):
@@ -538,8 +554,85 @@ def wait_for_state(self, describe_func, params: dict, field: Union[str, None, li
                 CdpError("Timeout waiting for function {0} with params [{1}] to return field {2} with state {3}"
                          .format(describe_func.__name__, str(params), field, str(state))))
 
+    def _scrub_inputs(self, inputs):
+        # Used in main call() function
+        logging.debug("Scrubbing inputs in payload")
+        # Remove unused submission values as the API rejects them
+        payload = {x: y for x, y in inputs.items() if y is not None}
+        # Remove and issue warning for empty string submission values as the API rejects them
+        _ = [self.throw_warning(
+            CdpWarning('Removing empty string arg %s from submission' % x))
+            for x, y in payload.items() if y == '']
+        payload = {x: y for x, y in payload.items() if y != ''}
+        return payload
+
+    def _handle_paging(self, response, call_function, payload):
+        # Used in main call() function
+        while 'nextToken' in response:
+            token = response.pop('nextToken')
+            next_page = call_function(
+                **payload, startingToken=token, pageSize=self.DEFAULT_PAGE_SIZE)
+            for key in next_page.keys():
+                if isinstance(next_page[key], str):
+                    response[key] = next_page[key]
+                elif isinstance(next_page[key], list):
+                    response[key] += (next_page[key])
+        return response
+
+    def _handle_call_errors(self, err, squelch):
+        # Used in main call() function
+        # Note that the cascade of behaviors here is designed to be convenient for Ansible module development
+        parsed_err = CdpError(err)
+        if self.debug:
+            log = self.get_log()
+            parsed_err.update(sdk_out=log, sdk_out_lines=log.splitlines())
+        if self.strict_errors is True:
+            self.throw_error(parsed_err)
+        if isinstance(err, ClientError):
+            if squelch is not None:
+                for item in squelch:
+                    if item.value in str(parsed_err.__getattribute__(item.field)):
+                        warning = item.warning if item.warning is not None else str(parsed_err.violations)
+                        self.throw_warning(CdpWarning(warning))
+                        return item.default
+        return parsed_err
+
+    def _handle_redirect_call(self, client, call_function, payload, headers):
+        # cdpcli/extensions/redirect.py
+        http, resp = client.make_api_call(
+            client.meta.method_to_api_mapping[call_function],
+            payload,
+            allow_redirects=False
+        )
+        if not http.is_redirect:
+            self.throw_error(CdpError("Redirect headers supplied but no redirect URL from API call"))
+        redirect_url = http.headers.get('Location', None)
+
+        if redirect_url is not None:
+            with open(self.expand_file_path(payload['file']), 'rb') as f:
+                http, full_response = client.make_request(
+                    operation_name=client.meta.method_to_api_mapping[call_function],
+                    method='post',
+                    url_path=redirect_url,
+                    headers=self._scrub_inputs(inputs=headers),
+                    body=f
+                )
+        else:
+            self.throw_error(CdpError("Redirect call attempted but redirect URL was empty"))
+        return full_response
+
+    def _handle_std_call(self, client, call_function, payload):
+        func_to_call = getattr(client, call_function)
+        raw_response = func_to_call(**payload)
+        if raw_response is not None and 'nextToken' in raw_response:
+            logging.debug("Found paged results in %s" % call_function)
+            full_response = self._handle_paging(raw_response, func_to_call, payload)
+        else:
+            full_response = raw_response
+        return full_response
+
     def call(self, svc: str, func: str, ret_field: str = None, squelch: ['Squelch'] = None, ret_error: bool = False,
-             **kwargs: Union[dict, bool, str, list]) -> Union[list, dict, 'CdpError']:
+             redirect_headers: dict = None, **kwargs: Union[dict, bool, str, list]) -> Union[list, dict, 'CdpError']:
         """
         Wraps the call to an underlying CDP CLI Service, handles common errors, and parses output
 
@@ -550,61 +643,34 @@ def call(self, svc: str, func: str, ret_field: str = None, squelch: ['Squelch']
             squelch (list(Squelch)): list of Descriptions of Error squelching options
             ret_error (bool): Whether to return the error object if generated,
                 defaults to False and raise instead
-            **kwargs (dict): Keyword Args to be supplied to the Function, eg userId
+            redirect_headers (dict): Dict of http submission headers for the call, triggers redirected upload call.
+            **kwargs (dict): Keyword Args to be supplied to the Function, e.g. userId
 
         Returns (dict, list, None): Output of CDP CLI Call
         """
         try:
-            call_function = getattr(self._client(service=svc, parameters=kwargs), func)
             if self.scrub_inputs:
-                # Remove unused submission values as the API rejects them
-                payload = {x: y for x, y in kwargs.items() if y is not None}
-                # Remove and issue warning for empty string submission values as the API rejects them
-                _ = [self.throw_warning(
-                    CdpWarning('Removing empty string arg %s from submission' % x))
-                    for x, y in payload.items() if y == '']
-                payload = {x: y for x, y in payload.items() if y != ''}
+                payload = self._scrub_inputs(inputs=kwargs)
             else:
                 payload = kwargs
 
-            resp = call_function(**payload)
+            svc_client = self._client(service=svc, parameters=payload)
 
-            if 'nextToken' in resp:
-                while 'nextToken' in resp:
-                    logging.debug("Found paged results in %s" % func)
-                    token = resp.pop('nextToken')
-                    next_page = call_function(
-                        **payload, startingToken=token, pageSize=self.DEFAULT_PAGE_SIZE)
-                    for key in next_page.keys():
-                        if isinstance(next_page[key], str):
-                            resp[key] = next_page[key]
-                        elif isinstance(next_page[key], list):
-                            resp[key] += (next_page[key])
+            if redirect_headers is not None:
+                full_response = self._handle_redirect_call(svc_client, func, payload, redirect_headers)
+            else:
+                full_response = self._handle_std_call(svc_client, func, payload)
 
             if ret_field is not None:
-                if not resp:
+                if not full_response:
                     self.throw_warning(CdpWarning('Call Response is empty, cannot return child field %s' % ret_field))
                 else:
-                    return resp[ret_field]
-
-            return resp
+                    return full_response[ret_field]
+            return full_response
 
         except Exception as err:
-            # Note that the cascade of behaviors here is designed to be convenient for Ansible module development
-            parsed_err = CdpError(err)
-            if self.debug:
-                log = self.get_log()
-                parsed_err.update(sdk_out=log, sdk_out_lines=log.splitlines())
-            if self.strict_errors is True:
-                self.throw_error(parsed_err)
-            if isinstance(err, ClientError):
-                if squelch is not None:
-                    for item in squelch:
-                        if item.value in str(parsed_err.__getattribute__(item.field)):
-                            warning = item.warning if item.warning is not None else str(parsed_err.violations)
-                            self.throw_warning(CdpWarning(warning))
-                            return item.default
-            if ret_error is True:
+            parsed_err = self._handle_call_errors(err, squelch)
+            if ret_error is True or not isinstance(parsed_err, CdpError):
                 return parsed_err
             self.throw_error(parsed_err)
 

src/cdpy/df.py

@@ -3,7 +3,8 @@
 from cdpy.common import CdpSdkBase, Squelch, CdpError, CdpWarning
 from cdpcli.extensions.df.createdeployment import CreateDeploymentOperationCaller
 
-ENTITLEMENT_DISABLED='DataFlow not enabled on CDP Tenant'
+ENTITLEMENT_DISABLED = 'DataFlow not enabled on CDP Tenant'
+
 
 class CdpyDf(CdpSdkBase):
     def __init__(self, *args, **kwargs):
@@ -156,7 +157,7 @@ def list_readyflows(self, name=None):
                 Squelch(value='NOT_FOUND',
                         warning='No ReadyFlows found within your CDP Tenant'),
                 Squelch(value='PATH_DISABLED',
-                        warning=ENTITLEMENT_DISABLED)                        
+                        warning=ENTITLEMENT_DISABLED)
             ],
         )
         if name is not None:
@@ -198,7 +199,7 @@ def import_readyflow(self, def_crn):
                 Squelch(value='NOT_FOUND',
                         warning='No ReadyFlow Definition with crn %s found' % def_crn),
                 Squelch(value='PATH_DISABLED',
-                        warning=ENTITLEMENT_DISABLED)                        
+                        warning=ENTITLEMENT_DISABLED)
             ],
             readyflowCrn=def_crn
         )
@@ -211,7 +212,7 @@ def delete_added_readyflow(self, def_crn):
                 Squelch(value='NOT_FOUND',
                         warning='No ReadyFlow Definition with crn %s found' % def_crn),
                 Squelch(value='PATH_DISABLED',
-                        warning=ENTITLEMENT_DISABLED)                        
+                        warning=ENTITLEMENT_DISABLED)
             ],
             readyflowCrn=def_crn
         )
@@ -224,7 +225,7 @@ def describe_added_readyflow(self, def_crn, sort_versions=True):
                 Squelch(value='NOT_FOUND',
                         warning='No ReadyFlow Definition with crn %s found' % def_crn),
                 Squelch(value='PATH_DISABLED',
-                        warning=ENTITLEMENT_DISABLED)                        
+                        warning=ENTITLEMENT_DISABLED)
             ],
             readyflowCrn=def_crn
         )
@@ -251,6 +252,54 @@ def describe_customflow(self, def_crn, sort_versions=True):
             out['versions'] = sorted(result['versions'], key=lambda d: d['version'], reverse=True)
         return out
 
+    def import_customflow(self, def_file, name, description=None, comments=None):
+        # cdpcli/extensions/df/__init__.py: DfExtension._df_upload_flow
+
+        return self.sdk.call(
+            svc='df', func='import_flow_definition', squelch=[
+                Squelch(value='PATH_DISABLED', warning=ENTITLEMENT_DISABLED),
+                Squelch(field='status_code', value='409')
+            ],
+            redirect_headers={
+                'Content-Type': 'application/json',
+                'Flow-Definition-Name': self.sdk.encode_value(name),
+                'Flow-Definition-Description': self.sdk.encode_value(description),
+                'Flow-Definition-Comments': self.sdk.encode_value(comments)
+            },
+            name=name,
+            file=def_file,
+            description=description,
+            comments=comments
+        )
+
+    def import_customflow_version(self, def_crn, def_file, comments=None):
+        self.sdk.validate_crn(def_crn, 'flow')
+        return self.sdk.call(
+            svc='df', func='import_flow_definition_version', ret_field='.', squelch=[
+                Squelch(value='PATH_DISABLED', warning=ENTITLEMENT_DISABLED),
+                Squelch(field='status_code', value='409')
+            ],
+            redirect_headers={
+                'Content-Type': 'application/json',
+                'Flow-Definition-Comments': self.sdk.encode_value(comments)
+            },
+            flowCrn=def_crn,
+            file=def_file,
+            comments=comments
+        )
+
+    def delete_customflow(self, def_crn):
+        self.sdk.validate_crn(def_crn, 'flow')
+        return self.sdk.call(
+            svc='df', func='delete_flow', ret_field='flow', squelch=[
+                Squelch(value='NOT_FOUND',
+                        warning='No Flow Definition with crn %s found' % def_crn),
+                Squelch(value='PATH_DISABLED',
+                        warning=ENTITLEMENT_DISABLED)
+            ],
+            flowCrn=def_crn
+        )
+
     def get_version_crn_from_flow_definition(self, flow_name, version=None):
         summary_list = self.list_flow_definitions(name=flow_name)
         if summary_list:

src/cdpy/dw.py

@@ -63,6 +63,18 @@ def describe_dbc(self, cluster_id, dbc_id):
             dbcId=dbc_id
         )
 
+    def describe_data_visualization(self, cluster_id, data_viz_id):
+        return self.sdk.call(
+            svc='dw', func='describe_data_visualization', ret_field='dataVisualization', squelch=[
+                Squelch('NOT_FOUND'),
+                Squelch('not found', field='violations'),
+                Squelch('INVALID_ARGUMENT'),
+                Squelch(value='PATH_DISABLED', warning=ENTITLEMENT_DISABLED)
+            ],
+            clusterId=cluster_id,
+            dataVisualizationId=data_viz_id,
+        )
+
     def list_clusters(self, env_crn=None):
         resp = self.sdk.call(
             svc='dw', func='list_clusters', ret_field='clusters', squelch=[
@@ -74,6 +86,15 @@ def list_clusters(self, env_crn=None):
             return [x for x in resp if env_crn == x['environmentCrn']]
         return resp
 
+    def list_data_visualizations(self, cluster_id):
+        return self.sdk.call(
+            svc='dw', func='list_data_visualizations', ret_field='dataVisualizations', squelch=[
+                Squelch(value='NOT_FOUND', default=list()),
+                Squelch(value='PATH_DISABLED', warning=ENTITLEMENT_DISABLED, default=list())
+            ],
+            clusterId=cluster_id
+        )
+
     def gather_clusters(self, env_crn=None):
         self.sdk.validate_crn(env_crn)
         clusters = self.list_clusters(env_crn=env_crn)
@@ -107,6 +128,17 @@ def create_cluster(self, env_crn: str, overlay: bool, aws_public_subnets: list =
             ]
         )
 
+    def create_data_visualization(self, cluster_id: str, name: str, config: dict = None):
+        return self.sdk.call(
+            svc='dw', func='create_data_visualization', ret_field='dataVisualizationId',
+            squelch=[
+                Squelch(value='PATH_DISABLED', warning=ENTITLEMENT_DISABLED)
+            ],
+            clusterId=cluster_id,
+            name=name,
+            config=config,
+        )
+
     def delete_cluster(self, cluster_id: str, force: bool = False):
         return self.sdk.call(
             svc='dw', func='delete_cluster', squelch=[
@@ -116,6 +148,27 @@ def delete_cluster(self, cluster_id: str, force: bool = False):
             clusterId=cluster_id, force=force
         )
 
+    def delete_data_visualization(self, cluster_id: str, data_viz_id: str):
+        return self.sdk.call(
+            svc='dw', func='delete_data_visualization', squelch=[
+                Squelch('NOT_FOUND'),
+                Squelch(value='PATH_DISABLED', warning=ENTITLEMENT_DISABLED)
+            ],
+            clusterId=cluster_id,
+            dataVisualizationId=data_viz_id,
+        )
+
+    def update_data_visualization(self, cluster_id: str, data_viz_id: str, config: dict):
+        return self.sdk.call(
+            svc='dw', func='update_data_visualization', squelch=[
+                Squelch('NOT_FOUND'),
+                Squelch(value='PATH_DISABLED', warning=ENTITLEMENT_DISABLED)
+            ],
+            clusterId=cluster_id,
+            dataVisualizationId=data_viz_id,
+            config=config,
+        )
+
     def create_vw(self, cluster_id:str, dbc_id:str, vw_type:str, name:str, template:str = None,
                   autoscaling_min_cluster:int = None, autoscaling_max_cluster:int = None,
                   common_configs:dict = None, application_configs:dict = None, ldap_groups:list = None,