Feature/add configurable retry

CHANGELOG.md

@@ -3,6 +3,7 @@
 ## [Version 1.4.3](https://github.com/dataiku/dss-plugin-api-connect/releases/tag/v1.4.3) - Bugfix - 2026-07-27
 
 - Fix templating for multiform body
+- Adding a configurable retry for several HTTP errors
 
 ## [Version 1.4.2](https://github.com/dataiku/dss-plugin-api-connect/releases/tag/v1.4.2) - Bugfix - 2026-07-22
 

custom-recipes/api-connect/recipe.json

@@ -408,6 +408,64 @@
             "description": "-1 for no limit",
             "type": "INT",
             "defaultValue": -1
+        },
+        {
+            "name": "http_errors_retry_strategy",
+            "label": "Retry on error logic",
+            "description": "",
+            "type": "SELECT",
+            "defaultValue": null,
+            "selectChoices":[
+                {"value": null, "label": "No retry"},
+                {"value": "linear", "label": "Linear backoff"},
+                {"value": "exponential", "label": "Exponential backoff"}
+            ]
+        },
+        {
+            "name": "http_errors_to_retry",
+            "label": "Errors to retry",
+            "description": "Click to select errors that can trigger a retry",
+            "type": "MULTISELECT",
+            "defaultValue": null,
+            "selectChoices":[
+                {"value": "408", "label": "408 Request Timeout"},
+                {"value": "429", "label": "429 Too many requests"},
+                {"value": "503", "label": "503 Service Unavailable"},
+                {"value": "504", "label": "504 Gateway Time out"}
+            ],
+            "visibilityCondition": "(['exponential', 'linear'].includes(model.http_errors_retry_strategy))"
+        },
+        {
+            "name": "http_errors_initial_delay",
+            "label": "Initial delay",
+            "description": "in seconds",
+            "type": "INT",
+            "defaultValue": 1,
+            "visibilityCondition": "(['exponential'].includes(model.http_errors_retry_strategy))"
+        },
+        {
+            "name": "http_errors_maximum_delay",
+            "label": "Maximum delay",
+            "description": "in seconds",
+            "type": "INT",
+            "defaultValue": 120,
+            "visibilityCondition": "(['exponential'].includes(model.http_errors_retry_strategy))"
+        },
+        {
+            "name": "http_errors_delay",
+            "label": "Delay",
+            "description": "in seconds",
+            "type": "INT",
+            "defaultValue": 1,
+            "visibilityCondition": "(['linear'].includes(model.http_errors_retry_strategy))"
+        },
+        {
+            "name": "http_errors_maximum_retries",
+            "label": "Maximum number of retries",
+            "description": "in seconds",
+            "type": "INT",
+            "defaultValue": 5,
+            "visibilityCondition": "(['linear'].includes(model.http_errors_retry_strategy))"
         }
     ],
     "resourceKeys": []

custom-recipes/api-connect/recipe.py

@@ -3,9 +3,10 @@
 from dataiku.customrecipe import get_input_names_for_role, get_recipe_config, get_output_names_for_role
 import pandas as pd
 from safe_logger import SafeLogger
-from dku_utils import get_dku_key_values, get_endpoint_parameters, get_secure_credentials, get_user_secrets
+from dku_utils import get_dku_key_values, get_endpoint_parameters, get_secure_credentials, get_user_secrets, get_retry_handler_parameters_from_config
 from rest_api_recipe_session import RestApiRecipeSession
 from dku_constants import DKUConstants
+from retry_handler import RetryHandler
 
 
 logger = SafeLogger("api-connect plugin", forbidden_keys=DKUConstants.FORBIDDEN_KEYS)
@@ -53,6 +54,13 @@ def get_partitioning_keys(id_list, dku_flow_variables):
 partitioning_keys = get_partitioning_keys(input_parameters_dataset, dku_flow_variables)
 custom_key_values.update(partitioning_keys)
 input_parameters_dataframe = input_parameters_dataset.get_dataframe(infer_with_pandas=False)
