Merge pull request #37 from dataiku/feature/sc-86553-429-error-on-multiple-files-uploads

Handling http 429, error on multiple files uploads

Author: alexbourret

python-lib/common.py

@@ -1,8 +1,14 @@
 import re
+import datetime
+import time
 try:
     import urlparse
 except:
     import urllib.parse as urlparse
+from safe_logger import SafeLogger
+from sharepoint_constants import SharePointConstants
+
+logger = SafeLogger("sharepoint-online plugin", ["Authorization", "sharepoint_username", "sharepoint_password", "client_secret"])
 
 
 def get_rel_path(path):
@@ -55,3 +61,50 @@ def parse_query_string_to_dict(query_string):
 def parse_url(tenant_name):
     url_tokens = urlparse.urlparse(tenant_name.strip('/'))
     return url_tokens.scheme, url_tokens.netloc, url_tokens.path
+
+
+def is_request_performed(response):
+    if response is None:
+        return False
+    if response.status_code in [429, 503]:
+        logger.warning("Error {}, headers = {}".format(response.status_code, response.headers))
+        seconds_before_retry = decode_retry_after_header(response)
+        logger.warning("Sleeping for {} seconds".format(seconds_before_retry))
+        time.sleep(seconds_before_retry)
+        return False
+    return True
+
+
+def decode_retry_after_header(response):
+    seconds_before_retry = SharePointConstants.DEFAULT_WAIT_BEFORE_RETRY
+    raw_header_value = response.headers.get("Retry-After", str(SharePointConstants.DEFAULT_WAIT_BEFORE_RETRY))
+    if raw_header_value.isdigit():
+        seconds_before_retry = int(raw_header_value)
+    else:
+        # Date format, "Wed, 21 Oct 2015 07:28:00 GMT"
+        try:
+            datetime_now = datetime.datetime.now()
+            datetime_header = datetime.datetime.strptime(raw_header_value, '%a, %d %b %Y %H:%M:%S GMT')
+            if datetime_header.timestamp() > datetime_now.timestamp():
+                # target date in the future
+                seconds_before_retry = (datetime_header - datetime_now).seconds
+        except Exception as err:
+            logger.error("decode_retry_after_header error {}".format(err))
+            seconds_before_retry = SharePointConstants.DEFAULT_WAIT_BEFORE_RETRY
+    return seconds_before_retry
+
+
+class ItemsLimit():
+    def __init__(self, records_limit=-1):
+        self.has_no_limit = (records_limit == -1)
+        self.records_limit = records_limit
+        self.counter = 0
+
+    def is_reached(self, number_of_new_records=None):
+        if self.has_no_limit:
+            return False
+        self.counter += number_of_new_records or 1
+        return self.counter > self.records_limit
+
+    def add_record(self):
+        self.counter += 1

python-lib/sharepoint_client.py

@@ -13,7 +13,7 @@
 from robust_session import RobustSession
 from sharepoint_constants import SharePointConstants
 from dss_constants import DSSConstants
-from common import is_email_address, get_value_from_path, parse_url, get_value_from_paths
+from common import is_email_address, get_value_from_path, parse_url, get_value_from_paths, is_request_performed, ItemsLimit
 from safe_logger import SafeLogger
 
 
