Add exponential backoff for flaky MS Graph API (#17)

* Add exponential backoff for flaky MS Graph API

* bump version v26.10.3 -> v26.16.0

* Update changelog

* bump version v26.16.0 -> v26.16.1

Author: adammcdonagh

CHANGELOG.md

@@ -1,5 +1,10 @@
 # Changelog
 
+## v26.16.1
+
+- Add exponential backoff for MS Graph API calls
+- Add optional timeout to protocol definition
+
 ## v26.10.3
 
 - Fixed an issue where renaming when uploading a file did not actually rename the file

mypy.ini

@@ -19,7 +19,7 @@ check_untyped_defs = true
 disallow_incomplete_defs = true
 disallow_subclassing_any = false
 disallow_untyped_calls = true
-disallow_untyped_decorators = true
+disallow_untyped_decorators = false
 disallow_untyped_defs = true
 warn_return_any = true
 warn_unreachable = true

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "otf-addons-o365"
-version = "v26.10.3"
+version = "v26.16.1"
 authors = [{ name = "Adam McDonagh", email = "adam@elitemonkey.net" }]
 license = { text = "GPLv3" }
 classifiers = [
@@ -46,7 +46,7 @@ dev = [
 profile = 'black'
 
 [tool.bumpver]
-current_version = "v26.10.3"
+current_version = "v26.16.1"
 version_pattern = "vYY.WW.PATCH[-TAG]"
 commit_message = "bump version {old_version} -> {new_version}"
 commit = true

src/opentaskpy/addons/o365/remotehandlers/schemas/transfer/sharepoint_destination/protocol.json

@@ -9,6 +9,10 @@
         "opentaskpy.addons.o365.remotehandlers.sharepoint.SharepointTransfer"
       ]
     },
+    "timeout": {
+      "type": "integer",
+      "default": 30
+    },
     "refreshToken": {
       "type": "string"
     },

src/opentaskpy/addons/o365/remotehandlers/schemas/transfer/sharepoint_source/protocol.json

@@ -9,6 +9,10 @@
         "opentaskpy.addons.o365.remotehandlers.sharepoint.SharepointTransfer"
       ]
     },
+    "timeout": {
+      "type": "integer",
+      "default": 30
+    },
     "refreshToken": {
       "type": "string"
     },

src/opentaskpy/addons/o365/remotehandlers/sharepoint.py

@@ -3,16 +3,25 @@
 import glob
 import math
 import re
+import traceback
 from datetime import datetime
 from os import path
 from time import sleep
+from typing import Any
 
 import opentaskpy.otflogging
 import requests
 from dateutil.tz import tzlocal
 from opentaskpy.config.variablecaching import cache_utils
 from opentaskpy.exceptions import RemoteTransferError
 from opentaskpy.remotehandlers.remotehandler import RemoteTransferHandler
+from tenacity import (
+    RetryCallState,
+    retry,
+    retry_if_exception_type,
+    stop_after_attempt,
+    wait_exponential,
+)
 
 from .creds import get_access_token
 
@@ -24,6 +33,63 @@ class SharepointTransfer(RemoteTransferHandler):
 
     TASK_TYPE = "T"
 
