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,76 @@
             "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_retry_scope",
+            "label": "Retry scope",
+            "description": "Apply the retry budget to the entire input dataset or independently to each input row",
+            "type": "SELECT",
+            "defaultValue": "dataset",
+            "selectChoices":[
+                {"value": "dataset", "label": "Per dataset"},
+                {"value": "row", "label": "Per row"}
+            ],
+            "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": "Number of times to retry a request after an error",
+            "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)
@@ -49,10 +50,18 @@ def get_partitioning_keys(id_list, dku_flow_variables):
 custom_key_values.update(user_secrets)
 display_metadata = config.get("display_metadata", False)
 maximum_number_rows = config.get("maximum_number_rows", -1)
+retry_scope = config.get("http_errors_retry_scope", "dataset")
 input_parameters_dataset = dataiku.Dataset(input_A_names[0])
 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 +73,9 @@ 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,
+    retry_scope=retry_scope
 )
 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": "Number of times to retry a request after an error",
+            "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, retry_scope="dataset"):
         self.custom_key_values = custom_key_values
         self.credential_parameters = credential_parameters
         self.secure_credentials = secure_credentials
@@ -30,6 +30,8 @@ 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
+        self.retry_scope = retry_scope
 
     @staticmethod
     def get_column_to_parameter_dict(parameter_columns, parameter_renamings):
@@ -46,6 +48,9 @@ def process_dataframe(self, input_parameters_dataframe, is_raw_output):
         time_last_request = None
         session = requests.Session()
         for index, input_parameters_row in input_parameters_dataframe.iterrows():
+            retry_handler = self.retry_handler
+            if self.retry_scope == "row" and retry_handler:
+                retry_handler = retry_handler.recreate()
             rows_count = 0
             self.initial_parameter_columns = {}
             for column_name in self.column_to_parameter_dict:
@@ -68,7 +73,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=retry_handler
             )
             self.client.time_last_request = time_last_request
             while self.client.has_more_data():