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

.github/workflows/main.yml

@@ -3,7 +3,7 @@ name: CI
 on: push
 
 env:
-  PYTHON_VERSIONS: '[ "3.8", "3.9", "3.10", "3.11", "3.12", "3.13" ]'
+  PYTHON_VERSIONS: '[ "3.8", "3.9", "3.10", "3.11", "3.12", "3.13.3" ]'
 
 jobs:
 

CHANGELOG.md

@@ -6,6 +6,20 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [1.14.1] - 2025-07-25
+### Added
+- Session mechanism for significantly decrease number of an http load on data-api for apps with intensive calling 
+- Added possibility to adjust some params related to connection pool
+  - `POOL_CONNECTIONS_COUNT`: Total pools count
+  - `POOL_MAX_SIZE`: Max connections count per pool/host 
+  - `POOL_BLOCK`: Wait until connection released or not (instantly raise an exception)
+  - `MAX_RETRY_COUNT`: If 0 then retires will be disabled, otherwise retrying logic will be used
+- Move retrying logic from `tenacity` to internal `urllib3.util.Retry(...)`
+- Removed redundant dependency `tenacity` from `python-sdk`
+- Bump version for `py3.13` to `py3.13.3` at CI version matrix in order to fix broken tests for logging
+- Bump version for `fakeredis` to fix some tests
+
+
 ## [1.14.0] - 2025-04-17
 ### Fixed
 - merge_events parameter for scheduled data time apps should result in correct start/end times in a final app event.
@@ -405,7 +419,8 @@ env variables, that should be used to configure logging.
 - Event classes: `StreamEvent`, `ScheduledEvent` and `TaskEvent`.
 
 
-[Unreleased] https://github.com/corva-ai/python-sdk/compare/v1.14.0...master
+[Unreleased] https://github.com/corva-ai/python-sdk/compare/v1.14.1...master
+[1.14.1] https://github.com/corva-ai/python-sdk/compare/v1.14.0...v1.14.1
 [1.14.0] https://github.com/corva-ai/python-sdk/compare/v1.13.1...v1.14.0
 [1.13.1] https://github.com/corva-ai/python-sdk/compare/v1.13.0...v1.13.1
 [1.13.0] https://github.com/corva-ai/python-sdk/compare/v1.12.1...v1.13.0

docs/antora-playbook.yml

@@ -7,7 +7,7 @@ content:
       start_path: docs
       branches: []
 #      branches: HEAD  # Use this for local development
-      tags: [v1.14.0]
+      tags: [v1.14.1]
 asciidoc:
   attributes:
     page-toclevels: 5

docs/antora.yml

@@ -1,3 +1,3 @@
 name: corva-sdk
-version: ~
+version: 1.14.1
 nav: [modules/ROOT/nav.adoc]

setup.py

@@ -39,12 +39,11 @@
     packages=setuptools.find_packages("src"),
     package_dir={"": "src"},
     install_requires=[
-        "fakeredis[lua] >=2.26.2, <3.0.0",
+        "fakeredis[lua] >=2.26.2, <2.30.0",
         "pydantic >=1.8.2, <2.0.0",
         "redis >=5.2.1, <6.0.0",
         "requests >=2.32.3, <3.0.0",
         "urllib3 <2",  # lambda doesnt support version 2 yet
-        "tenacity >=8.2.3, <9.0.0",
     ],
     python_requires='>=3.8, <4.0',
     license='The Unlicense',

src/corva/api.py

@@ -1,17 +1,12 @@
 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_requests_session, get_retry_strategy
+from corva.configuration import SETTINGS
 
 
 class Api:
@@ -21,26 +16,34 @@ 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] = 3,
+        backoff_factor_retries: Optional[float] = 1,
+        pool_conn_count: Optional[int] = None,
+        pool_max_size: Optional[int] = None,
+        pool_block: Optional[bool] = None,
     ):
         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._session = get_requests_session(
+            retry_strategy=get_retry_strategy(
+                max_retries=max_retries or SETTINGS.MAX_RETRY_COUNT,
+                backoff_factor=backoff_factor_retries or SETTINGS.BACKOFF_FACTOR,
+            ),
+            pool_connections_count=(pool_conn_count or SETTINGS.POOL_CONNECTIONS_COUNT),
+            pool_max_size=pool_max_size or SETTINGS.POOL_MAX_SIZE,
+            pool_block=pool_block or SETTINGS.POOL_BLOCK,
+        )
 
     @property
     def default_headers(self):
@@ -49,16 +52,6 @@ def default_headers(self):
             "X-Corva-App": self.app_key,
         }
 
-    @property
-    def max_retries(self) -> int:
-        return self._max_retries
-
-    @max_retries.setter
-    def max_retries(self, value: int):
-        if not (0 <= value <= 10):
-            raise ValueError("Values between 0 and 10 are allowed")
-        self._max_retries = value
-
     def get(self, path: str, **kwargs):
         return self._request("GET", path, **kwargs)
 
@@ -99,15 +92,15 @@ 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,12 +109,12 @@ 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,
@@ -148,70 +141,25 @@ 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(
-                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."
-            )
+        return self._execute_request(
+            method=method,
+            url=url,
+            params=params,
+            data=data,
+            headers=headers,
+            timeout=timeout,
+        )
 
     def get_dataset(
         self,

src/corva/api_utils.py

@@ -0,0 +1,57 @@
+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
+)
+
+# All HTTP methods allowed, see this discussion:
+# https://corva.slack.com/archives/C0411LUPVL6/p1753451234091869
+ALLOWED_RETRY_METHODS = (
+    "GET",
+    "POST",
+    "PUT",
+    "PATCH",
+    "DELETE",
+    "OPTIONS",
+    "HEAD",
+    "TRACE",
+)
+
+
+def get_retry_strategy(max_retries: int, backoff_factor: float = 1) -> Retry:
+    return Retry(
+        total=max_retries,
+        backoff_factor=backoff_factor,
+        status_forcelist=RETRYABLE_STATUS_CODES,
+        raise_on_status=False,
+        allowed_methods=ALLOWED_RETRY_METHODS,
+    )
+
+
+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,14 @@ 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  # Wait until connection released
+
+    # retry
+    MAX_RETRY_COUNT: int = 3  # If `0` then retries will be disabled
+    BACKOFF_FACTOR: float = 1.0
+
 
 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,
         )
 

src/corva/logger.py

@@ -12,6 +12,8 @@
 CORVA_LOGGER = logging.getLogger('corva')
 CORVA_LOGGER.setLevel(SETTINGS.LOG_LEVEL)
 
+logging.getLogger("urllib3.connectionpool").setLevel(SETTINGS.LOG_LEVEL)
+
 # unset to pass messages to ancestor loggers, including OTel Log Sending handler
 # see https://github.com/corva-ai/otel/pull/37
 # see https://corvaqa.atlassian.net/browse/EE-31

src/version.py

@@ -1 +1 @@
-VERSION = "1.14.0"
+VERSION = "1.14.1"