diff --git a/CHANGELOG.md b/CHANGELOG.md index 29b7560..7aac3bf 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/custom-recipes/api-connect/recipe.json b/custom-recipes/api-connect/recipe.json index 2a1a757..3d9d632 100644 --- a/custom-recipes/api-connect/recipe.json +++ b/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": [] diff --git a/custom-recipes/api-connect/recipe.py b/custom-recipes/api-connect/recipe.py index 54fd9ad..6ead838 100644 --- a/custom-recipes/api-connect/recipe.py +++ b/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) diff --git a/python-connectors/api-connect_dataset/connector.json b/python-connectors/api-connect_dataset/connector.json index db18a68..7fd2134 100644 --- a/python-connectors/api-connect_dataset/connector.json +++ b/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))" } ] } diff --git a/python-connectors/api-connect_dataset/connector.py b/python-connectors/api-connect_dataset/connector.py index baff370..3e88c26 100644 --- a/python-connectors/api-connect_dataset/connector.py +++ b/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('.') diff --git a/python-lib/dku_utils.py b/python-lib/dku_utils.py index 692300a..498238b 100644 --- a/python-lib/dku_utils.py +++ b/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 diff --git a/python-lib/rest_api_client.py b/python-lib/rest_api_client.py index d57e5d4..b789b6d 100644 --- a/python-lib/rest_api_client.py +++ b/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: diff --git a/python-lib/rest_api_recipe_session.py b/python-lib/rest_api_recipe_session.py index f8ef583..f4753da 100644 --- a/python-lib/rest_api_recipe_session.py +++ b/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(): diff --git a/python-lib/retry_handler.py b/python-lib/retry_handler.py new file mode 100644 index 0000000..de20b13 --- /dev/null +++ b/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 diff --git a/tests/python/integration/test_scenario.py b/tests/python/integration/test_scenario.py index 34d8bff..a042367 100644 --- a/tests/python/integration/test_scenario.py +++ b/tests/python/integration/test_scenario.py @@ -65,3 +65,7 @@ def test_run_api_connect_mtls(user_dss_clients): def test_run_api_connect_multipart_form_data(user_dss_clients): dss_scenario.run(user_dss_clients, project_key=TEST_PROJECT_KEY, scenario_id="MULTIPARTFORMDATA") + + +def test_run_api_connect_multipart_form_data(user_dss_clients): + dss_scenario.run(user_dss_clients, project_key=TEST_PROJECT_KEY, scenario_id="MULTIPARTFORMDATATEMPLATING")