Merge pull request #118 from jonasscheid/fix/117-ensembl-rest-fail-loud

jonasscheid · web-flow · commit 86cabc0f7f25 · 2026-06-11T11:24:46.000+02:00
Fail loud on EnsemblRESTAdapter transient failures + client-side rate limiting
diff --git a/.github/workflows/python-test-conda.yml b/.github/workflows/python-test-conda.yml
@@ -24,9 +24,12 @@ jobs:
         auto-activate-base: false
     - name: Install epytope
       shell: bash -el {0}
+      # --no-cache-dir avoids a corrupt runner pip cache making a present
+      # package look unavailable ("from versions: none"); retries/timeout
+      # harden against transient PyPI index hiccups.
       run: |
-        pip install "numpy<2"
-        pip install ".[dev]"
+        pip install --no-cache-dir --retries 5 --timeout 60 "numpy<2"
+        pip install --no-cache-dir --retries 5 --timeout 60 ".[dev]"
     - name: Install Test dependencies
       shell: bash -el {0}
       run: |
diff --git a/.gitignore b/.gitignore
@@ -9,3 +9,8 @@ dist/*
 .idea
 *.pyc
 epytope/tutorials/.ipynb_checkpoints/*
+
+# Local agent/workspace files (never commit)
+CLAUDE.md
+.claude/
+docs/superpowers/
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -3,6 +3,12 @@
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/)
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## Unreleased
+
+### `Fixed`
+
+- [#118](https://github.com/KohlbacherLab/epytope/pull/118) `EnsemblRESTAdapter` now raises `EnsemblRateLimitError` / `EnsemblConnectionError` on definitive transient failures (rate-limit retries exhausted, connection error, timeout) instead of silently returning `None`, which previously caused silently truncated variant→peptide results under rate limiting. Added a configurable client-side rate limiter (`max_requests_per_second`, `max_requests_per_hour`) and a unit-tested `_request` failure path. [#117](https://github.com/KohlbacherLab/epytope/issues/117)
+
 ## v4.0.0 - 2026-04-01
 
 ### `Added`
diff --git a/epytope/IO/EnsemblRESTAdapter.py b/epytope/IO/EnsemblRESTAdapter.py
@@ -9,6 +9,7 @@
 
 import logging
 import time
+from collections import deque
 
 import pandas as pd
 import requests
@@ -17,6 +18,47 @@
 
 from epytope.IO.ADBAdapter import ADBAdapter, EAdapterFields, EIdentifierTypes
 
+
+class EnsemblRESTError(Exception):
+    """Definitive failure talking to the Ensembl REST API (not a 'not found')."""
+
+
+class EnsemblRateLimitError(EnsemblRESTError):
+    """429 rate-limit retries exhausted."""
+
+
+class EnsemblConnectionError(EnsemblRESTError):
+    """Connection error or timeout talking to the Ensembl REST API."""
+
+
+class _RateLimiter:
+    """Sliding-window limiter enforcing several (max_calls, period_seconds) caps.
+
+    Blocks in ``acquire()`` until making a call would not breach any window.
+    Not thread-safe (the adapter is used single-threaded).
+    """
+
+    def __init__(self, limits):
+        # limits: iterable of (max_calls, period_seconds)
+        self._windows = [(n, p, deque()) for n, p in limits]
+
+    def acquire(self):
+        while True:
+            now = time.monotonic()
+            wait = 0.0
+            for n, p, hits in self._windows:
+                while hits and hits[0] <= now - p:
+                    hits.popleft()
+                if len(hits) >= n:
+                    wait = max(wait, hits[0] + p - now)
+            if wait <= 0:
+                break
+            time.sleep(wait)
+        now = time.monotonic()
+        for _, _, hits in self._windows:
+            hits.append(now)
+
+
 _DB_TO_SPECIES = {
     "hsapiens_gene_ensembl": "homo_sapiens",
     "mmusculus_gene_ensembl": "mus_musculus",
@@ -41,11 +83,15 @@ class EnsemblRESTAdapter(ADBAdapter):
     :param str species: Species name (default: 'homo_sapiens')
     """
 
-    def __init__(self, server='https://rest.ensembl.org', species='homo_sapiens'):
+    def __init__(self, server='https://rest.ensembl.org', species='homo_sapiens',
+                 max_requests_per_second=13, max_requests_per_hour=50000,
+                 max_rate_limit_retries=3):
         self._server = server.rstrip('/')
         self._species = species
+        self._max_rate_limit_retries = max_rate_limit_retries
 
         self._session = requests.Session()
+        # 429 is handled in _request (Retry-After); urllib3 covers 5xx + connection.
         retry = Retry(
             total=3,
             status_forcelist=[500, 502, 503, 504],
@@ -55,6 +101,11 @@ def __init__(self, server='https://rest.ensembl.org', species='homo_sapiens'):
         self._session.mount('https://', HTTPAdapter(max_retries=retry))
         self._session.mount('http://', HTTPAdapter(max_retries=retry))
 
+        self._rate_limiter = _RateLimiter([
+            (max_requests_per_second, 1.0),
+            (max_requests_per_hour, 3600.0),
+        ])
+
         self._ids_cache = {}
         self._sequence_cache = {}
         self._gene_cache = {}
@@ -90,46 +141,61 @@ def _resolve_refseq_to_ensembl(self, refseq_id, expected_type, species):
     def _request(self, endpoint, params=None, content_type='application/json',
                  method='GET', data=None):
         """
-        Make a rate-limit-aware request to the Ensembl REST API.
+        Make a rate-limited request to the Ensembl REST API.
 
-        Returns parsed JSON (dict/list) or raw text string depending on content_type.
-        Returns None on error.
+        :return: parsed JSON (dict/list) or raw text for 2xx; None for a genuine
+                 'not found' (HTTP 400/404).
+        :raises EnsemblRateLimitError: 429 rate-limit retries exhausted.
+        :raises EnsemblConnectionError: connection error or timeout.
+        :raises EnsemblRESTError: other definitive HTTP failure (e.g. 5xx exhausted).
         """
         url = self._server + endpoint
         headers = {"Content-Type": content_type, "Accept": content_type}
 
-        for attempt in range(4):  # 1 initial + 3 retries
+        for _ in range(self._max_rate_limit_retries + 1):
+            self._rate_limiter.acquire()
             try:
                 if method == 'POST':
-                    headers["Content-Type"] = "application/json"
-                    headers["Accept"] = content_type
-                    resp = self._session.post(url, headers=headers, json=data, timeout=30)
+                    resp = self._session.post(
+                        url, headers=headers | {"Content-Type": "application/json"},
+                        json=data, timeout=30)
                 else:
-                    resp = self._session.get(url, headers=headers, params=params, timeout=30)
-
-                # Handle rate limiting
-                if resp.status_code == 429:
-                    retry_after = float(resp.headers.get("Retry-After", 1.0))
-                    logging.warning("Rate limited by Ensembl REST API, sleeping %.1fs", retry_after)
-                    time.sleep(retry_after)
-                    continue
+                    resp = self._session.get(
+                        url, headers=headers, params=params, timeout=30)
+            except (requests.exceptions.ConnectionError,
+                    requests.exceptions.Timeout) as e:
+                raise EnsemblConnectionError(
+                    f"Ensembl REST request to {endpoint} failed: {e}") from e
+            except requests.exceptions.RequestException as e:
+                # Includes RetryError raised when urllib3 exhausts 5xx retries.
+                raise EnsemblRESTError(
+                    f"Ensembl REST request to {endpoint} failed: {e}") from e
+
+            if resp.status_code == 429:
+                retry_after = float(resp.headers.get("Retry-After", 1.0))
+                logging.warning(
+                    "Rate limited by Ensembl REST API, sleeping %.1fs", retry_after)
+                time.sleep(retry_after)
+                continue
 
-                if resp.status_code == 400:
-                    logging.warning("Bad request to Ensembl REST API: %s", endpoint)
-                    return None
+            if resp.status_code in (400, 404):
+                logging.warning("Ensembl REST API: no entry for %s (HTTP %d)",
+                                endpoint, resp.status_code)
+                return None
 
-                resp.raise_for_status()
+            if not resp.ok:
+                raise EnsemblRESTError(
+                    f"Ensembl REST API error {resp.status_code} for {endpoint}")
 
-                if content_type == 'text/plain':
-                    return resp.text
+            if content_type == 'text/plain':
+                return resp.text
+            try:
                 return resp.json()
+            except ValueError:
+                return None  # empty/invalid 2xx body -> 'no data'
 
-            except requests.exceptions.RequestException as e:
-                logging.warning("Ensembl REST API request failed: %s", e)
-                return None
-
-        logging.warning("Ensembl REST API rate limit retries exhausted for %s", endpoint)
-        return None
+        raise EnsemblRateLimitError(
+            f"Ensembl REST API rate limit retries exhausted for {endpoint}")
 
     def get_product_sequence(self, product_id, **kwargs):
         """
diff --git a/epytope/test/TestEnsemblRESTAdapter.py b/epytope/test/TestEnsemblRESTAdapter.py
@@ -0,0 +1,163 @@
+from unittest import TestCase
+from unittest import mock
+
+import requests
+
+from epytope.IO.EnsemblRESTAdapter import (
+    EnsemblRESTAdapter,
+    EnsemblRESTError,
+    EnsemblRateLimitError,
+    EnsemblConnectionError,
+    _RateLimiter,
+)
+
+
+class TestRateLimiter(TestCase):
+    def test_blocks_when_window_exceeded(self):
+        clock = {"t": 1000.0}
+        sleeps = []
+
+        def fake_sleep(secs):
+            sleeps.append(secs)
+            clock["t"] += secs
+
+        limiter = _RateLimiter([(2, 10.0)])
+        with mock.patch("time.monotonic",
+                        lambda: clock["t"]), \
+             mock.patch("time.sleep", fake_sleep):
+            limiter.acquire()   # 1st call: fits
+            limiter.acquire()   # 2nd call: fits
+            limiter.acquire()   # 3rd call: window full -> must wait 10s
+
+        self.assertEqual(sleeps, [10.0])
+
+    def test_no_block_under_limit(self):
+        clock = {"t": 500.0}
+        sleeps = []
+        limiter = _RateLimiter([(5, 1.0)])
+        with mock.patch("time.monotonic",
+                        lambda: clock["t"]), \
+             mock.patch("time.sleep",
+                        lambda s: sleeps.append(s)):
+            for _ in range(5):
+                limiter.acquire()
+        self.assertEqual(sleeps, [])
+
+
+class FakeResponse:
+    def __init__(self, status_code=200, json_data=None, text="", headers=None,
+                 json_error=False):
+        self.status_code = status_code
+        self._json = json_data
+        self.text = text
+        self.headers = headers or {}
+        self._json_error = json_error
+
+    @property
+    def ok(self):
+        return self.status_code < 400
+
+    def json(self):
+        if self._json_error:  # mimic requests' JSONDecodeError (a ValueError) on empty body
+            raise ValueError("No JSON could be decoded")
+        return self._json
+
+
+class FakeSession:
+    """Returns queued FakeResponses (or raises queued exceptions) per call.
+
+    If more than one outcome remains it pops the next; the last outcome
+    repeats indefinitely (so a single 429 means 'always 429').
+    """
+
+    def __init__(self, outcomes):
+        self._outcomes = list(outcomes)
+        self.calls = []
+
+    def _next(self, *args, **kwargs):
+        self.calls.append((args, kwargs))
+        outcome = self._outcomes.pop(0) if len(self._outcomes) > 1 else self._outcomes[0]
+        if isinstance(outcome, Exception):
+            raise outcome
+        return outcome
+
+    def get(self, *args, **kwargs):
+        return self._next(*args, **kwargs)
+
+    def post(self, *args, **kwargs):
+        return self._next(*args, **kwargs)
+
+
+def _adapter_with(outcomes):
+    adapter = EnsemblRESTAdapter()
+    adapter._session = FakeSession(outcomes)
+    return adapter
+
+
+class TestRequestFailureMapping(TestCase):
+    def setUp(self):
+        # 429 path sleeps on Retry-After; patch it so tests stay fast.
+        self._sleep_patch = mock.patch(
+            "time.sleep", lambda s: None)
+        self._sleep_patch.start()
+
+    def tearDown(self):
+        self._sleep_patch.stop()
+
+    def test_429_exhausted_raises_ratelimit(self):
+        adapter = _adapter_with([FakeResponse(429, headers={"Retry-After": "0"})])
+        with self.assertRaises(EnsemblRateLimitError):
+            adapter._request("/lookup/id/ENST1")
+
+    def test_connection_error_raises_connection(self):
+        adapter = _adapter_with([requests.exceptions.ConnectionError("boom")])
+        with self.assertRaises(EnsemblConnectionError):
+            adapter._request("/lookup/id/ENST1")
+
+    def test_timeout_raises_connection(self):
+        adapter = _adapter_with([requests.exceptions.Timeout("slow")])
+        with self.assertRaises(EnsemblConnectionError):
+            adapter._request("/lookup/id/ENST1")
+
+    def test_retryerror_raises_base_rest_error(self):
+        adapter = _adapter_with([requests.exceptions.RetryError("5xx exhausted")])
+        with self.assertRaises(EnsemblRESTError) as ctx:
+            adapter._request("/lookup/id/ENST1")
+        self.assertNotIsInstance(
+            ctx.exception, (EnsemblRateLimitError, EnsemblConnectionError))
+
+    def test_non_retryable_error_status_raises_base_rest_error(self):
+        # 403 (non-2xx, non-429, non-404) -> base error, not a subtype
+        adapter = _adapter_with([FakeResponse(403)])
+        with self.assertRaises(EnsemblRESTError) as ctx:
+            adapter._request("/lookup/id/FORBIDDEN")
+        self.assertNotIsInstance(
+            ctx.exception, (EnsemblRateLimitError, EnsemblConnectionError))
+
+    def test_200_unparseable_json_returns_none(self):
+        adapter = _adapter_with([FakeResponse(200, json_error=True)])
+        self.assertIsNone(adapter._request("/lookup/id/EMPTY"))
+
+    def test_400_returns_none(self):
+        adapter = _adapter_with([FakeResponse(400)])
+        self.assertIsNone(adapter._request("/lookup/id/BAD"))
+
+    def test_404_returns_none(self):
+        adapter = _adapter_with([FakeResponse(404)])
+        self.assertIsNone(adapter._request("/lookup/id/MISSING"))
+
+    def test_200_json_returns_parsed(self):
+        adapter = _adapter_with([FakeResponse(200, json_data={"id": "ENST1"})])
+        self.assertEqual(adapter._request("/lookup/id/ENST1"), {"id": "ENST1"})
+
+    def test_200_text_returns_text(self):
+        adapter = _adapter_with([FakeResponse(200, text="MEEPQS")])
+        self.assertEqual(
+            adapter._request("/sequence/id/ENSP1", content_type="text/plain"), "MEEPQS")
+
+    def test_429_then_success(self):
+        adapter = _adapter_with([
+            FakeResponse(429, headers={"Retry-After": "0"}),
+            FakeResponse(200, json_data={"id": "ENST1"}),
+        ])
+        self.assertEqual(adapter._request("/lookup/id/ENST1"), {"id": "ENST1"})
