feat(DEVC-1752): added requests.Session reuse; replaced third-party tenacity with builtin urllib3 Retry

Author: kossman

src/corva/api.py

@@ -1,17 +1,13 @@
 import json
 import posixpath
 import re
-from http import HTTPStatus
+
 from typing import List, Optional, Sequence, Union
 
 import requests
-from tenacity import (
-    RetryError,
-    retry,
-    retry_if_result,
-    stop_after_attempt,
-    wait_random_exponential,
-)
+
+from corva.api_utils import get_retry_strategy, get_requests_session
+from corva.configuration import SETTINGS
 
 
 class Api:
@@ -21,26 +17,31 @@ class Api:
     convenient URL usage and reasonable timeouts to API requests.
     """
 
-    TIMEOUT_LIMITS = (3, 30)  # seconds
-    DEFAULT_MAX_RETRIES = int(0)
-
     def __init__(
         self,
         *,
         api_url: str,
         data_api_url: str,
         api_key: str,
         app_key: str,
-        timeout: Optional[int] = None,
         app_connection_id: Optional[int] = None,
+        max_retries: Optional[int] = SETTINGS.MAX_RETRY_COUNT,
+        pool_connections_count: Optional[int] = SETTINGS.POOL_CONNECTIONS_COUNT,
+        pool_max_size: Optional[int] = SETTINGS.POOL_MAX_SIZE,
+        pool_block: Optional[bool] = SETTINGS.POOL_BLOCK,
     ):
         self.api_url = api_url
         self.data_api_url = data_api_url
         self.api_key = api_key
         self.app_key = app_key
         self.app_connection_id = app_connection_id
-        self.timeout = timeout or self.TIMEOUT_LIMITS[1]
-        self._max_retries = self.DEFAULT_MAX_RETRIES
+        self._retry_strategy = get_retry_strategy(max_retries) if max_retries not in (None, 0) else None
+        self._session = get_requests_session(
+            retry_strategy=self._retry_strategy,
+            pool_connections_count=pool_connections_count,
+            pool_max_size=pool_max_size,
+            pool_block=pool_block,
+        )
 
     @property
     def default_headers(self):
@@ -99,15 +100,14 @@ def _get_url(self, path: str):
 
         return posixpath.join(self.api_url, path)
 
-    @staticmethod
     def _execute_request(
+        self,
         method: str,
         url: str,
         params: Optional[dict],
         data: Optional[dict],
         headers: Optional[dict] = None,
-        timeout: Optional[int] = None,
-    ):
+    ) -> requests.Response:
         """Executes the request.
 
         Args:
@@ -116,18 +116,17 @@ def _execute_request(
             data: request body, that will be casted to json.
             params: url query string params.
             headers: additional headers to include in request.
-            timeout: custom request timeout in seconds.
 
         Returns:
             requests.Response instance.
         """
-        return requests.request(
+
+        return self._session.request(
             method=method,
             url=url,
             params=params,
             json=data,
             headers=headers,
-            timeout=timeout,
         )
 
     def _request(
@@ -138,7 +137,6 @@ def _request(
         data: Optional[dict] = None,
         params: Optional[dict] = None,
         headers: Optional[dict] = None,
-        timeout: Optional[int] = None,
     ) -> requests.Response:
         """Prepares HTTP request.
 
@@ -148,70 +146,24 @@ def _request(
           data: request body, that will be casted to json.
           params: url query string params.
           headers: additional headers to include in request.
-          timeout: custom request timeout in seconds.
 
         Returns:
           requests.Response instance.
         """
-        retryable_status_codes = [
-            HTTPStatus.TOO_MANY_REQUESTS,  # 428
-            HTTPStatus.INTERNAL_SERVER_ERROR,  # 500
-            HTTPStatus.BAD_GATEWAY,  # 502
-            HTTPStatus.SERVICE_UNAVAILABLE,  # 503
-            HTTPStatus.GATEWAY_TIMEOUT,  # 504
-        ]
-
-        timeout = timeout or self.timeout
-        self._validate_timeout(timeout)
-
         url = self._get_url(path)
 
         headers = {
             **self.default_headers,
             **(headers or {}),
         }
 
-        if self.max_retries > 0:
-            retry_decorator = retry(
-                stop=stop_after_attempt(self.max_retries),
-                wait=wait_random_exponential(multiplier=0.25, max=10),
-                retry=retry_if_result(
-                    lambda r: r.status_code in retryable_status_codes
-                ),
-            )
-            retrying_request = retry_decorator(self._execute_request)
-            try:
-                response = retrying_request(
-                    method=method,
-                    url=url,
-                    params=params,
-                    data=data,
-                    headers=headers,
-                    timeout=timeout,
-                )
-            except RetryError as e:
-                if not e.last_attempt.failed:
-                    response = e.last_attempt.result()
-                else:
-                    raise
-        else:
-            response = self._execute_request(
+        return self._execute_request(
                 method=method,
                 url=url,
                 params=params,
                 data=data,
                 headers=headers,
-                timeout=timeout,
-            )
-
-        return response
-
-    def _validate_timeout(self, timeout: int) -> None:
-        if self.TIMEOUT_LIMITS[0] > timeout or self.TIMEOUT_LIMITS[1] < timeout:
-            raise ValueError(
-                f"Timeout must be between {self.TIMEOUT_LIMITS[0]} and "
-                f"{self.TIMEOUT_LIMITS[1]} seconds."
-            )
+        )
 
     def get_dataset(
         self,

src/corva/api_utils.py

@@ -0,0 +1,44 @@
+from typing import Optional
+
+import requests
+from requests.adapters import HTTPAdapter
+from urllib3 import Retry
+
+RETRYABLE_STATUS_CODES = (
+    429,  # HTTPStatus.TOO_MANY_REQUESTS
+    500,  # HTTPStatus.INTERNAL_SERVER_ERROR
+    502,  # HTTPStatus.BAD_GATEWAY
+    503,  # HTTPStatus.SERVICE_UNAVAILABLE
+    504,  # HTTPStatus.GATEWAY_TIMEOUT
+)
+
+
+def get_retry_strategy(max_retries: int) -> Retry:
+    return Retry(
+        total=max_retries,
+        backoff_factor=0.3,
+        status_forcelist=RETRYABLE_STATUS_CODES,
+        allowed_methods={'GET', 'POST'},  # Support both
+        raise_on_status=False,
+    )
+
+
+def get_requests_session(
+        pool_connections_count: int,
+        pool_max_size: int,
+        pool_block: bool,
+        retry_strategy: Optional[Retry] = None,
+) -> requests.Session:
+    adapter = HTTPAdapter(
+        max_retries=retry_strategy,
+        pool_connections=pool_connections_count,
+        pool_maxsize=pool_max_size,
+        pool_block=pool_block,
+    )
+
+    session = requests.Session()
+
+    session.mount('https://', adapter)
+    session.mount('http://', adapter)
+
+    return session

src/corva/configuration.py

@@ -23,5 +23,13 @@ class Settings(pydantic.BaseSettings):
     # secrets
     SECRETS_CACHE_TTL: int = int(datetime.timedelta(minutes=5).total_seconds())
 
+    # keep-alive
+    POOL_CONNECTIONS_COUNT: int = 20    # Total pools count
+    POOL_MAX_SIZE: int = 20             # Max connections count per pool/host
+    POOL_BLOCK: bool = True             # If all conn ack - wait until pool connection released, do not throw exc
+
+    # retry
+    MAX_RETRY_COUNT: int = 3            # If `0` then retires will be disabled
+
 
 SETTINGS = Settings()

src/corva/handlers.py

@@ -173,7 +173,6 @@ def wrapper(
             data_api_url=SETTINGS.DATA_API_ROOT_URL,
             api_key=api_key,
             app_key=SETTINGS.APP_KEY,
-            timeout=None,
             app_connection_id=event.app_connection_id,
         )
 
@@ -282,7 +281,6 @@ def wrapper(
             data_api_url=SETTINGS.DATA_API_ROOT_URL,
             api_key=api_key,
             app_key=SETTINGS.APP_KEY,
-            timeout=None,
             app_connection_id=event.app_connection_id,
         )
 
@@ -392,7 +390,6 @@ def wrapper(
             data_api_url=SETTINGS.DATA_API_ROOT_URL,
             api_key=api_key,
             app_key=SETTINGS.APP_KEY,
-            timeout=None,
             app_connection_id=None,
         )