+    @staticmethod
+    def _log_retry_attempt(retry_state: RetryCallState) -> None:
+        """Log details before tenacity sleeps and retries a request."""
+        if not retry_state.args:
+            return
+
+        self = retry_state.args[0]
+        method = str(retry_state.args[1]) if len(retry_state.args) > 1 else "UNKNOWN"
+        url = str(retry_state.args[2]) if len(retry_state.args) > 2 else "UNKNOWN"
+        sleep_for = (
+            retry_state.next_action.sleep
+            if retry_state.next_action is not None
+            else "unknown"
+        )
+        next_attempt = retry_state.attempt_number + 1
+        exception = retry_state.outcome.exception() if retry_state.outcome else None
+        exception_traceback = ""
+
+        if exception is not None:
+            exception_traceback = "".join(
+                traceback.format_exception(
+                    type(exception), exception, exception.__traceback__
+                )
+            )
+
+        self.logger.warning(
+            "Retrying SharePoint request %s %s after exception: %s "
+            "(failed attempt %s, retrying attempt %s, next sleep %ss)\n%s",
+            method,
+            url,
+            exception if exception is not None else "unknown",
+            retry_state.attempt_number,
+            next_attempt,
+            sleep_for,
+            exception_traceback,
+        )
+        self.logger.info(f"Sleeping for {sleep_for} seconds before retry")
+
+    @retry(
+        reraise=True,
+        stop=stop_after_attempt(6),
+        wait=wait_exponential(multiplier=2, min=5, max=60),
+        retry=retry_if_exception_type(requests.exceptions.ReadTimeout),
+        before_sleep=_log_retry_attempt,
+    )
+    def _request(self, method: str, url: str, **kwargs: Any) -> requests.Response:
+        """Perform a request with retry for transient timeout failures."""
+        method_upper = method.upper()
+        self.logger.debug(f"Making request to {url} with method {method}")
+        if method_upper == "GET":
+            return requests.get(url, **kwargs)  # pylint: disable=missing-timeout
+        if method_upper == "POST":
+            return requests.post(url, **kwargs)  # pylint: disable=missing-timeout
+        if method_upper == "PUT":
+            return requests.put(url, **kwargs)  # pylint: disable=missing-timeout
+        raise ValueError(f"Unsupported HTTP method for retry wrapper: {method}")
+
     def __init__(self, spec: dict):
         """Initialise the SharepointTransfer handler.
 
@@ -52,10 +118,14 @@ def __init__(self, spec: dict):
             "Authorization": "Bearer " + self.credentials["access_token"],
             "Content-Type": "application/json",
         }
-        response = requests.get(
+
+        self.timeout = self.spec["protocol"].get("timeout", 30)
+
+        response = self._request(
+            "GET",
             f"https://graph.microsoft.com/v1.0/sites/{self.spec['siteHostname']}:/sites/{self.spec['siteName']}",
             headers=self.headers,
-            timeout=5,
+            timeout=self.timeout,
         ).json()
 
         # Check the response is OK
@@ -117,13 +187,14 @@ def create_folder(self, parent_id: str | None, folder: str) -> str:
             create_folder_url = f"https://graph.microsoft.com/v1.0/sites/{self.site_id}/drive/root/children"
         else:
             create_folder_url = f"https://graph.microsoft.com/v1.0/sites/{self.site_id}/drive/items/{parent_id}/children"
-        response = requests.post(
+        response = self._request(
+            "POST",
             create_folder_url,
             headers={
                 "Authorization": "Bearer " + self.credentials["access_token"],
                 "Content-Type": "application/json",
             },
-            timeout=60,
+            timeout=self.timeout,
             json={"name": folder, "folder": {}},
         )
         if response.status_code != 201:
@@ -164,7 +235,7 @@ def handle_post_copy_action(self, files: dict) -> int:
                     headers={
                         "Authorization": "Bearer " + self.credentials["access_token"],
                     },
-                    timeout=60,
+                    timeout=self.timeout,
                 )
                 if response.status_code != 204:
                     self.logger.error(f"Failed to delete file: {file_name}")
@@ -216,7 +287,7 @@ def handle_post_copy_action(self, files: dict) -> int:
                 response = requests.patch(
                     file_url,
                     headers=patch_headers,
-                    timeout=60,
+                    timeout=self.timeout,
                     json=patch_body,
                 )
                 if response.status_code == 409:
@@ -239,7 +310,7 @@ def handle_post_copy_action(self, files: dict) -> int:
                                 "Bearer " + self.credentials["access_token"]
                             ),
                         },
-                        timeout=60,
+                        timeout=self.timeout,
                     )
                     # Check the response was a success
                     if response.status_code != 204:
@@ -253,7 +324,7 @@ def handle_post_copy_action(self, files: dict) -> int:
                     response = requests.patch(
                         file_url,
                         headers=patch_headers,
-                        timeout=60,
+                        timeout=self.timeout,
                         json=patch_body,
                     )
                 if response.status_code != 200:
@@ -287,12 +358,13 @@ def create_or_get_folder(self, destination_path: str) -> str | None:
             else:
                 # get_file_url_from_path returns a path-based URL, not a drive item ID;
                 # resolve it to the actual item ID so it can be used in parentReference.id
-                response = requests.get(
+                response = self._request(
+                    "GET",
                     folder_url,
                     headers={
                         "Authorization": "Bearer " + self.credentials["access_token"],
                     },
-                    timeout=60,
+                    timeout=self.timeout,
                 )
                 if response.status_code == 404:
                     self.logger.info(f"Folder {folder} does not exist, creating")
@@ -351,10 +423,11 @@ def list_files(
                     "Content-Type": "application/json",
                 }
 
-                response = requests.get(
+                response = self._request(
+                    "GET",
                     url,
                     headers=headers,
-                    timeout=30,
+                    timeout=self.timeout,
                 ).json()
 
                 if "value" in response and response["value"]:
@@ -476,7 +549,7 @@ def push_files_from_worker(
                             "Content-Type": "application/json",
                         },
                         data=f,
-                        timeout=60,
+                        timeout=self.timeout,
                     )
                     if response.status_code != 409:
                         break
@@ -535,13 +608,14 @@ def _do_upload_session(self, file: str, file_name: str) -> int:
             self.logger.info(f"File {file_name} already exists. Replacing file.")
             upload_session_url = f"{file_url}:/createUploadSession"
 
-        response = requests.post(
+        response = self._request(
+            "POST",
             upload_session_url,
             headers={
                 "Authorization": "Bearer " + self.credentials["access_token"],
                 "Content-Type": "application/json",
             },
-            timeout=60,
+            timeout=self.timeout,
         )
         if response.status_code != 200:
             self.logger.error(f"Failed to create upload session: {file_name}")
@@ -587,7 +661,8 @@ def _do_upload_session(self, file: str, file_name: str) -> int:
                 chunk = f.read(chunk_end - chunk_start + 1)
 
                 # PUT the chunk to the upload session url
-                response = requests.put(
+                response = self._request(
+                    "PUT",
                     upload_session_url,
                     data=chunk,
                     headers=headers,
@@ -642,12 +717,13 @@ def pull_files_to_worker(self, files: dict, local_staging_directory: str) -> int
                 # Download file using item url
                 self.logger.info(f"Downloading file: {file_name}")
                 download_url = f"{file_url}:/content"
-                response = requests.get(
+                response = self._request(
+                    "GET",
                     download_url,
                     headers={
                         "Authorization": "Bearer " + self.credentials["access_token"],
                     },
-                    timeout=60,
+                    timeout=self.timeout,
                 )
 
                 # Check the response was a success
@@ -705,12 +781,13 @@ def get_file_url_from_path(self, file_path: str) -> str | None:
 
                 # If the path starts with a / then it's a document library, we need to get the id of the document library
                 # Do a GET request to /sites/{siteId}/drives to get the document libraries
-                response = requests.get(
+                response = self._request(
+                    "GET",
                     f"https://graph.microsoft.com/v1.0/sites/{self.site_id}/drives",
                     headers={
                         "Authorization": "Bearer " + self.credentials["access_token"],
                     },
-                    timeout=60,
+                    timeout=self.timeout,
                 )
                 if response.status_code != 200:
                     self.logger.error("Failed to get document libraries")
@@ -737,12 +814,13 @@ def get_file_url_from_path(self, file_path: str) -> str | None:
 
             return f"https://graph.microsoft.com/v1.0/sites/{self.site_id}/drive/root:/{re.sub(r'/+', '/', file_path)}"
 
-        response = requests.get(
+        response = self._request(
+            "GET",
             item_url,
             headers={
                 "Authorization": "Bearer " + self.credentials["access_token"],
             },
-            timeout=60,
+            timeout=self.timeout,
         )
         # Check the response was a success
         if response.status_code != 200: