From 8e1e47717350488dd88adec9b3f73de6f5464661 Mon Sep 17 00:00:00 2001 From: Scorpion197 Date: Wed, 16 Sep 2026 20:53:37 +0200 Subject: [PATCH] fix(client): keep --all crawls inside the API rate limit A --all crawl fires up to 1,000 pages at one route back-to-back. The route bucket holds 10 requests and refills at 1/s, so the eleventh page gets a 429 and, without --retry, the crawl dies and every page already fetched is thrown away. Any export past ~1,000 records failed by design. The pagination guide asks clients to add delays between pages and the rate-limiting guide says every response carries the binding bucket's X-RateLimit-* headers, so the crawl now uses them: - After a page that empties the bucket, wait for one token before asking for the next page (Reset / Limit, or the full reset without a Limit). - A 429 mid-crawl pauses and re-requests the same page instead of failing, with or without --retry: a GET is safe to repeat and earlier pages are already in hand. The wait follows Retry-After, then the bucket reset, then the backoff table; past the 60s ceiling or once the table runs out the 429 is reported as before. - Pauses are announced on stderr so a long export does not look hung. Single-page requests are unchanged. The retry loop is split into _send, which returns the last response, so the crawl can read headers without duplicating it. https://docs.dualentry.com/developers/guides/rate-limiting https://docs.dualentry.com/developers/guides/pagination --- src/dualentry_cli/client.py | 104 +++++++++++++++-- tests/test_client.py | 227 ++++++++++++++++++++++++++++++++++++ 2 files changed, 324 insertions(+), 7 deletions(-) diff --git a/src/dualentry_cli/client.py b/src/dualentry_cli/client.py index da16788..c03ab30 100644 --- a/src/dualentry_cli/client.py +++ b/src/dualentry_cli/client.py @@ -2,6 +2,7 @@ from __future__ import annotations +import math import os import sys import time @@ -9,6 +10,7 @@ from typing import Any import httpx +import typer from dualentry_cli import USER_AGENT @@ -33,6 +35,14 @@ _MAX_RETRY_AFTER = 60 +# Every response reports the binding token bucket. A route bucket holds 10 +# requests and refills at 1/s, so a crawl that fires pages back-to-back drains +# it by the eleventh page; the guide asks clients to slow down before that. +# https://docs.dualentry.com/developers/guides/rate-limiting +_RATE_LIMIT_LIMIT_HEADER = "X-RateLimit-Limit" +_RATE_LIMIT_REMAINING_HEADER = "X-RateLimit-Remaining" +_RATE_LIMIT_RESET_HEADER = "X-RateLimit-Reset" + # Hard ceiling for --all crawls: page_size 100 x 1000 pages = 100_000 items. # Truncation must warn; never report the truncated length as the API total. _MAX_PAGES = 1000 @@ -51,6 +61,50 @@ def _retry_after_seconds(response: httpx.Response) -> int | None: return seconds if seconds >= 0 else None +def _bucket_is_empty(response: httpx.Response) -> bool: + """Whether X-RateLimit-Remaining says the next request would be rejected.""" + raw = response.headers.get(_RATE_LIMIT_REMAINING_HEADER) + if raw is None: + return False + try: + return int(raw.strip()) <= 0 + except (TypeError, ValueError): + return False + + +def _seconds_until_reset(response: httpx.Response) -> int | None: + """Whole seconds until X-RateLimit-Reset (a Unix timestamp), or None if absent, unusable or already past.""" + raw = response.headers.get(_RATE_LIMIT_RESET_HEADER) + if raw is None: + return None + try: + reset_at = int(raw.strip()) + except (TypeError, ValueError): + return None + wait = math.ceil(reset_at - time.time()) + return wait if wait > 0 else None + + +def _seconds_until_next_token(response: httpx.Response) -> int | None: + """ + Whole seconds until one more request is admitted, or None if the headers do not say. + + Reset is when the bucket is full again and Limit is its size, so a single + token is back after that span divided by the limit. Without a usable Limit + the only safe wait is the full reset. + """ + until_full = _seconds_until_reset(response) + if until_full is None: + return None + try: + limit = int(response.headers.get(_RATE_LIMIT_LIMIT_HEADER, "").strip()) + except ValueError: + return until_full + if limit <= 0: + return until_full + return math.ceil(until_full / limit) + + def _is_retryable(response: httpx.Response) -> bool: """ Whether this response should be retried with the same idempotency key. @@ -180,10 +234,13 @@ def _request(self, method: str, path: str, **kwargs) -> dict: headers = dict(kwargs.pop("headers", None) or {}) headers.setdefault(_IDEMPOTENCY_HEADER, str(uuid.uuid4())) kwargs["headers"] = headers + response = self._send(method, path, **kwargs) + return self._handle_response(response, sent_idempotency_key=keyed) + def _send(self, method: str, path: str, **kwargs) -> httpx.Response: + """Send one logical request, applying --retry if enabled, and return the last response.""" if not self._retry: - response = self._client.request(method, path, **kwargs) - return self._handle_response(response, sent_idempotency_key=keyed) + return self._client.request(method, path, **kwargs) # Retry logic with visible feedback for attempt, backoff in enumerate(_RETRY_DELAYS): @@ -191,10 +248,10 @@ def _request(self, method: str, path: str, **kwargs) -> dict: try: response = self._client.request(method, path, **kwargs) if not _is_retryable(response): - return self._handle_response(response, sent_idempotency_key=keyed) + return response retry_after = _retry_after_seconds(response) if retry_after is not None and retry_after > _MAX_RETRY_AFTER: - return self._handle_response(response, sent_idempotency_key=keyed) + return response except _RETRYABLE_EXCEPTIONS: pass @@ -204,8 +261,34 @@ def _request(self, method: str, path: str, **kwargs) -> dict: time.sleep(delay) # Final attempt - response = self._client.request(method, path, **kwargs) - return self._handle_response(response, sent_idempotency_key=keyed) + return self._client.request(method, path, **kwargs) + + def _fetch_page(self, path: str, params: dict[str, Any]) -> httpx.Response: + """ + Fetch one page of a crawl, pausing on 429 instead of failing. + + A GET is safe to repeat and the pages before it are already in hand, so + a throttled page is waited out and asked for again. This runs whether + or not --retry is set. The wait comes from Retry-After, then from the + bucket reset time, then from the backoff table; past the ceiling, or + once the table runs out, the 429 is reported like any other error. + """ + for backoff in _RETRY_DELAYS: + response = self._send("GET", path, params=params) + if response.status_code != 429: + return response + wait = _retry_after_seconds(response) + if wait is None: + wait = _seconds_until_reset(response) + if wait is not None and wait > _MAX_RETRY_AFTER: + return response + self._pause(wait if wait is not None else backoff) + return self._send("GET", path, params=params) + + @staticmethod + def _pause(seconds: int) -> None: + typer.secho(f"Rate limit reached; pausing {seconds}s before the next page...", fg=typer.colors.YELLOW, err=True) + time.sleep(seconds) def get(self, path: str, params: dict[str, Any] | None = None) -> dict: return self._request("GET", path, params=params) @@ -233,7 +316,8 @@ def paginate( total = 0 for _ in range(_MAX_PAGES): - data = self.get(path, params=params) + response = self._fetch_page(path, params) + data = self._handle_response(response) items = data.get("items", []) all_items.extend(items) total = data.get("count", start_offset + len(all_items)) @@ -243,6 +327,12 @@ def paginate( if start_offset + len(all_items) >= total or not items: break params["offset"] += page_size + # Another page follows: if this one emptied the bucket, wait for a token + # rather than spend a request on a 429. + if _bucket_is_empty(response): + wait = _seconds_until_next_token(response) + if wait is not None: + self._pause(min(wait, _MAX_RETRY_AFTER)) result: dict[str, Any] = {"items": all_items, "count": total} if start_offset + len(all_items) < total: diff --git a/tests/test_client.py b/tests/test_client.py index aa552fa..a9364cf 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -563,3 +563,230 @@ def test_max_items_truncation_sets_next_offset(self): assert data["items"] == [{"id": 1}, {"id": 2}] assert data["count"] == 9 assert data["next_offset"] == 2 + + +class TestPaginateRateLimits: + """ + A --all crawl must respect the per-route token bucket instead of dying on it. + + https://docs.dualentry.com/developers/guides/rate-limiting + https://docs.dualentry.com/developers/guides/pagination + """ + + BASE = "https://api.dualentry.com/public/v2" + NOW = 1_000_000.0 + + @pytest.fixture + def sleeps(self, monkeypatch): + """Record what the client would sleep and pin the clock, without actually sleeping.""" + recorded = [] + monkeypatch.setattr("dualentry_cli.client.time", SimpleNamespace(sleep=recorded.append, time=lambda: self.NOW)) + return recorded + + @staticmethod + def _client(): + from dualentry_cli.client import DualEntryClient + + # retry=False on purpose: a crawl must survive throttling without the global flag. + return DualEntryClient(api_url="https://api.dualentry.com", api_key="test_key", retry=False) + + @staticmethod + def _offsets(route) -> list[str]: + return [c.request.url.params.get("offset", "0") for c in route.calls] + + @respx.mock + def test_remaining_zero_waits_for_one_token_before_the_next_page(self, sleeps): + """ + The guide asks clients to slow down preemptively. Reset is when the bucket is full + again and Limit is its size, so one token is back after Reset / Limit. + """ + drained = {"X-RateLimit-Limit": "10", "X-RateLimit-Remaining": "0", "X-RateLimit-Reset": str(int(self.NOW) + 40)} + respx.get(f"{self.BASE}/invoices/").mock( + side_effect=[ + httpx.Response(200, headers=drained, json={"items": [{"id": 1}, {"id": 2}], "count": 3}), + httpx.Response(200, json={"items": [{"id": 3}], "count": 3}), + ] + ) + + data = self._client().paginate("/invoices/", page_size=2) + + assert data["items"] == [{"id": 1}, {"id": 2}, {"id": 3}] + assert sleeps == [4], "40s to refill 10 tokens is 4s per token" + + @respx.mock + def test_remaining_zero_without_a_limit_waits_for_the_full_reset(self, sleeps): + drained = {"X-RateLimit-Remaining": "0", "X-RateLimit-Reset": str(int(self.NOW) + 4)} + respx.get(f"{self.BASE}/invoices/").mock( + side_effect=[ + httpx.Response(200, headers=drained, json={"items": [{"id": 1}], "count": 2}), + httpx.Response(200, json={"items": [{"id": 2}], "count": 2}), + ] + ) + + self._client().paginate("/invoices/", page_size=1) + + assert sleeps == [4], "no bucket size, so the only safe wait is the full refill" + + @respx.mock + def test_remaining_above_zero_does_not_wait(self, sleeps): + headers = {"X-RateLimit-Remaining": "3", "X-RateLimit-Reset": str(int(self.NOW) + 9)} + respx.get(f"{self.BASE}/invoices/").mock( + side_effect=[ + httpx.Response(200, headers=headers, json={"items": [{"id": 1}], "count": 2}), + httpx.Response(200, headers=headers, json={"items": [{"id": 2}], "count": 2}), + ] + ) + + self._client().paginate("/invoices/", page_size=1) + + assert sleeps == [] + + @respx.mock + def test_missing_rate_limit_headers_behave_as_before(self, sleeps): + respx.get(f"{self.BASE}/invoices/").mock( + side_effect=[ + httpx.Response(200, json={"items": [{"id": 1}], "count": 2}), + httpx.Response(200, json={"items": [{"id": 2}], "count": 2}), + ] + ) + + data = self._client().paginate("/invoices/", page_size=1) + + assert len(data["items"]) == 2 + assert sleeps == [] + + @respx.mock + def test_no_wait_after_the_last_page(self, sleeps): + """Nothing follows the final page, so an empty bucket there must not delay the result.""" + drained = {"X-RateLimit-Remaining": "0", "X-RateLimit-Reset": str(int(self.NOW) + 5)} + respx.get(f"{self.BASE}/invoices/").mock(return_value=httpx.Response(200, headers=drained, json={"items": [{"id": 1}], "count": 1})) + + self._client().paginate("/invoices/", page_size=1) + + assert sleeps == [] + + @pytest.mark.parametrize("reset", ["soon", "", "2.5", str(int(NOW) - 30)]) + @respx.mock + def test_unusable_or_past_reset_does_not_wait(self, sleeps, reset): + drained = {"X-RateLimit-Remaining": "0", "X-RateLimit-Reset": reset} + respx.get(f"{self.BASE}/invoices/").mock( + side_effect=[ + httpx.Response(200, headers=drained, json={"items": [{"id": 1}], "count": 2}), + httpx.Response(200, json={"items": [{"id": 2}], "count": 2}), + ] + ) + + self._client().paginate("/invoices/", page_size=1) + + assert sleeps == [], f"reset {reset!r} gives nothing to wait for" + + @respx.mock + def test_preemptive_wait_is_capped_at_the_ceiling(self, sleeps): + from dualentry_cli.client import _MAX_RETRY_AFTER + + drained = {"X-RateLimit-Remaining": "0", "X-RateLimit-Reset": str(int(self.NOW) + 3600)} + respx.get(f"{self.BASE}/invoices/").mock( + side_effect=[ + httpx.Response(200, headers=drained, json={"items": [{"id": 1}], "count": 2}), + httpx.Response(200, json={"items": [{"id": 2}], "count": 2}), + ] + ) + + self._client().paginate("/invoices/", page_size=1) + + assert sleeps == [_MAX_RETRY_AFTER], "never sleep an hour on a header; let the server say 429 if it must" + + @respx.mock + def test_throttled_page_is_refetched_after_retry_after_without_the_retry_flag(self, sleeps): + """A GET is safe to repeat and earlier pages are in hand, so a 429 pauses the crawl instead of ending it.""" + route = respx.get(f"{self.BASE}/invoices/").mock( + side_effect=[ + httpx.Response(200, json={"items": [{"id": 1}, {"id": 2}], "count": 3}), + httpx.Response(429, headers={"Retry-After": "3"}, json={"errors": {"__all__": ["slow down"]}}), + httpx.Response(200, json={"items": [{"id": 3}], "count": 3}), + ] + ) + + data = self._client().paginate("/invoices/", page_size=2) + + assert data["items"] == [{"id": 1}, {"id": 2}, {"id": 3}] + assert "next_offset" not in data + assert sleeps == [3] + assert self._offsets(route) == ["0", "2", "2"], "the throttled page is asked for again, not skipped" + + @respx.mock + def test_throttled_page_without_retry_after_waits_for_the_reset(self, sleeps): + headers = {"X-RateLimit-Remaining": "0", "X-RateLimit-Reset": str(int(self.NOW) + 6)} + respx.get(f"{self.BASE}/invoices/").mock( + side_effect=[ + httpx.Response(429, headers=headers, json={}), + httpx.Response(200, json={"items": [{"id": 1}], "count": 1}), + ] + ) + + data = self._client().paginate("/invoices/", page_size=1) + + assert data["items"] == [{"id": 1}] + assert sleeps == [6] + + @respx.mock + def test_throttled_page_with_no_headers_backs_off_then_gives_up(self, sleeps): + """Without any timing from the server, use the backoff table, and stop when it runs out.""" + from dualentry_cli.client import _RETRY_DELAYS, APIError + + route = respx.get(f"{self.BASE}/invoices/").mock(return_value=httpx.Response(429, json={})) + + with pytest.raises(APIError) as exc: + self._client().paginate("/invoices/", page_size=1) + + assert exc.value.status_code == 429 + assert sleeps == _RETRY_DELAYS + assert route.call_count == len(_RETRY_DELAYS) + 1 + + @respx.mock + def test_throttled_page_beyond_the_ceiling_is_reported_not_slept_through(self, sleeps): + from dualentry_cli.client import APIError + + route = respx.get(f"{self.BASE}/invoices/").mock(return_value=httpx.Response(429, headers={"Retry-After": "3600"}, json={})) + + with pytest.raises(APIError) as exc: + self._client().paginate("/invoices/", page_size=1) + + assert route.call_count == 1 + assert sleeps == [] + assert "3600" in exc.value.detail + + @respx.mock + def test_other_errors_mid_crawl_still_raise_at_once(self, sleeps): + from dualentry_cli.client import APIError + + route = respx.get(f"{self.BASE}/invoices/").mock( + side_effect=[ + httpx.Response(200, json={"items": [{"id": 1}], "count": 2}), + httpx.Response(500, json={}), + ] + ) + + with pytest.raises(APIError) as exc: + self._client().paginate("/invoices/", page_size=1) + + assert exc.value.status_code == 500 + assert route.call_count == 2 + assert sleeps == [] + + @pytest.mark.usefixtures("sleeps") + @respx.mock + def test_pause_is_announced_on_stderr(self, capsys): + respx.get(f"{self.BASE}/invoices/").mock( + side_effect=[ + httpx.Response(429, headers={"Retry-After": "2"}, json={}), + httpx.Response(200, json={"items": [{"id": 1}], "count": 1}), + ] + ) + + self._client().paginate("/invoices/", page_size=1) + + captured = capsys.readouterr() + assert captured.out == "", "stdout is reserved for the data" + assert "Rate limit reached" in captured.err + assert "pausing 2s" in captured.err