@@ -820,12 +820,17 @@ def __init__(self, sharepoint_user_name, sharepoint_password, sharepoint_url, sh
         self.form_digest_value = self.get_form_digest_value()
 
     def get(self, url, headers=None, params=None):
+        retries_limit = ItemsLimit(SharePointConstants.MAX_RETRIES)
         headers = headers or {}
         headers["Accept"] = DSSConstants.APPLICATION_JSON
         headers["Authorization"] = self.get_authorization_bearer()
-        return requests.get(url, headers=headers, params=params)
+        response = None
+        while not is_request_performed(response) and not retries_limit.is_reached():
+            response = requests.get(url, headers=headers, params=params)
+        return response
 
     def post(self, url, headers=None, json=None, data=None, params=None):
+        retries_limit = ItemsLimit(SharePointConstants.MAX_RETRIES)
         headers = headers or {}
         default_headers = {
            "Accept": DSSConstants.APPLICATION_JSON_NOMETADATA,
@@ -835,7 +840,10 @@ def post(self, url, headers=None, json=None, data=None, params=None):
         if self.form_digest_value:
             default_headers.update({"X-RequestDigest": self.form_digest_value})
         default_headers.update(headers)
-        return requests.post(url, headers=default_headers, json=json, data=data, params=params, timeout=SharePointConstants.TIMEOUT_SEC)
+        response = None
+        while not is_request_performed(response) and not retries_limit.is_reached():
+            response = requests.post(url, headers=default_headers, json=json, data=data, params=params, timeout=SharePointConstants.TIMEOUT_SEC)
+        return response
 
     @staticmethod
     def close():

python-lib/sharepoint_constants.py

@@ -3,6 +3,7 @@ class SharePointConstants(object):
     COMMENT_COLUMN = 'comment'
     DATE_FORMAT = "%Y-%m-%d %H:%M:%S"
     DEFAULT_VIEW_ENDPOINT = "DefaultView/ViewFields"
+    DEFAULT_WAIT_BEFORE_RETRY = 60
     ENTITY_PROPERTY_NAME = 'EntityPropertyName'
     ERROR_CONTAINER = 'error'
     EXPENDABLES_FIELDS = {"Author": "Title", "Editor": "Title"}

tests/python/unit/test_common.py

@@ -1,7 +1,14 @@
-from common import get_value_from_path
+from common import get_value_from_path, is_request_performed, decode_retry_after_header
+from sharepoint_constants import SharePointConstants
 import pytest
 
 
+class MockResponse:
+    def __init__(self, status_code, headers):
+        self.status_code = status_code
+        self.headers = headers
+
+
 class TestCommonMethods:
     def setup_class(self):
         self.dictionary_to_search = {
@@ -15,6 +22,14 @@ def setup_class(self):
         self.ok_path_1 = ["a", "b", "c"]
         self.ok_path_2 = ["a", "d"]
         self.ko_path = ["a", "c"]
+        self.mock_response_none = None
+        self.mock_response_http_200 = MockResponse(200, {"Retry-After": "1"})
+        self.mock_response_http_429_digit_1s = MockResponse(429, {"Retry-After": "1"})
+        self.mock_response_http_429_no_header = MockResponse(429, {})
+        self.mock_response_http_503_digit_1s = MockResponse(503, {"Retry-After": "1"})
+        self.mock_response_http_429_date_in_past = MockResponse(429, {"Retry-After": "Wed, 21 Oct 2015 07:28:00 GMT"})
+        self.mock_response_http_429_date_in_future = MockResponse(429, {"Retry-After": "Wed, 21 Oct 9999 07:28:00 GMT"})
+        self.mock_response_http_429_garbage = MockResponse(429, {"Retry-After": "blablablabla"})
 
     def test_get_value_from_path_long_path(self):
         key = get_value_from_path(self.dictionary_to_search, self.ok_path_1)
@@ -31,3 +46,40 @@ def test_get_value_from_path_wrong_path(self):
     def test_get_value_from_path_wrong_path_custom_reply(self):
         key = get_value_from_path(self.dictionary_to_search, self.ko_path, default_reply="ko")
         assert key == "ko"
+
+    def test_is_request_performed_none(self):
+        mock_response = None
+        response = is_request_performed(mock_response)
+        assert response is False
+
+    def test_is_request_performed_error_200(self):
+        response = is_request_performed(self.mock_response_http_200)
+        assert response is True
+
+    def test_is_request_performed_error_429(self):
+        response = is_request_performed(self.mock_response_http_429_digit_1s)
+        assert response is False
+
+    def test_is_request_performed_error_503(self):
+        response = is_request_performed(self.mock_response_http_503_digit_1s)
+        assert response is False
+
+    def test_decode_retry_after_header_seconds(self):
+        seconds_before_retry = decode_retry_after_header(self.mock_response_http_429_digit_1s)
+        assert seconds_before_retry == 1
+
+    def test_decode_retry_after_header_future_date(self):
+        seconds_before_retry = decode_retry_after_header(self.mock_response_http_429_date_in_future)
+        assert seconds_before_retry >= 4000
+
+    def test_decode_retry_after_header_past_date(self):
+        seconds_before_retry = decode_retry_after_header(self.mock_response_http_429_date_in_past)
+        assert seconds_before_retry == SharePointConstants.DEFAULT_WAIT_BEFORE_RETRY
+
+    def test_decode_retry_after_header_garbage(self):
+        seconds_before_retry = decode_retry_after_header(self.mock_response_http_429_garbage)
+        assert seconds_before_retry == SharePointConstants.DEFAULT_WAIT_BEFORE_RETRY
+
+    def test_decode_retry_after_header_no_header(self):
+        seconds_before_retry = decode_retry_after_header(self.mock_response_http_429_no_header)
+        assert seconds_before_retry == SharePointConstants.DEFAULT_WAIT_BEFORE_RETRY