seamapi · andrii-balitskyi · Jul 15, 2024 · Jul 3, 2024 · Jul 3, 2024 · Jul 3, 2024
@@ -1,7 +1,8 @@
-from typing import Dict
+from typing import Dict, Optional
 from urllib.parse import urljoin
 import niquests as requests
 from importlib.metadata import version
+from urllib3.util import Retry
 
 from .constants import LTS_VERSION
 from .exceptions import (
@@ -17,16 +18,27 @@
     "seam-lts-version": LTS_VERSION,
 }
 
+DEFAULT_RETRIES = Retry()
+
 
 class SeamHttpClient(requests.Session, AbstractSeamHttpClient):
-    def __init__(self, base_url: str, auth_headers: Dict[str, str], **kwargs):
+    def __init__(
+        self,
+        base_url: str,
+        auth_headers: Dict[str, str],
+        retries: Optional[Retry] = DEFAULT_RETRIES,
+        **kwargs
+    ):
         super().__init__(**kwargs)
 
         self.base_url = base_url
-        headers = {**auth_headers, **kwargs.get("headers", {}), **SDK_HEADERS}
 
+        headers = {**auth_headers, **kwargs.get("headers", {}), **SDK_HEADERS}
         self.headers.update(headers)
 
+        if retries:
+            self.retries = retries
+
     def request(self, method, url, *args, **kwargs):
         url = urljoin(self.base_url, url)
         response = super().request(method, url, *args, **kwargs)

@@ -1,5 +1,6 @@
 from typing import Any, Optional, Union, Dict
 from typing_extensions import Self
+from urllib3.util.retry import Retry
 
 from .constants import LTS_VERSION
 from .parse_options import parse_options
@@ -24,6 +25,7 @@ def __init__(
         workspace_id: Optional[str] = None,
         endpoint: Optional[str] = None,
         wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]] = True,
+        retries: Optional[Retry] = None,
     ):
         """
         Parameters
@@ -38,6 +40,8 @@ def __init__(
           The API endpoint to which the request should be sent.
         wait_for_action_attempt : bool or dict, optional
           Controls whether to wait for an action attempt to complete, either as a boolean or as a dictionary specifying `timeout` and `poll_interval`. Defaults to `False`.
+        retries : urllib3.util.Retry, optional
+          Configuration for retry behavior on failed requests.
         """
 
         self.lts_version = Seam.lts_version
@@ -50,7 +54,9 @@ def __init__(
         )
         defaults = {"wait_for_action_attempt": wait_for_action_attempt}
 
-        self.client = SeamHttpClient(base_url=endpoint, auth_headers=auth_headers)
+        self.client = SeamHttpClient(
+            base_url=endpoint, auth_headers=auth_headers, retries=retries
+        )
 
         Routes.__init__(self, client=self.client, defaults=defaults)
 
@@ -61,9 +67,13 @@ def from_api_key(
         *,
         endpoint: Optional[str] = None,
         wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]] = True,
+        retries: Optional[Retry] = None,
     ) -> Self:
         return cls(
-            api_key, endpoint=endpoint, wait_for_action_attempt=wait_for_action_attempt
+            api_key,
+            endpoint=endpoint,
+            wait_for_action_attempt=wait_for_action_attempt,
+            retries=retries,
         )
 
     @classmethod
@@ -74,10 +84,12 @@ def from_personal_access_token(
         *,
         endpoint: Optional[str] = None,
         wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]] = True,
+        retries: Optional[Retry] = None,
     ) -> Self:
         return cls(
             personal_access_token=personal_access_token,
             workspace_id=workspace_id,
             endpoint=endpoint,
             wait_for_action_attempt=wait_for_action_attempt,
+            retries=retries,
         )
@@ -1,6 +1,7 @@
 from typing import Any, Dict, Optional, Union
 import niquests as requests
 from typing_extensions import Self
+from urllib3.util import Retry
 
 from .auth import get_auth_headers_for_multi_workspace_personal_access_token
 from .constants import LTS_VERSION
@@ -36,6 +37,7 @@ def __init__(
         *,
         endpoint: Optional[str] = None,
         wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]] = True,
+        retries: Optional[Retry] = None,
     ):
         """
         Parameters
@@ -46,6 +48,8 @@ def __init__(
           The API endpoint to which the request should be sent.
         wait_for_action_attempt : bool or dict, optional
           Controls whether to wait for an action attempt to complete, either as a boolean or as a dictionary specifying `timeout` and `poll_interval`. Defaults to `False`.
+        retries : urllib3.util.Retry, optional
+          Configuration for retry behavior on failed requests.
         """
 
         self.lts_version = SeamMultiWorkspace.lts_version
@@ -58,6 +62,7 @@ def __init__(
         self.client = SeamHttpClient(
             base_url=endpoint,
             auth_headers=auth_headers,
+            retries=retries,
         )
 
         defaults = {"wait_for_action_attempt": wait_for_action_attempt}
@@ -72,9 +77,11 @@ def from_personal_access_token(
         *,
         endpoint: Optional[str] = None,
         wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]] = True,
+        retries: Optional[Retry] = None,
     ) -> Self:
         return cls(
             personal_access_token=personal_access_token,
             endpoint=endpoint,
             wait_for_action_attempt=wait_for_action_attempt,
+            retries=retries,
         )