Merge pull request #37 from dataiku/bug/sc-133490-stable-output-schema

Bug/sc 133490 stable output schema

Author: alexbourret

CHANGELOG.md

@@ -4,13 +4,15 @@
 
 - Add Brotli compression
 - Faster recurring calls
+- dku_error column kept at all time in API-Connect recipe output schema
 - Updated code-env descriptor for DSS 12
 
 
 ## [Version 1.1.4](https://github.com/dataiku/dss-plugin-api-connect/releases/tag/v1.1.4) - Feature and bugfix release - 2023-02-28
 
 - Add Brotli compression
 - Faster recurring calls
+- dku_error column kept at all time in API-Connect recipe output schema
 
 ## [Version 1.1.3](https://github.com/dataiku/dss-plugin-api-connect/releases/tag/v1.1.3) - Bugfix release - 2023-04-18
 

custom-recipes/api-connect/recipe.json

@@ -237,6 +237,19 @@
             "type": "SEPARATOR",
             "label": "Advanced"
         },
+        {
+            "name": "behaviour_when_error",
+            "label": "Error behaviour",
+            "description": "Decide how the recipe should react when an input line results in an error",
+            "type": "SELECT",
+            "defaultValue": "keep-error-column",
+            "selectChoices":[
+                {"value": "ignore", "label": "Ignore the error"},
+                {"value": "add-error-column", "label": "Add an error column"},
+                {"value": "keep-error-column", "label": "Error column always in schema"},
+                {"value": "raise", "label": "Fail the job"}
+            ]
+        },
         {
             "name": "ignore_ssl_check",
             "label": "Ignore SSL check",

custom-recipes/api-connect/recipe.py

@@ -35,6 +35,7 @@ def get_partitioning_keys(id_list, dku_flow_variables):
 logger.info("config={}".format(logger.filter_secrets(config)))
 
 credential_parameters = config.get("credential", {})
+behaviour_when_error = config.get("behaviour_when_error", "add-error-column")
 endpoint_parameters = get_endpoint_parameters(config)
 extraction_key = endpoint_parameters.get("extraction_key", "")
 is_raw_output = endpoint_parameters.get("raw_output", True)
@@ -58,7 +59,8 @@ def get_partitioning_keys(id_list, dku_flow_variables):
     parameter_columns,
     parameter_renamings,
     display_metadata,
-    maximum_number_rows=maximum_number_rows
+    maximum_number_rows=maximum_number_rows,
+    behaviour_when_error=behaviour_when_error
 )
 results = recipe_session.process_dataframe(input_parameters_dataframe, is_raw_output)
 

python-lib/dku_constants.py

@@ -4,3 +4,4 @@ class DKUConstants(object):
     FORM_DATA_BODY_FORMAT = "FORM_DATA"
     PLUGIN_VERSION = "1.2.0"
     RAW_BODY_FORMAT = "RAW"
+    REPONSE_ERROR_KEY = "dku_error"

python-lib/rest_api_client.py

@@ -17,7 +17,7 @@ class RestAPIClientError(ValueError):
 
 class RestAPIClient(object):
 
-    def __init__(self, credential, endpoint, custom_key_values={}, session=None):
+    def __init__(self, credential, endpoint, custom_key_values={}, session=None, behaviour_when_error=None):
         logger.info("Initialising RestAPIClient, credential={}, endpoint={}".format(logger.filter_secrets(credential), endpoint))
 
         #  presets_variables contains all variables available in templates using the {{variable_name}} notation
@@ -57,6 +57,7 @@ def __init__(self, credential, endpoint, custom_key_values={}, session=None):
         self.timeout = endpoint.get("timeout", -1)
         if self.timeout > 0:
             self.requests_kwargs.update({"timeout": self.timeout})
+        self.behaviour_when_error = behaviour_when_error or "add-error-column"
 
         self.requests_kwargs.update({"params": self.params})
         self.pagination = Pagination()
@@ -90,6 +91,8 @@ def __init__(self, credential, endpoint, custom_key_values={}, session=None):
             key_value_body = endpoint.get("key_value_body", {})
             self.requests_kwargs.update({"json": get_dku_key_values(key_value_body)})
         self.metadata = {}
+        if self.behaviour_when_error == "keep-error-column":
+            self.metadata = {DKUConstants.REPONSE_ERROR_KEY: None}
         self.call_number = 0
         self.session = session or requests.Session()
 
@@ -133,8 +136,12 @@ def request(self, method, url, can_raise_exeption=True, **kwargs):
         request_start_time = time.time()
         self.time_last_request = request_start_time
         error_message = None
+        status_code = None
+        response_headers = None
         try:
             response = self.request_with_redirect_retry(method, url, **kwargs)
+            status_code = response.status_code
+            response_headers = response.headers
         except Exception as err:
             self.pagination.is_last_batch_empty = True
             error_message = "Error: {}".format(err)
@@ -143,19 +150,19 @@ def request(self, method, url, can_raise_exeption=True, **kwargs):
 
         request_finish_time = time.time()
         self.set_metadata("request_duration", request_finish_time - request_start_time)
-        self.set_metadata("status_code", response.status_code)
-        self.set_metadata("response_headers", "{}".format(response.headers))
+        self.set_metadata("status_code", status_code)
+        self.set_metadata("response_headers", "{}".format(response_headers))
 
         if error_message:
