Transport improvements (#70)

Author: mephenor

.pre-commit-config.yaml

@@ -48,7 +48,7 @@ repos:
       - id: no-commit-to-branch
         args: [--branch, dev, --branch, int, --branch, main]
   - repo: https://github.com/astral-sh/ruff-pre-commit
-    rev: v0.14.3
+    rev: v0.14.4
     hooks:
       - id: ruff
         args: [--fix, --exit-non-zero-on-fix]

.pyproject_generation/pyproject_custom.toml

@@ -1,6 +1,6 @@
 [project]
 name = "ghga_service_commons"
-version = "5.3.0"
+version = "6.0.0"
 description = "A library that contains common functionality used in services of GHGA"
 readme = "README.md"
 authors = [

lock/requirements-dev.txt

@@ -23,7 +23,7 @@ classifiers = [
     "Intended Audience :: Developers",
 ]
 name = "ghga_service_commons"
-version = "5.3.0"
+version = "6.0.0"
 description = "A library that contains common functionality used in services of GHGA"
 dependencies = [
     "pydantic >=2, <3",

lock/requirements.txt

@@ -0,0 +1,33 @@
+# Copyright 2021 - 2025 Universität Tübingen, DKFZ, EMBL, and Universität zu Köln
+# for the German Human Genome-Phenome Archive (GHGA)
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Pluggable, custom and composable logic around HTTP calls."""
+
+from .config import CompositeCacheConfig, CompositeConfig
+from .factory import (
+    AsyncCacheTransport,
+    AsyncRateLimitingTransport,
+    AsyncRetryTransport,
+    CompositeTransportFactory,
+)
+
+__all__ = [
+    "AsyncCacheTransport",
+    "AsyncRateLimitingTransport",
+    "AsyncRetryTransport",
+    "CompositeCacheConfig",
+    "CompositeConfig",
+    "CompositeTransportFactory",
+]

pyproject.toml

@@ -25,24 +25,32 @@ class CacheTransportConfig(BaseSettings):
     Currently only in memory storage is available.
     """
 
-    cache_ttl: NonNegativeInt = Field(
+    client_cache_capacity: PositiveInt = Field(
+        default=128,
+        description="Maximum number of entries to store in the cache. Older entries are evicted once this limit is reached.",
+    )
+    client_cache_ttl: NonNegativeInt = Field(
         default=60,
         description="Number of seconds after which a stored response is considered stale.",
     )
-    cache_capacity: PositiveInt = Field(
-        default=128,
-        description="Maximum number of entries to store in the cache. Older entries are evicted once this limit is reached.",
+    client_cacheable_methods: list[str] = Field(
+        default=["POST", "GET"],
+        description="HTTP methods for which responses are allowed to be cached.",
+    )
+    client_cacheable_status_codes: list[int] = Field(
+        default=[200, 201],
+        description="HTTP response status code for which responses are allowed to be cached.",
     )
 
 
 class RateLimitingTransportConfig(BaseSettings):
     """Configuration options for a rate limiting HTTPTransport."""
 
-    jitter: NonNegativeFloat = Field(
+    per_request_jitter: NonNegativeFloat = Field(
         default=0.0,
         description="Max amount of jitter (in seconds) to add to each request.",
     )
-    reset_after: PositiveInt = Field(
+    retry_after_applicable_for_num_requests: PositiveInt = Field(
         default=1,
         description="Amount of requests after which the stored delay from a 429 response is ignored again. "
         + "Can be useful to adjust if concurrent requests are fired in quick succession.",
@@ -52,18 +60,23 @@ class RateLimitingTransportConfig(BaseSettings):
 class RetryTransportConfig(BaseSettings):
     """Configuration options for an HTTPTransport providing retry logic."""
 
-    exponential_backoff_max: NonNegativeInt = Field(
+    client_exponential_backoff_max: NonNegativeInt = Field(
         default=60,
         description="Maximum number of seconds to wait between retries when using"
         + " exponential backoff retry strategies. The client timeout might need to be adjusted accordingly.",
     )
-    max_retries: NonNegativeInt = Field(
+    client_num_retries: NonNegativeInt = Field(
         default=3, description="Number of times to retry failed API calls."
     )
-    retry_status_codes: list[NonNegativeInt] = Field(
+    client_retry_status_codes: list[NonNegativeInt] = Field(
         default=[408, 429, 500, 502, 503, 504],
         description="List of status codes that should trigger retrying a request.",
     )
+    client_reraise_from_retry_error: bool = Field(
+        default=True,
+        description="Specifies if the exception wrapped in the final RetryError is reraised "
+        "or the RetryError is returned as is.",
+    )
 
 
 class CompositeConfig(RateLimitingTransportConfig, RetryTransportConfig):

src/ghga_service_commons/transports/__init__.py

@@ -15,8 +15,8 @@
 
 """Provides factories for different flavors of httpx.AsyncHTTPTransport."""
 
-from hishel import AsyncCacheTransport, AsyncInMemoryStorage
-from httpx import AsyncHTTPTransport, Limits
+from hishel import AsyncCacheTransport, AsyncInMemoryStorage, Controller
+from httpx import AsyncBaseTransport, AsyncHTTPTransport, Limits
 
 from .config import CompositeCacheConfig, CompositeConfig
 from .ratelimiting import AsyncRateLimitingTransport
@@ -28,11 +28,21 @@ class CompositeTransportFactory:
 
     @classmethod
     def _create_common_transport_layers(
-        cls, config: CompositeConfig, limits: Limits | None = None
+        cls,
+        config: CompositeConfig,
+        base_transport: AsyncBaseTransport | None = None,
+        limits: Limits | None = None,
     ):
-        """Creates wrapped transports reused between different factory methods."""
+        """Creates wrapped transports reused between different factory methods.
+
+        If provided, limits are applied to the AsyncHTTPTransport instance this method creates.
+        If provided, a custom base_transport class is used and any limits are ignored.
+        Those have to be provided directly to the custom base_transport passed into this method.
+        """
         base_transport = (
-            AsyncHTTPTransport(limits=limits) if limits else AsyncHTTPTransport()
+            base_transport or AsyncHTTPTransport(limits=limits)
+            if limits
+            else AsyncHTTPTransport()
         )
         ratelimiting_transport = AsyncRateLimitingTransport(
             config=config, transport=base_transport
@@ -44,16 +54,44 @@ def _create_common_transport_layers(
 
     @classmethod
     def create_ratelimiting_retry_transport(
-        cls, config: CompositeConfig, limits: Limits | None = None
+        cls,
+        config: CompositeConfig,
+        base_transport: AsyncBaseTransport | None = None,
+        limits: Limits | None = None,
     ) -> AsyncRetryTransport:
-        """Creates a retry transport, wrapping a rate limiting transport, wrapping an AsyncHTTPTransport."""
-        return cls._create_common_transport_layers(config, limits=limits)
+        """Creates a retry transport, wrapping, in sequence, a rate limiting transport and AsyncHTTPTransport.
+
+        If provided, limits are applied to the wrapped AsyncHTTPTransport instance.
+        If provided, a custom base_transport class is used and any limits are ignored.
+        Those have to be provided directly to the custom base_transport passed into this method.
+        """
+        return cls._create_common_transport_layers(
+            config, base_transport=base_transport, limits=limits
+        )
 
     @classmethod
     def create_cached_ratelimiting_retry_transport(
-        cls, config: CompositeCacheConfig, limits: Limits | None = None
+        cls,
+        config: CompositeCacheConfig,
+        base_transport: AsyncBaseTransport | None = None,
+        limits: Limits | None = None,
     ) -> AsyncCacheTransport:
-        """Creates a retry transport, wrapping a rate limiting transport, wrapping a cache transport, wrapping an AsyncHTTPTransport."""
-        storage = AsyncInMemoryStorage(ttl=config.cache_ttl)
-        retry_transport = cls._create_common_transport_layers(config, limits=limits)
-        return AsyncCacheTransport(transport=retry_transport, storage=storage)
+        """Creates a cache transport, wrapping, in sequence, a retry, rate limiting transport and AsyncHTTPTransport.
+
+        If provided, limits are applied to the wrapped AsyncHTTPTransport instance.
+        If provided, a custom base_transport class is used and any limits are ignored.
+        Those have to be provided directly to the custom base_transport passed into this method.
+        """
+        retry_transport = cls._create_common_transport_layers(
+            config, base_transport=base_transport, limits=limits
+        )
+        controller = Controller(
+            cacheable_methods=config.client_cacheable_methods,
+            cacheable_status_codes=config.client_cacheable_status_codes,
+        )
+        storage = AsyncInMemoryStorage(
+            ttl=config.client_cache_ttl, capacity=config.client_cache_capacity
+        )
+        return AsyncCacheTransport(
+            controller=controller, transport=retry_transport, storage=storage
+        )

src/ghga_service_commons/transports/config.py

@@ -44,10 +44,10 @@ class AsyncRateLimitingTransport(httpx.AsyncBaseTransport):
     def __init__(
         self, config: RateLimitingTransportConfig, transport: httpx.AsyncBaseTransport
     ) -> None:
-        self._jitter = config.jitter
+        self._jitter = config.per_request_jitter
         self._transport = transport
         self._num_requests = 0
-        self._reset_after: int = config.reset_after
+        self._reset_after: int = config.retry_after_applicable_for_num_requests
         self._last_retry_after_received: float = 0
         self._wait_time: float = 0
 

src/ghga_service_commons/transports/factory.py

@@ -17,12 +17,12 @@
 
 import time
 from collections.abc import Callable
+from contextlib import suppress
 from logging import getLogger
 from types import TracebackType
 from typing import Any, Self
 
 import httpx
-import tenacity
 from tenacity import (
     AsyncRetrying,
     RetryCallState,
@@ -39,12 +39,12 @@
 
 def _default_wait_strategy(config: RetryTransportConfig):
     """Wait strategy using exponential backoff, not waiting for 429 responses."""
-    return wait_exponential_ignore_429(max=config.exponential_backoff_max)
+    return wait_exponential_ignore_429(max=config.client_exponential_backoff_max)
 
 
 def _default_stop_strategy(config: RetryTransportConfig):
     """Basic stop strategy aborting retrying after a configured number of attempts."""
-    return stop_after_attempt(config.max_retries)
+    return stop_after_attempt(config.client_num_retries)
 
 
 def _log_retry_stats(retry_state: RetryCallState):
@@ -65,13 +65,15 @@ def _log_retry_stats(retry_state: RetryCallState):
 
     # Enrich with details from current attempt for debugging
     if outcome := retry_state.outcome:
-        result = outcome.result()
-        if isinstance(result, httpx.Response):
-            stats["response_status_code"] = result.status_code
-            stats["response_headers"] = result.headers
-        elif isinstance(result, Exception):
-            stats["exception_type"] = type(result)
-            stats["exception_message"] = str(result)
+        try:
+            result = outcome.result()
+        except Exception as exc:
+            stats["exception_type"] = type(exc)
+            stats["exception_message"] = str(exc)
+        else:
+            if isinstance(result, httpx.Response):
+                stats["response_status_code"] = result.status_code
+                stats["response_headers"] = result.headers
 
     log.info(
         "Retry attempt number %i for function %s.",
@@ -90,14 +92,15 @@ class wait_exponential_ignore_429(wait_exponential):  # noqa: N801
 
     def __call__(self, retry_state: RetryCallState) -> float:
         """Copied from base class and adjusted."""
-        if (
-            retry_state.outcome
-            and (result := retry_state.outcome.result())
-            and isinstance(result, httpx.Response)
-            and result.status_code == 429
-            and not result.headers.get("Should-Wait")
-        ):
-            return 0
+        if retry_state.outcome:
+            with suppress(Exception):
+                result = retry_state.outcome.result()
+                if (
+                    isinstance(result, httpx.Response)
+                    and result.status_code == 429
+                    and not result.headers.get("Should-Wait")
+                ):
+                    return 0
         try:
             exp = self.exp_base ** (retry_state.attempt_number - 1)
             result = self.multiplier * exp
@@ -133,16 +136,9 @@ def __init__(
 
     async def handle_async_request(self, request: httpx.Request) -> httpx.Response:
         """Handles HTTP requests and adds retry logic around calls."""
-        response = None
-        try:
-            response = await self._retry_handler(
-                fn=self._transport.handle_async_request, request=request
-            )
-        except tenacity.RetryError as exc:
-            # last_attempt is potentially an unawaited future, need to explicitly reraise
-            # to get the correct inner instance displayed
-            exc.reraise()
-        return response
+        return await self._retry_handler(
+            fn=self._transport.handle_async_request, request=request
+        )
 
     async def aclose(self) -> None:  # noqa: D102
         await self._transport.aclose()
@@ -167,7 +163,7 @@ def _configure_retry_handler(
 ):
     """Configure the AsyncRetrying instance that is used for handling retryable responses/exceptions."""
     return AsyncRetrying(
-        reraise=True,
+        reraise=config.client_reraise_from_retry_error,
         retry=(
             retry_if_exception_type(
                 (
@@ -177,7 +173,8 @@ def _configure_retry_handler(
                 )
             )
             | retry_if_result(
-                lambda response: response.status_code in config.retry_status_codes
+                lambda response: response.status_code
+                in config.client_retry_status_codes
             )
         ),
         stop=stop_strategy(config),