Basic custom transports

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.13.1
+    rev: v0.14.2
     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.1.0"
+version = "5.2.0"
 description = "A library that contains common functionality used in services of GHGA"
 readme = "README.md"
 authors = [
@@ -14,15 +14,18 @@ api = [
     "uvicorn[standard]>=0.35, <0.37",
     "ghga-service-commons[http,objectstorage]",
 ]
-http = [
-    "httpx>=0.27, <0.29",
-]
+http = ["httpx>=0.27, <0.29"]
 auth = ["jwcrypto>=1.5.6, <2", "pydantic[email]>=2, <3"]
 crypt = ["pynacl>=1.5, <2", "crypt4gh>=1.6, <2"]
 dev = ["requests>=2.31, <3"]
 objectstorage = ["hexkit>=6"]
+transports = [
+    "hishel>=0.1.1, <0.2",
+    "tenacity >=9.0.0, <10",
+    "ghga-service-commons[transports]",
+]
 
-all = ["ghga-service-commons[api,auth,crypt,dev,http,objectstorage]"]
+all = ["ghga-service-commons[api,auth,crypt,dev,http,objectstorage,transports]"]
 
 [project.license]
 text = "Apache 2.0"
@@ -49,7 +52,7 @@ legacy_tox_ini = """
     deps =
         --no-deps -r ./lock/requirements-dev.txt
     commands =
-        py3{9,10}: pip install --no-deps \
+        py310: pip install --no-deps \
           backports.asyncio.runner==1.2.0 \
           exceptiongroup==1.3.0
         pytest {posargs}

lock/requirements-dev.txt

@@ -23,7 +23,7 @@ classifiers = [
     "Intended Audience :: Developers",
 ]
 name = "ghga_service_commons"
-version = "5.1.0"
+version = "5.2.0"
 description = "A library that contains common functionality used in services of GHGA"
 dependencies = [
     "pydantic >=2, <3",
@@ -55,8 +55,13 @@ dev = [
 objectstorage = [
     "hexkit>=6",
 ]
+transports = [
+    "hishel>=0.1.1, <0.2",
+    "tenacity >=9.0.0, <10",
+    "ghga-service-commons[transports]",
+]
 all = [
-    "ghga-service-commons[api,auth,crypt,dev,http,objectstorage]",
+    "ghga-service-commons[api,auth,crypt,dev,http,objectstorage,transports]",
 ]
 
 [project.urls]
@@ -174,4 +179,24 @@ asyncio_mode = "strict"
 asyncio_default_fixture_loop_scope = "function"
 
 [tool.tox]
-legacy_tox_ini = "    [tox]\n    env_list = py3{10,11,12,13}\n\n    [gh-actions]\n    python =\n        3.10: py310\n        3.11: py311\n        3.12: py312\n        3.13: py313\n\n    [testenv]\n    pass_env =\n        TC_HOST\n        DOCKER_HOST\n    deps =\n        --no-deps -r ./lock/requirements-dev.txt\n    commands =\n        py3{9,10}: pip install --no-deps backports.asyncio.runner==1.2.0 exceptiongroup==1.3.0\n        pytest {posargs}\n"
+legacy_tox_ini = """
+    [tox]
+    env_list = py3{10,11,12,13}
+
+    [gh-actions]
+    python =
+        3.10: py310
+        3.11: py311
+        3.12: py312
+        3.13: py313
+
+    [testenv]
+    pass_env =
+        TC_HOST
+        DOCKER_HOST
+    deps =
+        --no-deps -r ./lock/requirements-dev.txt
+    commands =
+        py310: pip install --no-deps backports.asyncio.runner==1.2.0 exceptiongroup==1.3.0
+        pytest {posargs}
+"""

lock/requirements.txt

@@ -0,0 +1,73 @@
+# 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.
+
+"""TODO"""
+
+from pydantic import Field, NonNegativeFloat, NonNegativeInt, PositiveInt
+from pydantic_settings import BaseSettings
+
+
+class CacheTransportConfig(BaseSettings):
+    """TODO"""
+
+    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.",
+    )
+
+
+class RatelimitingTransportConfig(BaseSettings):
+    """TODO"""
+
+    jitter: NonNegativeFloat = Field(
+        default=0.05, description="Max amoun of jitter to add to each request"
+    )
+    reset_after: int | None = Field(
+        default=None,
+        description="Amount of requests after which the stored delay from a 429 response is ignored again. If set to `None`, it's never forgotten.",
+    )
+
+
+class RetryTransportConfig(BaseSettings):
+    """TODO"""
+
+    exponential_backoff_max: NonNegativeInt = Field(
+        default=60,
+        description="Maximum number of seconds to wait for when using exponential backoff retry strategies.",
+    )
+    log_retries: bool = Field(
+        default=False, description="If true, retry informtion will be logged"
+    )
+    max_retries: NonNegativeInt = Field(
+        default=3, description="Number of times to retry failed API calls."
+    )
+    retry_status_codes: list[NonNegativeInt] = Field(
+        default=[408, 429, 500, 502, 503, 504],
+        description="List of status codes that should trigger retrying a request.",
+    )
+
+
+class CompositeConfig(RatelimitingTransportConfig, RetryTransportConfig):
+    """TOOD"""
+
+
+class CompositeCacheConfig(
+    CacheTransportConfig, RatelimitingTransportConfig, RetryTransportConfig
+):
+    """TOOD"""

pyproject.toml

@@ -0,0 +1,51 @@
+# 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.
+
+"""Provides factories for different flavors of httpx.AsyncHTTPTransport."""
+
+from hishel import AsyncCacheTransport, AsyncInMemoryStorage
+from httpx import AsyncHTTPTransport
+
+from .ratelimiting import AsyncRatelimitingTransport
+from .retry import AsyncRetryTransport
+
+
+class CompositeTransportFactory:
+    """TODO"""
+
+    @classmethod
+    def _create_common_transport_layers(cls, config, transport):
+        """TODO"""
+        retry_transport = AsyncRetryTransport(config=config, transport=transport)
+        ratelimiting_transport = AsyncRatelimitingTransport(
+            config=config, transport=retry_transport
+        )
+        return ratelimiting_transport
+
+    @classmethod
+    def create_ratelimiting_retry_transport(cls, config) -> AsyncRatelimitingTransport:
+        """TODO"""
+        base_transport = AsyncHTTPTransport()
+        return cls._create_common_transport_layers(config, base_transport)
+
+    @classmethod
+    def create_ratelimiting_retry_transport_with_cache(
+        cls, config
+    ) -> AsyncRatelimitingTransport:
+        """TODO"""
+        base_transport = AsyncHTTPTransport()
+        storage = AsyncInMemoryStorage(ttl=config.cache_ttl)
+        cache_tranport = AsyncCacheTransport(transport=base_transport, storage=storage)
+        return cls._create_common_transport_layers(config, cache_tranport)

src/ghga_service_commons/transports/config.py

@@ -0,0 +1,89 @@
+# 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.
+
+"""Provides an httpx.AsyncTransport that handles rate limiting responses."""
+
+import asyncio
+import random
+from datetime import datetime, timezone
+from types import TracebackType
+from typing import Self
+
+import httpx
+
+from ghga_service_commons.transports.config import RatelimitingTransportConfig
+
+
+class AsyncRatelimitingTransport(httpx.AsyncBaseTransport):
+    """TODO"""
+
+    def __init__(
+        self, config: RatelimitingTransportConfig, transport: httpx.AsyncBaseTransport
+    ) -> None:
+        self._jitter = config.jitter
+        self._num_requests = 0
+        self._reset_after: int | None = config.reset_after
+        self._transport = transport
+        self._last_call_time = datetime.now(timezone.utc)
+        self._wait_time: float = 0
+
+    async def handle_async_request(self, request: httpx.Request) -> httpx.Response:
+        """
+        Handles HTTP requests while also implementing HTTP caching.
+
+        :param request: An HTTP request
+        :type request: httpx.Request
+        :return: An HTTP response
+        :rtype: httpx.Response
+        """
+        # Caculate seconds since the last request has been fired and corresponding wait time
+        time_elapsed = (self._last_call_time - datetime.now(timezone.utc)).seconds
+        remaining_wait = max(0, self._wait_time - time_elapsed)
+
+        # Add jitter to both cases and sleep
+        if remaining_wait < self._jitter:
+            await asyncio.sleep(random.uniform(remaining_wait, self._jitter))  # noqa: S311
+        else:
+            await asyncio.sleep(
+                random.uniform(remaining_wait, remaining_wait + self._jitter)  # noqa: S311
+            )
+
+        # Update timestamp and delegate call
+        self._last_call_time = datetime.now(timezone.utc)
+        response = await self._transport.handle_async_request(request=request)
+
+        # Update state
+        self._num_requests += 1
+        if response.status_code == 429:
+            self._wait_time = float(response.headers["Retry-After"])
+            self._num_requests = 0
+        elif self._reset_after and self._reset_after <= self._num_requests:
+            self._wait_time = 0
+            self._num_requests = 0
+        return response
+
+    async def aclose(self) -> None:
+        await self._transport.aclose()
+
+    async def __aenter__(self) -> Self:
+        return self
+
+    async def __aexit__(
+        self,
+        exc_type: type[BaseException] | None = None,
+        exc_value: BaseException | None = None,
+        traceback: TracebackType | None = None,
+    ) -> None:
+        await self.aclose()