+backoff_type, initial_delay, maximum_number_of_retries, maximum_duration_of_retry, status_codes_to_retry = get_retry_handler_parameters_from_config(config)
+retry_handler = None
+if backoff_type:
+    retry_handler = RetryHandler(
+        backoff_type=backoff_type, initial_delay=initial_delay, maximum_number_of_retries=maximum_number_of_retries,
+        maximum_duration_of_retry=maximum_duration_of_retry, status_codes_to_retry=status_codes_to_retry
+    )
 
 recipe_session = RestApiRecipeSession(
     custom_key_values,
@@ -64,7 +72,8 @@ def get_partitioning_keys(id_list, dku_flow_variables):
     parameter_renamings,
     display_metadata,
     maximum_number_rows=maximum_number_rows,
-    behaviour_when_error=behaviour_when_error
+    behaviour_when_error=behaviour_when_error,
+    retry_handler=retry_handler
 )
 results = recipe_session.process_dataframe(input_parameters_dataframe, is_raw_output)
 

python-connectors/api-connect_dataset/connector.json

@@ -356,6 +356,64 @@
             "description": "-1 for no limit",
             "type": "INT",
             "defaultValue": -1
+        },
+        {
+            "name": "http_errors_retry_strategy",
+            "label": "Retry on error logic",
+            "description": "",
+            "type": "SELECT",
+            "defaultValue": null,
+            "selectChoices":[
+                {"value": null, "label": "No retry"},
+                {"value": "linear", "label": "Linear backoff"},
+                {"value": "exponential", "label": "Exponential backoff"}
+            ]
+        },
+        {
+            "name": "http_errors_to_retry",
+            "label": "Errors to retry",
+            "description": "Click to select errors that can trigger a retry",
+            "type": "MULTISELECT",
+            "defaultValue": null,
+            "selectChoices":[
+                {"value": "408", "label": "408 Request Timeout"},
+                {"value": "429", "label": "429 Too many requests"},
+                {"value": "503", "label": "503 Service Unavailable"},
+                {"value": "504", "label": "504 Gateway Time out"}
+            ],
+            "visibilityCondition": "(['exponential', 'linear'].includes(model.http_errors_retry_strategy))"
+        },
+        {
+            "name": "http_errors_initial_delay",
+            "label": "Initial delay",
+            "description": "in seconds",
+            "type": "INT",
+            "defaultValue": 1,
+            "visibilityCondition": "(['exponential'].includes(model.http_errors_retry_strategy))"
+        },
+        {
+            "name": "http_errors_maximum_delay",
+            "label": "Maximum delay",
+            "description": "in seconds",
+            "type": "INT",
+            "defaultValue": 120,
+            "visibilityCondition": "(['exponential'].includes(model.http_errors_retry_strategy))"
+        },
+        {
+            "name": "http_errors_delay",
+            "label": "Delay",
+            "description": "in seconds",
+            "type": "INT",
+            "defaultValue": 1,
+            "visibilityCondition": "(['linear'].includes(model.http_errors_retry_strategy))"
+        },
+        {
+            "name": "http_errors_maximum_retries",
+            "label": "Maximum number of retries",
+            "description": "in seconds",
+            "type": "INT",
+            "defaultValue": 5,
+            "visibilityCondition": "(['linear'].includes(model.http_errors_retry_strategy))"
         }
     ]
 }

python-connectors/api-connect_dataset/connector.py

@@ -5,10 +5,11 @@
 from dku_utils import (
     get_dku_key_values, get_endpoint_parameters,
     parse_keys_for_json, get_value_from_path, get_secure_credentials,
-    decode_csv_data, decode_bytes, get_user_secrets
+    decode_csv_data, decode_bytes, get_user_secrets, get_retry_handler_parameters_from_config
 )
 from dku_constants import DKUConstants
 import json
+from retry_handler import RetryHandler
 
 
 logger = SafeLogger("api-connect plugin", forbidden_keys=DKUConstants.FORBIDDEN_KEYS)
@@ -26,7 +27,14 @@ def __init__(self, config, plugin_config):
         custom_key_values = get_dku_key_values(config.get("custom_key_values", {}))
         user_secrets = get_user_secrets(config)
         custom_key_values.update(user_secrets)
-        self.client = RestAPIClient(credential, secure_credentials, endpoint_parameters, custom_key_values)
+        backoff_type, initial_delay, maximum_number_of_retries, maximum_duration_of_retry, status_codes_to_retry = get_retry_handler_parameters_from_config(config)
+        retry_handler = None
+        if backoff_type:
+            retry_handler = RetryHandler(
+                backoff_type=backoff_type, initial_delay=initial_delay, maximum_number_of_retries=maximum_number_of_retries,
+                maximum_duration_of_retry=maximum_duration_of_retry, status_codes_to_retry=status_codes_to_retry
+            )
+        self.client = RestAPIClient(credential, secure_credentials, endpoint_parameters, custom_key_values, retry_handler=retry_handler)
         extraction_key = endpoint_parameters.get("extraction_key", None)
         self.extraction_key = extraction_key or ''
         self.extraction_path = self.extraction_key.split('.')