-            return {"error": error_message}
+            return {} if self.behaviour_when_error=="ignore" else {DKUConstants.REPONSE_ERROR_KEY: error_message}
 
         if response.status_code >= 400:
             error_message = "Error {}: {}".format(response.status_code, response.content)
             self.pagination.is_last_batch_empty = True
             if can_raise_exeption:
                 raise RestAPIClientError(error_message)
             else:
-                return {"error": error_message}
+                return {} if self.behaviour_when_error=="ignore" else {DKUConstants.REPONSE_ERROR_KEY: error_message}
         if response.status_code in [204]:
             self.pagination.update_next_page({}, response.links)
             return self.empty_json_response()
@@ -168,7 +175,7 @@ def request(self, method, url, can_raise_exeption=True, **kwargs):
             logger.error("response.content={}".format(response.content))
             if can_raise_exeption:
                 raise RestAPIClientError("The API did not return JSON as expected. {}".format(error_message))
-            return {"error": error_message}
+            return {} if self.behaviour_when_error=="ignore" else {DKUConstants.REPONSE_ERROR_KEY: error_message}
 
         self.pagination.update_next_page(json_response, response.links)
         return json_response

python-lib/rest_api_recipe_session.py

@@ -14,7 +14,7 @@
 class RestApiRecipeSession:
     def __init__(self, custom_key_values, credential_parameters, endpoint_parameters, extraction_key, parameter_columns, parameter_renamings,
                  display_metadata=False,
-                 maximum_number_rows=-1):
+                 maximum_number_rows=-1, behaviour_when_error=None):
         self.custom_key_values = custom_key_values
         self.credential_parameters = credential_parameters
         self.endpoint_parameters = endpoint_parameters
@@ -25,7 +25,8 @@ def __init__(self, custom_key_values, credential_parameters, endpoint_parameters
         self.display_metadata = display_metadata
         self.maximum_number_rows = maximum_number_rows
         self.is_row_limit = (self.maximum_number_rows > 0)
-        self.can_raise = False
+        self.behaviour_when_error = behaviour_when_error or "add-error-column"
+        self.can_raise = self.behaviour_when_error == "raise"
 
     @staticmethod
     def get_column_to_parameter_dict(parameter_columns, parameter_renamings):
@@ -55,7 +56,13 @@ def process_dataframe(self, input_parameters_dataframe, is_raw_output):
                 updated_endpoint_parameters,
                 self.custom_key_values
             ))
-            self.client = RestAPIClient(self.credential_parameters, updated_endpoint_parameters, custom_key_values=self.custom_key_values, session=session)
+            self.client = RestAPIClient(
+                self.credential_parameters,
+                updated_endpoint_parameters,
+                custom_key_values=self.custom_key_values,
+                session=session,
+                behaviour_when_error=self.behaviour_when_error
+            )
             self.client.time_last_request = time_last_request
             while self.client.has_more_data():
                 page_results = self.retrieve_next_page(is_raw_output)
@@ -69,17 +76,24 @@ def process_dataframe(self, input_parameters_dataframe, is_raw_output):
     def retrieve_next_page(self, is_raw_output):
         page_rows = []
         logger.info("retrieve_next_page: Calling next page")
-        json_response = self.client.paginated_api_call(can_raise_exeption=False)
-        metadata = self.client.get_metadata() if self.display_metadata else {}
+        json_response = self.client.paginated_api_call(can_raise_exeption=self.can_raise)
+        default_dict = {
+            DKUConstants.REPONSE_ERROR_KEY: json_response.get(DKUConstants.REPONSE_ERROR_KEY, None)
+        } if self.behaviour_when_error == "keep-error-column" else {}
+        metadata = self.client.get_metadata() if self.display_metadata else default_dict
         is_api_returning_dict = True
         if self.extraction_key:
             data_rows = get_value_from_path(json_response, self.extraction_key.split("."), can_raise=False)
-            if data_rows is None:
+            if data_rows is None or type(data_rows) != list:
+                if self.behaviour_when_error == "ignore":
+                    return []
                 error_message = "Extraction key '{}' was not found in the incoming data".format(self.extraction_key)
                 if self.can_raise:
                     raise DataikuException(error_message)
+                elif DKUConstants.REPONSE_ERROR_KEY in metadata:
+                    return [metadata]
                 else:
-                    return self.format_page_rows([{"error": error_message}], is_raw_output, metadata)
+                    return self.format_page_rows([{DKUConstants.REPONSE_ERROR_KEY: error_message}], is_raw_output, metadata)
             page_rows.extend(self.format_page_rows(data_rows, is_raw_output, metadata))
         else:
             # Todo: check api_response key is free and add something overwise
@@ -115,6 +129,9 @@ def format_page_rows(self, data_rows, is_raw_output, metadata=None):
             base_row.update(metadata)
             if is_raw_output:
                 if is_error_message(data_row):
+                    base_row.update({
+                        DKUConstants.API_RESPONSE_KEY: None
+                    })
                     base_row.update(parse_keys_for_json(data_row))
                 else:
                     base_row.update({
@@ -129,7 +146,7 @@ def format_page_rows(self, data_rows, is_raw_output, metadata=None):
 def is_error_message(jsons_response):
     if type(jsons_response) not in [dict, list]:
         return False
-    if "error" in jsons_response and len(jsons_response) == 1:
+    if DKUConstants.REPONSE_ERROR_KEY in jsons_response and len(jsons_response) == 1:
         return True
     else:
         return False