python-lib/dku_utils.py

@@ -319,3 +319,21 @@ def join_url(base_url, segment):
         segment = segment.lstrip("/")
         segments.append(segment)
     return "/".join(segments)
+
+
+def get_retry_handler_parameters_from_config(config):
+    backoff_type = initial_delay = maximum_number_of_retries = maximum_duration_of_retry = status_codes_to_retry = None
+    http_errors_retry_strategy = config.get("http_errors_retry_strategy", None)
+    if not http_errors_retry_strategy:
+        return backoff_type, initial_delay, maximum_number_of_retries, maximum_duration_of_retry, status_codes_to_retry
+    if http_errors_retry_strategy in ["linear", "exponential"]:
+        backoff_type = http_errors_retry_strategy
+    if backoff_type == "linear":
+        initial_delay = config.get("http_errors_delay")
+        maximum_number_of_retries = config.get("http_errors_maximum_retries", None)
+    if backoff_type == "exponential":
+        initial_delay = config.get("http_errors_initial_delay")
+        maximum_duration_of_retry = config.get("http_errors_maximum_delay", None)
+    if backoff_type:
+        status_codes_to_retry = config.get("http_errors_to_retry", [])
+    return backoff_type, initial_delay, maximum_number_of_retries, maximum_duration_of_retry, status_codes_to_retry

python-lib/rest_api_client.py

@@ -8,6 +8,7 @@
 from dku_utils import get_dku_key_values, get_dku_duplicated_key_values, template_dict, format_template, is_reponse_xml, xml_to_json
 from dku_constants import DKUConstants
 from rest_api_auth import get_auth
+from retry_handler import DefaultRetryHandler
 
 
 logger = SafeLogger("api-connect plugin", forbidden_keys=DKUConstants.FORBIDDEN_KEYS)
@@ -19,7 +20,7 @@ class RestAPIClientError(ValueError):
 
 class RestAPIClient(object):
 
-    def __init__(self, credential, secure_credentials, endpoint, custom_key_values={}, session=None, behaviour_when_error=None):
+    def __init__(self, credential, secure_credentials, endpoint, custom_key_values={}, session=None, behaviour_when_error=None, retry_handler=None):
         logger.info("Initialising RestAPIClient, credential={}, secure_credentials={}, endpoint={}".format(
             logger.filter_secrets(credential),
             logger.filter_secrets(secure_credentials),
@@ -134,6 +135,7 @@ def __init__(self, credential, secure_credentials, endpoint, custom_key_values={
                 self.secure_domain = "https://{}".format(self.secure_domain)
         else:
             self.session.auth = get_auth(credential)
+        self.retry_handler = retry_handler or DefaultRetryHandler()
 
     def get(self, url, can_raise_exeption=True, **kwargs):
         json_response = self.request("GET", url, can_raise_exeption=can_raise_exeption, **kwargs)
@@ -216,9 +218,15 @@ def request_with_cert(self, method, url, **kwargs):
                         )
                         tmp_key.seek(0)
                         kwargs["cert"] = (tmp_certificate.name, tmp_key.name)
-                        response = self.session.request(method, url, **kwargs)
+                        response = self.request_with_errors_retry(method, url, **kwargs)
                 return response
-        return self.session.request(method, url, **kwargs)
+        return self.request_with_errors_retry(method, url, **kwargs)
+
+    def request_with_errors_retry(self, method, url, **kwargs):
+        response = None
+        while self.retry_handler.should_retry(response):
+            response = self.session.request(method, url, **kwargs)
+        return response
 
     def paginated_api_call(self, can_raise_exeption=True):
         if self.pagination.params_must_be_blanked:

python-lib/rest_api_recipe_session.py

@@ -15,7 +15,7 @@
 class RestApiRecipeSession:
     def __init__(self, custom_key_values, credential_parameters, secure_credentials, endpoint_parameters, extraction_key, parameter_columns, parameter_renamings,
                  display_metadata=False,
-                 maximum_number_rows=-1, behaviour_when_error=None):
+                 maximum_number_rows=-1, behaviour_when_error=None, retry_handler=None):
         self.custom_key_values = custom_key_values
         self.credential_parameters = credential_parameters
         self.secure_credentials = secure_credentials
@@ -30,6 +30,7 @@ def __init__(self, custom_key_values, credential_parameters, secure_credentials,
         self.behaviour_when_error = behaviour_when_error or "add-error-column"
         self.can_raise = self.behaviour_when_error == "raise"
         self.csv_configuration = endpoint_parameters
+        self.retry_handler = retry_handler
 
     @staticmethod
     def get_column_to_parameter_dict(parameter_columns, parameter_renamings):
@@ -68,7 +69,8 @@ def process_dataframe(self, input_parameters_dataframe, is_raw_output):
                 updated_endpoint_parameters,
                 custom_key_values=self.custom_key_values,
                 session=session,
-                behaviour_when_error=self.behaviour_when_error
+                behaviour_when_error=self.behaviour_when_error,
+                retry_handler=self.retry_handler
             )
             self.client.time_last_request = time_last_request
             while self.client.has_more_data():

python-lib/retry_handler.py

@@ -0,0 +1,98 @@
+import requests
+import time
+from safe_logger import SafeLogger
+
+
+logger = SafeLogger("api-connect plugin retry handler")
+
+
+class DefaultRetryHandler():
+    def __init__(self):
+        pass
+
+    def should_retry(self, response):
+        if response is None:
+            return True
+        return False
+
+
+class RetryHandler():
+    def __init__(self, backoff_type=None, initial_delay=None, maximum_number_of_retries=None,
+                 maximum_duration_of_retry=None, status_codes_to_retry=None):
+        self.backoff_type = None
+        if backoff_type in ["linear", "exponential"]:
+            self.backoff_type = backoff_type
+        self.initial_delay = 0
+        if isinstance(initial_delay, int):
+            self.initial_delay = initial_delay
+        self.maximum_number_of_retries = None
+        if isinstance(maximum_number_of_retries, int):
+            self.maximum_number_of_retries = maximum_number_of_retries
+        self.maximum_duration_of_retry = None
+        if isinstance(maximum_duration_of_retry, int):
+            self.maximum_duration_of_retry = maximum_duration_of_retry
+        self.next_delay = None
+        self.status_codes_to_retry = []
+        if isinstance(status_codes_to_retry, list):
+            self.status_codes_to_retry = status_codes_to_retry
+        self.number_of_tries = 0
+        logger.info("Retry handler initialised with {}/{}/{}/{}/{}/".format(
+            self.backoff_type,
+            self.initial_delay,
+            self.maximum_number_of_retries,
+            self.maximum_duration_of_retry,
+            self.status_codes_to_retry
+        ))
+
+    def should_retry(self, response):
+        logger.debug("Should retry?")
+        if response is None:
+            return True
+        if isinstance(response, requests.Response):
+            logger.info("is response")
+            status_code = str(response.status_code)
+            logger.info("status_code={}".format(status_code))
+            if status_code in self.status_codes_to_retry:
+                logger.warning("HTTP error {}. Retrying.".format(status_code))
+                self._compute_next_delay()
+                if self._is_next_delay_too_long():
+                    logger.info("_is_next_delay_too_long: should not.")
+                    return False
+                if self._too_many_retries():
+                    logger.info("_too_many_retries: should not.")
+                    return False
+                self._sleep()
+                return True
+        return False
+
+    def _compute_next_delay(self):
+        self.number_of_tries += 1
+        if self.next_delay is None:
+            self.next_delay = self.initial_delay
+            return
+        if self.backoff_type=="linear":
+            # delay is same as last try
+            return
+        if self.backoff_type=="exponential":
+            self.next_delay = self.next_delay * 2
+
+    def _sleep(self):
+        if isinstance(self.next_delay, int):
+            logger.warning("Sleeping for {}s".format(self.next_delay))
+            time.sleep(self.next_delay)
+
+    def _too_many_retries(self):
+        if not self.maximum_number_of_retries:
+            return False
+        if self.number_of_tries > self.maximum_number_of_retries:
+            logger.warning("Maximum number of retries reached. Not retrying.")
+            return True
+        return False
+
+    def _is_next_delay_too_long(self):
+        if not self.maximum_duration_of_retry:
+            return False
+        if self.next_delay >= self.maximum_duration_of_retry:
+            logger.warning("Sleep time before retry reached the max. Not retrying.")
+            return True
+        return False