From 660c4698cc42464c1c771fc9af28c829a4e3c7e7 Mon Sep 17 00:00:00 2001 From: gonzalo-zenrows Date: Fri, 16 Oct 2026 19:03:07 +0200 Subject: [PATCH 1/6] feat(client): retry only failures another try can fix, with jittered backoff A cold Extract request can briefly answer 503 with "retry in a minute, not charged", and many clients retrying on the same fixed schedule hit the API again in lockstep. The client now retries by default and spreads its retries out. - retries now defaults to 2 (was 0); retries=0 turns retries off. - Retried: 429, 502, 503, 504 (none of them is charged) and a failure to connect. No longer retried: 500 and 422. Never retried: a read timeout, because the server may still finish, and charge, the request. - Waits honour Retry-After, capped at retry_max_wait (default 30 s), or else a random 0.25 s to min(retry_max_wait, 2 ** attempt) s. - retry_budget (default 90 s) stops retrying when the next wait would end that long after the call's first try. - The separate 422 retry loop in _worker is gone: one retry layer. - urllib3 >= 1.26 is now declared, for Retry(allowed_methods=...). --- README.md | 2 +- pyproject.toml | 3 + src/zenrows/client.py | 183 +++++++++++++--- tests/test_retries.py | 482 +++++++++++++++++++++++++++++++++++------- uv.lock | 2 + 5 files changed, 562 insertions(+), 110 deletions(-) diff --git a/README.md b/README.md index 68a15b1..edc99a7 100644 --- a/README.md +++ b/README.md @@ -57,7 +57,7 @@ Start using the API by [creating your API Key](https://www.zenrows.com/register? The SDK uses [requests](https://docs.python-requests.org/) for HTTP requests. The client's response will be a requests `Response`. -It also uses [Retry](https://urllib3.readthedocs.io/en/latest/reference/urllib3.util.html) to automatically retry failed requests (status codes 422, 429, 500, 502, 503, and 504, except Extract's `422 no_data_extracted`). Retries are not active by default; you need to specify the number of retries, as shown below. It already includes an exponential back-off retry delay between failed requests. +The client retries a request up to 2 times by default (`retries=2`), and only when another try can help: a `429`, `502`, `503` or `504` response, none of which is charged, or a failure to connect. It never retries any other status (including any `422`) or a read timeout, because the server may still be working on that request. Between tries it waits the server's `Retry-After` header, or else a random time between 0.25 seconds and `2 ** attempt` seconds (full jitter); either wait is capped at `retry_max_wait` (default 30 seconds). It starts no retry whose wait would end more than `retry_budget` seconds (default 90) after the first try. Pass `retries=0` to turn retries off. `client.fetch()` is the primary method for the main page-scraping product; `client.get()` still works and is kept as a deprecated alias. diff --git a/pyproject.toml b/pyproject.toml index 63bc1bb..49e9490 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -28,6 +28,9 @@ requires-python = ">=3.10" dependencies = [ # Legacy scraping client (zenrows.ZenRowsClient). "requests>=2.31", + # The client's retry policy subclasses urllib3's Retry; `allowed_methods` + # needs 1.26. + "urllib3>=1.26", # Batch (zenrows.batch) — pydantic v2 models + httpx transport. "httpx>=0.27", "pydantic>=2.9", diff --git a/src/zenrows/client.py b/src/zenrows/client.py index 2a38502..fc74b31 100644 --- a/src/zenrows/client.py +++ b/src/zenrows/client.py @@ -12,7 +12,9 @@ """ import asyncio +import contextvars import os +import random import time from concurrent.futures import ThreadPoolExecutor from functools import partial @@ -21,6 +23,7 @@ import requests import urllib3 from requests.adapters import HTTPAdapter +from urllib3.exceptions import InvalidHeader, MaxRetryError, ResponseError from urllib3.util.retry import Retry from zenrows.__version__ import __version__ @@ -28,11 +31,129 @@ DEFAULT_SCRAPER_URL = "https://api.zenrows.com/v1/" DEFAULT_USER_AGENT = f"zenrows/{__version__} python" -# Status codes the urllib3 retry layer treats as transient. 422 is retried -# too, but in `_worker`, because whether it is worth retrying depends on the -# response body (see `is_no_data_extracted`). -_RETRY_STATUSES = [429, 500, 502, 503, 504] -_RETRY_BACKOFF_FACTOR = 0.5 +# Status codes worth another try. None of them is charged: 429 is a rate +# limit, 503 is the gateway asking to retry later, 502 and 504 never reached +# the target. Every other 4xx/5xx is the final answer. +_RETRY_STATUSES = (429, 502, 503, 504) +DEFAULT_RETRIES = 2 +# Longest wait between two tries, in seconds: the cap on `Retry-After` and on +# the jittered backoff. Same cap as the Crawl client's `stop()` retry. +DEFAULT_RETRY_MAX_WAIT = 30.0 +# Longest a call may keep retrying, in seconds, counted from its first try: +# no retry starts if the wait before it would end past this. +DEFAULT_RETRY_BUDGET = 90.0 +# Shortest jittered wait, so a retry never fires back to back. +_RETRY_MIN_WAIT = 0.25 + +# Clock for the retry budget; tests replace it. +_monotonic = time.monotonic + +# When the current call's first try started. Set per call by +# `_RetryBudgetAdapter.send`; a context variable, so concurrent calls on +# the client's thread pool each see their own. +_call_started: contextvars.ContextVar[float | None] = contextvars.ContextVar( + "zenrows_call_started", default=None +) + + +class _JitteredRetry(Retry): + """urllib3 `Retry` with the Zenrows client retry policy. + + Between tries it waits the server's `Retry-After` (delta-seconds or + HTTP-date), capped at `retry_max_wait`; without one it waits a random time + in `[0.25, min(retry_max_wait, 2 ** attempt)]` seconds (full jitter, + `attempt` counting from 0). It never starts a retry whose wait would end + more than `retry_budget` seconds after the call's first try. + + Only the hooks urllib3 1.26 and 2.x share are overridden, so it works on + both (no `backoff_jitter` / `retry_after_max`). + """ + + # urllib3 also retries 413 when it carries `Retry-After`; we don't. + RETRY_AFTER_STATUS_CODES = frozenset({429, 503}) + + def __init__( + self, + *args: Any, + retry_max_wait: float = DEFAULT_RETRY_MAX_WAIT, + retry_budget: float = DEFAULT_RETRY_BUDGET, + **kwargs: Any, + ): + super().__init__(*args, **kwargs) + self.retry_max_wait = retry_max_wait + self.retry_budget = retry_budget + self._started: float | None = None + # The wait before the next try, drawn once in `increment` so the + # budget check and the sleep agree on it. + self._delay: float | None = None + + def new(self, **kw: Any) -> "_JitteredRetry": + retry = super().new(**kw) + retry.retry_max_wait = self.retry_max_wait + retry.retry_budget = self.retry_budget + started = self._started + if started is None: + started = _call_started.get() + if started is None: + started = _monotonic() + retry._started = started + return retry + + def increment( + self, + method: str | None = None, + url: str | None = None, + response: Any = None, + error: Exception | None = None, + _pool: Any = None, + _stacktrace: Any = None, + ) -> "_JitteredRetry": + retry = super().increment(method, url, response, error, _pool, _stacktrace) + retry._delay = retry._next_delay(response) + started = retry._started if retry._started is not None else _monotonic() + if _monotonic() - started + retry._delay > retry.retry_budget: + reason = error or ResponseError(f"retry budget of {retry.retry_budget}s spent") + raise MaxRetryError(_pool, url, reason) from reason + return retry + + def get_backoff_time(self) -> float: + if self._delay is not None: + return self._delay + return self._jitter() + + def get_retry_after(self, response: Any) -> float | None: + try: + retry_after = super().get_retry_after(response) + except InvalidHeader: + return None # unparseable: fall back to the jittered backoff + if retry_after is None: + return None + return min(retry_after, self.retry_max_wait) + + def _next_delay(self, response: Any) -> float: + # Mirrors `Retry.sleep`: a positive `Retry-After` wins over the backoff. + if response is not None and self.respect_retry_after_header: + retry_after = self.get_retry_after(response) + if retry_after: + return retry_after + return self._jitter() + + def _jitter(self) -> float: + attempt = min(max(len(self.history) - 1, 0), 30) + ceiling = min(self.retry_max_wait, 2.0**attempt) + return random.uniform(min(_RETRY_MIN_WAIT, ceiling), ceiling) + + +class _RetryBudgetAdapter(HTTPAdapter): + """Marks when each call's first try starts, for the retry budget.""" + + def send(self, request: Any, *args: Any, **kwargs: Any) -> requests.Response: + token = _call_started.set(_monotonic()) + try: + return super().send(request, *args, **kwargs) + finally: + _call_started.reset(token) + # Extract's answer when it knows the kind of page but none of the fields came # back filled. Retrying returns the same answer, so it is never retried. @@ -105,6 +226,13 @@ class ZenRowsClient: `base_url` defaults to the public production endpoint; override via constructor arg or `ZENROWS_SCRAPER_BASE_URL` env var. + + `retries` (default 2) retries a 429, 502, 503 or 504, and a failure to + connect, never any other status or a read timeout. Between tries it waits + the server's `Retry-After`, or else a random 0.25s to `2 ** attempt` + seconds; either is capped at `retry_max_wait` (default 30). No retry starts + if its wait would end more than `retry_budget` seconds (default 90) after + the first try. Pass `retries=0` to turn retries off. """ # Kept as a class attribute for compatibility — pre-1.4 callers @@ -114,10 +242,12 @@ class ZenRowsClient: def __init__( self, apikey: str, - retries: int = 0, + retries: int = DEFAULT_RETRIES, concurrency: int = 5, *, base_url: str | None = None, + retry_max_wait: float = DEFAULT_RETRY_MAX_WAIT, + retry_budget: float = DEFAULT_RETRY_BUDGET, ): if not apikey: raise ValueError("ZenRowsClient: apikey is required.") @@ -127,14 +257,25 @@ def __init__( self.retries = retries self.executor = ThreadPoolExecutor(max_workers=concurrency) self.requests_session = requests.Session() + self.retry_max_wait = retry_max_wait + self.retry_budget = retry_budget if retries > 0: - max_retries = Retry( + # Retries a 429/502/503/504 and a failure to connect, nothing + # else: a read error or timeout may come after the server + # accepted the request, which keeps running and may be charged. + max_retries = _JitteredRetry( total=retries, - backoff_factor=_RETRY_BACKOFF_FACTOR, + connect=retries, + read=False, # re-raise as is, so it stays a ReadTimeout + status=retries, + other=0, + allowed_methods=None, status_forcelist=_RETRY_STATUSES, raise_on_status=False, + retry_max_wait=retry_max_wait, + retry_budget=retry_budget, ) - adapter = HTTPAdapter(max_retries=max_retries) + adapter = _RetryBudgetAdapter(max_retries=max_retries) self.requests_session.mount("https://", adapter) self.requests_session.mount("http://", adapter) @@ -354,7 +495,7 @@ def _worker( final_headers["Connection"] = None final_headers.update(headers) - response = self.requests_session.request( + return self.requests_session.request( method, self.api_url, params=final_params, @@ -362,25 +503,3 @@ def _worker( data=data, **kwargs, ) - # Retry a 422 here rather than in urllib3, which can't read the body: - # Extract's `no_data_extracted` is a final answer, any other 422 is a - # failure to get the target's content and is worth another try. - attempt = 0 - while ( - response.status_code == 422 - and attempt < self.retries - and not is_no_data_extracted(response) - ): - attempt += 1 - # Same schedule as urllib3's Retry: no wait before the first retry. - if attempt > 1: - time.sleep(_RETRY_BACKOFF_FACTOR * 2 ** (attempt - 1)) - response = self.requests_session.request( - method, - self.api_url, - params=final_params, - headers=final_headers, - data=data, - **kwargs, - ) - return response diff --git a/tests/test_retries.py b/tests/test_retries.py index 0fb37d5..7d39ae9 100644 --- a/tests/test_retries.py +++ b/tests/test_retries.py @@ -1,114 +1,442 @@ +"""The ZenRowsClient retry policy, end to end against a local HTTP server. + +Retried: 429, 502, 503, 504 and a failure to connect. Never retried: any +other status (422 included) or a read timeout. Waits honour `Retry-After` +(capped at `retry_max_wait`) or else use full jitter, and no retry starts +past `retry_budget`. Sleeps are patched out; the budget runs on a fake clock. +""" + import asyncio -from unittest import TestCase +import socket +import threading +from collections.abc import Iterator +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from unittest.mock import patch +from urllib.parse import parse_qs, urlparse -from requests import Response, Session +import pytest +import requests +from requests.adapters import HTTPAdapter from zenrows import ZenRowsClient, is_no_data_extracted -from zenrows.client import _RETRY_STATUSES +from zenrows.client import _JitteredRetry -apikey = "APIKEY" -url = "http://example.com" -api_url_base = "https://api.zenrows.com/v1/" +APIKEY = "APIKEY" +TARGET = "http://example.com" +NO_DATA = b'{"code": "no_data_extracted", "title": "No data extracted"}' +NO_CONTENT = b'{"code": "RESP001", "title": "Could not get content"}' +DOMAIN_NOT_PREPARED = b'{"code": "REQS007", "title": "Domain not prepared"}' -class TestZenRowsClientRetries(TestCase): - @patch("zenrows.client.Retry") - @patch.object(Session, "mount") - def test_custom_session_not_initiated(self, mock_mount, mock_retry): - ZenRowsClient(apikey, retries=0) +# A scripted response: (status, headers, body). HANG answers only after the +# client's read timeout has passed. +HANG = (0, {}, b"") - mock_retry.assert_not_called() - self.assertEqual(mock_mount.call_count, 2) # called internally - @patch("zenrows.client.Retry") - def test_retry_parameters(self, mock_retry): - ZenRowsClient(apikey, retries=2) +class _Server: + """Answers each request with the next scripted response (the last one + repeats) and records the query of every request it got.""" - mock_retry.assert_called_once_with( - total=2, - backoff_factor=0.5, - status_forcelist=list(_RETRY_STATUSES), - raise_on_status=False, - ) + def __init__(self) -> None: + self.script: list[tuple[int, dict[str, str], bytes]] = [(200, {}, b"ok")] + self.queries: list[dict[str, list[str]]] = [] + self.on_request = lambda: None + server = self - @patch.object(Session, "mount") - def test_adapter_is_mount(self, mock_mount): - ZenRowsClient(apikey, retries=2) + class Handler(BaseHTTPRequestHandler): + def _answer(self) -> None: + length = int(self.headers.get("Content-Length") or 0) + if length: + self.rfile.read(length) + server.queries.append(parse_qs(urlparse(self.path).query)) + server.on_request() + status, headers, body = ( + server.script.pop(0) if len(server.script) > 1 else server.script[0] + ) + if status == 0: + threading.Event().wait(0.5) # not time.sleep: tests patch it + status = 200 + try: + self.send_response(status) + for name, value in headers.items(): + self.send_header(name, value) + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + except OSError: + pass # the client gave up first - self.assertEqual(mock_mount.call_count, 4) + do_GET = do_POST = _answer + def log_message(self, *args: object) -> None: + pass -def _fake_response(status_code: int, body: bytes = b"") -> Response: - response = Response() - response.status_code = status_code - response._content = body - return response + self.httpd = ThreadingHTTPServer(("127.0.0.1", 0), Handler) + self.url = f"http://127.0.0.1:{self.httpd.server_address[1]}/" + threading.Thread( + target=self.httpd.serve_forever, kwargs={"poll_interval": 0.01}, daemon=True + ).start() + @property + def calls(self) -> int: + return len(self.queries) -NO_DATA = b'{"code": "no_data_extracted", "title": "No data extracted"}' -NO_CONTENT = b'{"code": "RESP001", "title": "Could not get content"}' + def close(self) -> None: + self.httpd.shutdown() + self.httpd.server_close() + + +@pytest.fixture +def server() -> Iterator[_Server]: + srv = _Server() + yield srv + srv.close() + + +@pytest.fixture +def sleeps() -> Iterator[list[float]]: + """Every wait between tries, in seconds; none of them actually waits.""" + waited: list[float] = [] + with patch("urllib3.util.retry.time.sleep", side_effect=waited.append): + yield waited + + +def client(server: _Server, **kwargs) -> ZenRowsClient: + return ZenRowsClient(APIKEY, base_url=server.url, **kwargs) + + +# ---- configuration ---- + + +def test_default_is_two_retries() -> None: + zr = ZenRowsClient(APIKEY) + + retry = zr.requests_session.get_adapter("https://api.zenrows.com/v1/").max_retries + assert zr.retries == 2 + assert isinstance(retry, _JitteredRetry) + assert (retry.total, retry.status, retry.connect, retry.other) == (2, 2, 2, 0) + assert retry.read is False + assert retry.allowed_methods is None + assert set(retry.status_forcelist) == {429, 502, 503, 504} + assert (retry.retry_max_wait, retry.retry_budget) == (30.0, 90.0) + + +def test_retries_zero_mounts_no_retry(server: _Server, sleeps: list[float]) -> None: + zr = client(server, retries=0) + server.script = [(503, {}, b"")] + + response = zr.fetch(TARGET) + + assert type(zr.requests_session.get_adapter(server.url)) is HTTPAdapter + assert response.status_code == 503 + assert server.calls == 1 + assert sleeps == [] + + +def test_default_retries_twice_then_returns_the_last_response( + server: _Server, sleeps: list[float] +) -> None: + server.script = [(503, {}, b"")] + + response = client(server).fetch(TARGET) + + assert response.status_code == 503 + assert server.calls == 3 + assert len(sleeps) == 2 + + +# ---- what is retried ---- + + +@pytest.mark.parametrize("status", [429, 502, 503, 504]) +def test_retryable_status_is_retried_until_success( + server: _Server, sleeps: list[float], status: int +) -> None: + server.script = [(status, {}, b""), (200, {}, b"ok")] + + response = client(server).fetch(TARGET) + + assert response.status_code == 200 + assert response.text == "ok" + assert server.calls == 2 + + +def test_post_is_retried(server: _Server, sleeps: list[float]) -> None: + server.script = [(503, {}, b""), (200, {}, b"ok")] + + response = client(server).post(TARGET, data="payload") + + assert response.status_code == 200 + assert server.calls == 2 + + +@pytest.mark.parametrize( + ("status", "body"), + [(400, b""), (401, b""), (404, b""), (422, NO_CONTENT), (422, NO_DATA), (500, b"")], +) +def test_other_statuses_are_not_retried( + server: _Server, sleeps: list[float], status: int, body: bytes +) -> None: + server.script = [(status, {}, body)] + + response = client(server, retries=3).fetch(TARGET) + + assert response.status_code == status + assert server.calls == 1 + assert sleeps == [] + + +def test_413_with_retry_after_is_not_retried(server: _Server, sleeps: list[float]) -> None: + server.script = [(413, {"Retry-After": "1"}, b"")] + + response = client(server).fetch(TARGET) + + assert response.status_code == 413 + assert server.calls == 1 + + +def test_read_timeout_is_not_retried(server: _Server, sleeps: list[float]) -> None: + server.script = [HANG] + + with pytest.raises(requests.exceptions.ReadTimeout): + client(server, retries=3).fetch(TARGET, timeout=0.1) + + assert server.calls == 1 + assert sleeps == [] + + +def test_connection_refused_is_retried(sleeps: list[float]) -> None: + with socket.socket() as sock: + sock.bind(("127.0.0.1", 0)) + port = sock.getsockname()[1] # closed once the block ends: nothing listens + + zr = ZenRowsClient(APIKEY, base_url=f"http://127.0.0.1:{port}/") + with pytest.raises(requests.exceptions.ConnectionError): + zr.fetch(TARGET) + + assert len(sleeps) == 2 + + +# ---- how long it waits ---- + + +def test_jitter_draws_between_quarter_second_and_doubling_ceiling( + server: _Server, sleeps: list[float] +) -> None: + server.script = [(503, {}, b"")] * 4 + [(200, {}, b"ok")] + with patch("zenrows.client.random.uniform", side_effect=lambda lo, hi: hi) as uniform: + response = client(server, retries=4).fetch(TARGET) -@patch("zenrows.client.time.sleep") -class TestUnprocessableRetries(TestCase): - """A 422 is retried in the client, not in urllib3, because only the body - says whether another try can help.""" + assert response.status_code == 200 + assert [c.args for c in uniform.call_args_list] == [ + (0.25, 1.0), + (0.25, 2.0), + (0.25, 4.0), + (0.25, 8.0), + ] + assert sleeps == [1.0, 2.0, 4.0, 8.0] - @patch.object(Session, "request") - def test_no_data_extracted_is_not_retried(self, mock_request, mock_sleep): - mock_request.return_value = _fake_response(422, NO_DATA) - response = ZenRowsClient(apikey, retries=3).extract(url) +def test_jitter_stays_in_bounds(server: _Server, sleeps: list[float]) -> None: + for _ in range(20): + sleeps.clear() + server.script = [(503, {}, b"")] * 3 + [(200, {}, b"ok")] + client(server, retries=3).fetch(TARGET) + assert len(sleeps) == 3 + for attempt, waited in enumerate(sleeps): + assert 0.25 <= waited <= 2.0**attempt - self.assertEqual(response.status_code, 422) - self.assertEqual(mock_request.call_count, 1) - mock_sleep.assert_not_called() - @patch.object(Session, "request") - def test_other_422_is_retried_up_to_retries(self, mock_request, mock_sleep): - mock_request.return_value = _fake_response(422, NO_CONTENT) +def test_retry_max_wait_caps_the_jitter() -> None: + retry = _JitteredRetry(total=10, retry_max_wait=3.0) + for _ in range(6): + retry = retry.new(history=(*retry.history, None)) - response = ZenRowsClient(apikey, retries=3).fetch(url) + with patch("zenrows.client.random.uniform", side_effect=lambda lo, hi: (lo, hi)): + assert retry.get_backoff_time() == (0.25, 3.0) - self.assertEqual(response.status_code, 422) - self.assertEqual(mock_request.call_count, 4) # 1 + 3 retries - self.assertEqual([c.args[0] for c in mock_sleep.call_args_list], [1.0, 2.0]) - @patch.object(Session, "request") - def test_other_422_stops_retrying_on_success(self, mock_request, mock_sleep): - mock_request.side_effect = [_fake_response(422, NO_CONTENT), _fake_response(200, b"ok")] +def test_retry_after_is_honoured(server: _Server, sleeps: list[float]) -> None: + server.script = [(503, {"Retry-After": "7"}, b""), (200, {}, b"ok")] - response = ZenRowsClient(apikey, retries=3).fetch(url) + response = client(server).fetch(TARGET) - self.assertEqual(response.status_code, 200) - self.assertEqual(mock_request.call_count, 2) + assert response.status_code == 200 + assert sleeps == [7.0] - @patch.object(Session, "request") - def test_422_not_retried_without_retries(self, mock_request, mock_sleep): - mock_request.return_value = _fake_response(422, NO_CONTENT) - ZenRowsClient(apikey).fetch(url) +def test_retry_after_is_capped_at_30_seconds(server: _Server, sleeps: list[float]) -> None: + server.script = [(429, {"Retry-After": "120"}, b""), (200, {}, b"ok")] - self.assertEqual(mock_request.call_count, 1) + client(server).fetch(TARGET) - @patch.object(Session, "request") - def test_no_data_extracted_is_not_retried_async(self, mock_request, mock_sleep): - mock_request.return_value = _fake_response(422, NO_DATA) + assert sleeps == [30.0] - response = asyncio.run(ZenRowsClient(apikey, retries=3).extract_async(url)) - self.assertEqual(response.status_code, 422) - self.assertEqual(mock_request.call_count, 1) +def test_retry_after_is_capped_at_retry_max_wait(server: _Server, sleeps: list[float]) -> None: + server.script = [(503, {"Retry-After": "20"}, b""), (200, {}, b"ok")] + + client(server, retry_max_wait=5.0).fetch(TARGET) + + assert sleeps == [5.0] + + +def test_unparseable_retry_after_falls_back_to_jitter(server: _Server, sleeps: list[float]) -> None: + server.script = [(503, {"Retry-After": "soon"}, b""), (200, {}, b"ok")] + + response = client(server).fetch(TARGET) + + assert response.status_code == 200 + assert len(sleeps) == 1 + assert 0.25 <= sleeps[0] <= 1.0 + + +# ---- the retry budget ---- + + +class _Clock: + def __init__(self) -> None: + self.now = 0.0 + + def __call__(self) -> float: + return self.now + + +@pytest.fixture +def clock() -> Iterator[tuple[_Clock, list[float]]]: + """A fake clock for the retry budget, moved on by every wait, and the + list of those waits.""" + fake = _Clock() + waited: list[float] = [] + + def wait(seconds: float) -> None: + waited.append(seconds) + fake.now += seconds + + with ( + patch("zenrows.client._monotonic", fake), + patch("urllib3.util.retry.time.sleep", side_effect=wait), + ): + yield fake, waited + + +def test_no_retry_starts_past_the_budget( + server: _Server, clock: tuple[_Clock, list[float]] +) -> None: + _, sleeps = clock + server.script = [(503, {"Retry-After": "20"}, b"")] + + response = client(server, retries=10).fetch(TARGET) + + # Waits end at 20, 40, 60, 80 s; a fifth would end at 100 s > 90 s. + assert response.status_code == 503 + assert sleeps == [20.0, 20.0, 20.0, 20.0] + assert server.calls == 5 + + +def test_budget_counts_the_first_try(server: _Server, clock: tuple[_Clock, list[float]]) -> None: + fake, sleeps = clock + server.script = [(503, {"Retry-After": "10"}, b"")] + + def slow_first_try() -> None: + if server.calls == 1: + fake.now += 85.0 + + server.on_request = slow_first_try + + response = client(server).fetch(TARGET) + + # 85 s spent + a 10 s wait would end past 90 s. + assert response.status_code == 503 + assert server.calls == 1 + assert sleeps == [] + + +def test_budget_is_per_call(server: _Server, clock: tuple[_Clock, list[float]]) -> None: + _, sleeps = clock + zr = client(server, retries=1) + for _ in range(4): + server.script = [(503, {"Retry-After": "30"}, b""), (200, {}, b"ok")] + assert zr.fetch(TARGET).status_code == 200 + + assert sleeps == [30.0] * 4 + + +def test_custom_budget(server: _Server, clock: tuple[_Clock, list[float]]) -> None: + _, sleeps = clock + server.script = [(503, {"Retry-After": "10"}, b"")] + + response = client(server, retry_budget=5.0).fetch(TARGET) + + assert response.status_code == 503 + assert server.calls == 1 + assert sleeps == [] + + +# ---- Extract ---- + + +def test_no_data_extracted_is_not_retried(server: _Server, sleeps: list[float]) -> None: + server.script = [(422, {}, NO_DATA)] + + response = client(server, retries=3).extract(TARGET) + + assert is_no_data_extracted(response) + assert server.calls == 1 + + +def test_autoparse_fallback_is_unchanged(server: _Server, sleeps: list[float]) -> None: + server.script = [(403, {}, DOMAIN_NOT_PREPARED), (200, {}, b"{}")] + + response = client(server).extract(TARGET) + + assert response.status_code == 200 + assert server.calls == 2 + assert server.queries[0]["extract"] == ["auto"] + assert server.queries[1]["autoparse"] == ["True"] + assert "extract" not in server.queries[1] + assert sleeps == [] + + +def test_autoparse_fallback_is_unchanged_async(server: _Server, sleeps: list[float]) -> None: + server.script = [(403, {}, DOMAIN_NOT_PREPARED), (200, {}, b"{}")] + + response = asyncio.run(client(server).extract_async(TARGET)) + + assert response.status_code == 200 + assert server.calls == 2 + assert server.queries[1]["autoparse"] == ["True"] + + +def test_extract_retries_a_503_before_the_fallback(server: _Server, sleeps: list[float]) -> None: + server.script = [(503, {}, b""), (403, {}, DOMAIN_NOT_PREPARED), (200, {}, b"{}")] + + response = client(server).extract(TARGET) + + assert response.status_code == 200 + assert server.calls == 3 + assert len(sleeps) == 1 + assert server.queries[2]["autoparse"] == ["True"] + + +# ---- is_no_data_extracted ---- + + +def _fake_response(status_code: int, body: bytes = b"") -> requests.Response: + response = requests.Response() + response.status_code = status_code + response._content = body + return response + +def test_is_no_data_extracted_matches_any_case() -> None: + assert is_no_data_extracted(_fake_response(422, NO_DATA)) + assert is_no_data_extracted(_fake_response(422, b'{"code": "NO_DATA_EXTRACTED"}')) -class TestIsNoDataExtracted(TestCase): - def test_matches_422_no_data_extracted_any_case(self): - self.assertTrue(is_no_data_extracted(_fake_response(422, NO_DATA))) - self.assertTrue(is_no_data_extracted(_fake_response(422, b'{"code": "NO_DATA_EXTRACTED"}'))) - def test_rejects_other_codes_statuses_and_bodies(self): - self.assertFalse(is_no_data_extracted(_fake_response(422, NO_CONTENT))) - self.assertFalse(is_no_data_extracted(_fake_response(400, NO_DATA))) - self.assertFalse(is_no_data_extracted(_fake_response(422, b"not json"))) - self.assertFalse(is_no_data_extracted(_fake_response(422, b"[]"))) +def test_is_no_data_extracted_rejects_other_codes_statuses_and_bodies() -> None: + assert not is_no_data_extracted(_fake_response(422, NO_CONTENT)) + assert not is_no_data_extracted(_fake_response(400, NO_DATA)) + assert not is_no_data_extracted(_fake_response(422, b"not json")) + assert not is_no_data_extracted(_fake_response(422, b"[]")) diff --git a/uv.lock b/uv.lock index ecde950..8a93685 100644 --- a/uv.lock +++ b/uv.lock @@ -978,6 +978,7 @@ dependencies = [ { name = "requests" }, { name = "tqdm" }, { name = "typing-extensions", marker = "python_full_version < '3.11'" }, + { name = "urllib3" }, ] [package.dev-dependencies] @@ -999,6 +1000,7 @@ requires-dist = [ { name = "requests", specifier = ">=2.31" }, { name = "tqdm", specifier = ">=4.66" }, { name = "typing-extensions", marker = "python_full_version < '3.11'", specifier = ">=4.0" }, + { name = "urllib3", specifier = ">=1.26" }, ] [package.metadata.requires-dev] From ed9314514f27bf0672cf3b40710c7372e6c29b59 Mon Sep 17 00:00:00 2001 From: gonzalo-zenrows Date: Mon, 19 Oct 2026 22:55:04 +0200 Subject: [PATCH 2/6] fix(client): retry 422 and 500 again, and never replay a one-shot body The first retry change retried every POST and PUT. A POST whose body was a generator then went out a second time with an empty body, and a POST that the target had already processed could run twice. The same change also stopped retrying 422 and 500, which callers relied on. 1. A GET is always eligible. A POST or PUT is retried only with an Idempotency-Key header and a body the client can send again. A generator, an iterator or a stream that cannot seek is never retried. 2. A 500 is retried again. A 422 other than no_data_extracted is retried again in _worker, because urllib3 cannot read the body. It uses the same wait, the same retry count and the same budget clock as the urllib3 retries, so a 422 and a 5xx in one call share both. 3. Retry-After counts only as whole seconds, clamped to 0.25 s and 30 s. The 30 s cap no longer follows retry_max_wait. An HTTP date or a fraction falls back to the jittered wait, which never drops below 0.25 s. 4. The Autoparse fallback of extract(mode="auto") starts no retry after the budget of the first request, sync and async. 5. retries, retry_max_wait and retry_budget are checked when the client is created. An invalid value raises ValueError instead of failing at the first retry. 6. The README documents what is retried, the waits, the budget and the changed default, and its examples use the default retries. --- README.md | 59 +++++- src/zenrows/client.py | 317 ++++++++++++++++++++-------- tests/test_retries.py | 468 ++++++++++++++++++++++++++++++++++++------ 3 files changed, 689 insertions(+), 155 deletions(-) diff --git a/README.md b/README.md index edc99a7..44e7dca 100644 --- a/README.md +++ b/README.md @@ -30,6 +30,7 @@ This package ships three clients: - [PUT Requests](#put-requests) - [Concurrency](#concurrency) - [Extract](#extract) + - [Retries](#retries) - [Quickstart — Batch API (`ZenRowsBatchClient`)](#quickstart--batch-api-zenrowsbatchclient) - [Extract in a batch](#extract-in-a-batch) - [Upload URLs from a CSV](#upload-urls-from-a-csv) @@ -57,14 +58,14 @@ Start using the API by [creating your API Key](https://www.zenrows.com/register? The SDK uses [requests](https://docs.python-requests.org/) for HTTP requests. The client's response will be a requests `Response`. -The client retries a request up to 2 times by default (`retries=2`), and only when another try can help: a `429`, `502`, `503` or `504` response, none of which is charged, or a failure to connect. It never retries any other status (including any `422`) or a read timeout, because the server may still be working on that request. Between tries it waits the server's `Retry-After` header, or else a random time between 0.25 seconds and `2 ** attempt` seconds (full jitter); either wait is capped at `retry_max_wait` (default 30 seconds). It starts no retry whose wait would end more than `retry_budget` seconds (default 90) after the first try. Pass `retries=0` to turn retries off. +The client retries a failed request up to 2 times by default. Pass `retries=0` to send each request once. See [Retries](#retries) for what is retried and how long the client waits. `client.fetch()` is the primary method for the main page-scraping product; `client.get()` still works and is kept as a deprecated alias. ```python from zenrows import ZenRowsClient -client = ZenRowsClient("YOUR-API-KEY", retries=1) +client = ZenRowsClient("YOUR-API-KEY") url = "https://www.zenrows.com/" response = client.fetch(url, params={ @@ -130,7 +131,7 @@ The SDK also offers POST requests by calling the `client.post` method. It can re ```python from zenrows import ZenRowsClient -client = ZenRowsClient("YOUR-API-KEY", retries=1) +client = ZenRowsClient("YOUR-API-KEY") url = "https://httpbin.org/anything" response = client.post(url, data={ @@ -141,6 +142,8 @@ response = client.post(url, data={ print(response.text) ``` +A POST is sent once, even when it fails with a status the client would retry for a GET. To let the client retry it, pass an `Idempotency-Key` header and a body it can send again; see [Retries](#retries). + ### PUT Requests The SDK also offers PUT requests by calling the `client.put` method. It can receive a new parameter `data` that represents the data sent in, for example, a form. @@ -148,7 +151,7 @@ The SDK also offers PUT requests by calling the `client.put` method. It can rece ```python from zenrows import ZenRowsClient -client = ZenRowsClient("YOUR-API-KEY", retries=1) +client = ZenRowsClient("YOUR-API-KEY") url = "https://httpbin.org/anything" response = client.put(url, data={ @@ -159,6 +162,8 @@ response = client.put(url, data={ print(response.text) ``` +A PUT follows the same rule as a POST: it is retried only with an `Idempotency-Key` header and a body the client can send again. + ### Concurrency To limit the concurrency, it uses [asyncio](https://docs.python.org/3/library/asyncio.html), which will simultaneously send a maximum of requests. The concurrency is determined by the plan you are in, so take a look at the [pricing](https://www.zenrows.com/pricing) and set it accordingly. Take into account that each client instance will have its own limit, meaning that two different scripts will not share it, and 429 (Too Many Requests) errors might arise. @@ -171,7 +176,7 @@ We use `asyncio.gather` in the example below. It will wait for all the calls to from zenrows import ZenRowsClient import asyncio -client = ZenRowsClient("YOUR-API-KEY", concurrency=5, retries=1) +client = ZenRowsClient("YOUR-API-KEY", concurrency=5) async def main(): urls = [ @@ -206,6 +211,50 @@ print(response.json()) A `422` with `code: "no_data_extracted"` means Extract knows this kind of page but found no data on it. It isn't charged, and the client never retries it, even with `retries` set. `is_no_data_extracted(response)` tells you when you got one. See the [Extract docs](https://docs.zenrows.com/extract/setup). +### Retries + +> **Changed default.** Earlier versions did not retry unless you passed `retries`. The client now retries up to 2 times by default (`retries=2`). Pass `retries=0` to keep the old behaviour and send each request once. + +The client retries a request when another try can help: + +1. A `429`, `500`, `502`, `503` or `504` response. +2. A `422` response, except Extract's `422` with `code: "no_data_extracted"`, which is a final answer. +3. A failure to connect: connection refused, a name that does not resolve, or a connect timeout. The request never reached the server. + +It never retries these: + +1. Any other status, such as `400`, `401`, `402`, `403` or `404`. +2. A read timeout, or a connection that closes after the request was sent. The server may still be working on that request, and may charge it. + +Which requests it retries: + +1. A GET is always eligible. +2. A POST or PUT is eligible only when it carries an `Idempotency-Key` header and a body the client can send again: a string, bytes, a dict, or a file object that can seek. A generator, an iterator or another one-shot stream is never retried, because a second try would send it empty. Note that `headers` are forwarded to the target site, so the key reaches the target too. +3. Any other request is sent once. + +How long it waits between tries: + +1. When the response has a `Retry-After` header in whole seconds, the client waits that long, at least 0.25 seconds and at most 30 seconds. The 30-second cap is fixed; `retry_max_wait` does not change it. +2. Otherwise, or when `Retry-After` is an HTTP date, a fraction or any other value, it waits a random time between 0.25 seconds and `min(retry_max_wait, 2 ** attempt)` seconds, where `attempt` counts retries from 0. `retry_max_wait` defaults to 30 seconds. The wait is never shorter than 0.25 seconds. + +No retry starts after 90 s, counted from the call's first try (`retry_budget`, default 90). The budget does not cut a running try short, so a call can take longer than the budget by one try. `retries`, a `422` retry and a `5xx` retry share one count and one budget in the same call. In `extract(mode="auto")`, the Autoparse fallback gets its own retry count, but no retry of it starts after 90 s from the first request's first try. + +The client checks its options when you create it: `retries` must be an integer of 0 or more, and `retry_max_wait` and `retry_budget` must be finite numbers above 0. Anything else raises `ValueError`. + +A call cannot be cancelled. A wait between tries runs to its end, also for `fetch_async` and `extract_async`, which run the call on the client's thread pool. + +```python +from zenrows import ZenRowsClient + +client = ZenRowsClient("YOUR-API-KEY", retries=3, retry_max_wait=10, retry_budget=60) + +response = client.post( + "https://httpbin.org/anything", + headers={"Idempotency-Key": "a-unique-key-per-order"}, + data={"key1": "value1"}, +) +``` + ## Quickstart — Batch API (`ZenRowsBatchClient`) > Batch is open to every account and spends the same credits as a regular diff --git a/src/zenrows/client.py b/src/zenrows/client.py index fc74b31..2dca00c 100644 --- a/src/zenrows/client.py +++ b/src/zenrows/client.py @@ -13,17 +13,22 @@ import asyncio import contextvars +import math import os import random +import re import time +from collections.abc import Iterator from concurrent.futures import ThreadPoolExecutor +from contextlib import contextmanager +from dataclasses import dataclass from functools import partial from typing import Any import requests import urllib3 from requests.adapters import HTTPAdapter -from urllib3.exceptions import InvalidHeader, MaxRetryError, ResponseError +from urllib3.exceptions import MaxRetryError, ResponseError from urllib3.util.retry import Retry from zenrows.__version__ import __version__ @@ -31,39 +36,157 @@ DEFAULT_SCRAPER_URL = "https://api.zenrows.com/v1/" DEFAULT_USER_AGENT = f"zenrows/{__version__} python" -# Status codes worth another try. None of them is charged: 429 is a rate -# limit, 503 is the gateway asking to retry later, 502 and 504 never reached -# the target. Every other 4xx/5xx is the final answer. -_RETRY_STATUSES = (429, 502, 503, 504) +# Status codes worth another try, whatever the body says. A 422 is retried +# too, but in `_worker`, because whether it is worth retrying depends on the +# body (see `is_no_data_extracted`), which urllib3 cannot read. +_RETRY_STATUSES = (429, 500, 502, 503, 504) DEFAULT_RETRIES = 2 -# Longest wait between two tries, in seconds: the cap on `Retry-After` and on -# the jittered backoff. Same cap as the Crawl client's `stop()` retry. +# Longest jittered wait between two tries, in seconds. It does not cap +# `Retry-After`, which has its own fixed cap below. DEFAULT_RETRY_MAX_WAIT = 30.0 -# Longest a call may keep retrying, in seconds, counted from its first try: -# no retry starts if the wait before it would end past this. +# No retry starts after this many seconds, counted from the call's first try. +# It does not cut a running try short. DEFAULT_RETRY_BUDGET = 90.0 -# Shortest jittered wait, so a retry never fires back to back. +# Shortest wait between two tries, so a retry never fires back to back. _RETRY_MIN_WAIT = 0.25 +# Longest wait a `Retry-After` header can ask for. +_RETRY_AFTER_CAP = 30.0 +# `Retry-After` is honoured only as whole seconds. An HTTP date, a fraction, +# a sign or anything else falls back to the jittered wait. +_RETRY_AFTER_SECONDS = re.compile(r"[0-9]+") + +# Methods that are safe to send again. Any other method is retried only +# with an `Idempotency-Key` header. +_SAFE_METHODS = frozenset({"GET", "HEAD", "OPTIONS"}) # Clock for the retry budget; tests replace it. _monotonic = time.monotonic -# When the current call's first try started. Set per call by -# `_RetryBudgetAdapter.send`; a context variable, so concurrent calls on + +def _retry_after_seconds(value: str | None) -> float | None: + """The wait a `Retry-After` header asks for, or None when it is not whole + seconds. Clamped to [0.25, 30] seconds.""" + if value is None: + return None + value = value.strip() + if not _RETRY_AFTER_SECONDS.fullmatch(value): + return None + return min(max(float(int(value)), _RETRY_MIN_WAIT), _RETRY_AFTER_CAP) + + +def _jitter(attempt: int, retry_max_wait: float) -> float: + """A random wait in [0.25, max(0.25, min(retry_max_wait, 2 ** attempt))] + seconds (full jitter); `attempt` counts retries from 0.""" + ceiling = max(_RETRY_MIN_WAIT, min(retry_max_wait, 2.0 ** min(attempt, 30))) + return _RETRY_MIN_WAIT + (ceiling - _RETRY_MIN_WAIT) * random.random() + + +def _retry_delay(retry_after: str | None, attempt: int, retry_max_wait: float) -> float: + """The wait before retry number `attempt` (from 0): the `Retry-After` + header when it is whole seconds, or else the jittered wait.""" + seconds = _retry_after_seconds(retry_after) + if seconds is not None: + return seconds + return _jitter(attempt, retry_max_wait) + + +def _check_retry_options(retries: Any, retry_max_wait: Any, retry_budget: Any) -> None: + """Raise ValueError for a retry option the policy cannot use.""" + if isinstance(retries, bool) or not isinstance(retries, int) or retries < 0: + raise ValueError( + f"ZenRowsClient: retries must be an integer of 0 or more, got {retries!r}." + ) + for name, value in (("retry_max_wait", retry_max_wait), ("retry_budget", retry_budget)): + if ( + isinstance(value, bool) + or not isinstance(value, (int, float)) + or not math.isfinite(value) + or value <= 0 + ): + raise ValueError( + f"ZenRowsClient: {name} must be a finite number of seconds above 0, got {value!r}." + ) + + +def _can_send_again(request: requests.PreparedRequest) -> bool: + """True when sending `request` again is safe: a GET, HEAD or OPTIONS, or + any other method with an `Idempotency-Key` header; and in both cases a + body that can be sent again. A generator, an iterator or a stream that + cannot seek is read once, so a second try would send it empty.""" + method = (request.method or "").upper() + if method not in _SAFE_METHODS and "Idempotency-Key" not in request.headers: + return False + body = request.body + if body is None or isinstance(body, (bytes, str)): + return True + seekable = getattr(body, "seekable", None) + return callable(seekable) and bool(seekable()) and hasattr(body, "tell") + + +@dataclass +class _CallState: + """The retry state of one call. urllib3's retries and the 422 retry in + `_worker` share it, so they share one retry count and one budget.""" + + started: float + retries_used: int = 0 + # Whether the request being sent may be sent again; set per request. + can_send_again: bool = True + + +# The current call's retry state. A context variable, so concurrent calls on # the client's thread pool each see their own. -_call_started: contextvars.ContextVar[float | None] = contextvars.ContextVar( - "zenrows_call_started", default=None +_call_state: contextvars.ContextVar[_CallState | None] = contextvars.ContextVar( + "zenrows_call_state", default=None ) +@contextmanager +def _call_scope(started: float | None = None) -> Iterator[_CallState]: + """The current call's retry state, or a new one when there is none yet. + + `started` opens a new state whose budget counts from that time instead + of now: Extract's Autoparse fallback gets its own retry count but no + fresh budget.""" + state = _call_state.get() + if state is not None and started is None: + yield state + return + state = _CallState(started=_monotonic() if started is None else started) + token = _call_state.set(state) + try: + yield state + finally: + _call_state.reset(token) + + +@dataclass(frozen=True) +class _RetryPolicy: + """How often and how long the client retries. One instance per client.""" + + retries: int + retry_max_wait: float + retry_budget: float + + def next_delay(self, state: _CallState, retry_after: str | None) -> float | None: + """The wait before the call's next retry, or None when no retry may + start: the request cannot be sent again, the retries are used up, or + the wait would end past the budget. Counts the retry when it allows + one.""" + if not state.can_send_again or state.retries_used >= self.retries: + return None + delay = _retry_delay(retry_after, state.retries_used, self.retry_max_wait) + if _monotonic() - state.started + delay > self.retry_budget: + return None + state.retries_used += 1 + return delay + + class _JitteredRetry(Retry): - """urllib3 `Retry` with the Zenrows client retry policy. + """urllib3 `Retry` that follows the client's `_RetryPolicy`. - Between tries it waits the server's `Retry-After` (delta-seconds or - HTTP-date), capped at `retry_max_wait`; without one it waits a random time - in `[0.25, min(retry_max_wait, 2 ** attempt)]` seconds (full jitter, - `attempt` counting from 0). It never starts a retry whose wait would end - more than `retry_budget` seconds after the call's first try. + The policy decides whether a retry starts and how long to wait before + it. urllib3 still decides which statuses and errors are retryable. Only the hooks urllib3 1.26 and 2.x share are overridden, so it works on both (no `backoff_jitter` / `retry_after_max`). @@ -72,33 +195,30 @@ class _JitteredRetry(Retry): # urllib3 also retries 413 when it carries `Retry-After`; we don't. RETRY_AFTER_STATUS_CODES = frozenset({429, 503}) - def __init__( - self, - *args: Any, - retry_max_wait: float = DEFAULT_RETRY_MAX_WAIT, - retry_budget: float = DEFAULT_RETRY_BUDGET, - **kwargs: Any, - ): + def __init__(self, *args: Any, policy: _RetryPolicy | None = None, **kwargs: Any): super().__init__(*args, **kwargs) - self.retry_max_wait = retry_max_wait - self.retry_budget = retry_budget - self._started: float | None = None + self.policy = policy or _RetryPolicy( + retries=self.total if isinstance(self.total, int) else 0, + retry_max_wait=DEFAULT_RETRY_MAX_WAIT, + retry_budget=DEFAULT_RETRY_BUDGET, + ) + # The call this retry belongs to, when used outside the client's + # adapter (which sets the context variable). + self._state: _CallState | None = None # The wait before the next try, drawn once in `increment` so the # budget check and the sleep agree on it. self._delay: float | None = None def new(self, **kw: Any) -> "_JitteredRetry": retry = super().new(**kw) - retry.retry_max_wait = self.retry_max_wait - retry.retry_budget = self.retry_budget - started = self._started - if started is None: - started = _call_started.get() - if started is None: - started = _monotonic() - retry._started = started + retry.policy = self.policy + retry._state = self._current_state() return retry + def _current_state(self) -> _CallState: + state = self._state or _call_state.get() + return state if state is not None else _CallState(started=_monotonic()) + def increment( self, method: str | None = None, @@ -109,50 +229,34 @@ def increment( _stacktrace: Any = None, ) -> "_JitteredRetry": retry = super().increment(method, url, response, error, _pool, _stacktrace) - retry._delay = retry._next_delay(response) - started = retry._started if retry._started is not None else _monotonic() - if _monotonic() - started + retry._delay > retry.retry_budget: - reason = error or ResponseError(f"retry budget of {retry.retry_budget}s spent") + retry_after = response.headers.get("Retry-After") if response is not None else None + delay = retry.policy.next_delay(retry._current_state(), retry_after) + if delay is None: + reason = error or ResponseError("no retry left for this call") raise MaxRetryError(_pool, url, reason) from reason + retry._delay = delay return retry + def get_retry_after(self, response: Any) -> float | None: + return _retry_after_seconds(response.headers.get("Retry-After")) + def get_backoff_time(self) -> float: if self._delay is not None: return self._delay - return self._jitter() - - def get_retry_after(self, response: Any) -> float | None: - try: - retry_after = super().get_retry_after(response) - except InvalidHeader: - return None # unparseable: fall back to the jittered backoff - if retry_after is None: - return None - return min(retry_after, self.retry_max_wait) - - def _next_delay(self, response: Any) -> float: - # Mirrors `Retry.sleep`: a positive `Retry-After` wins over the backoff. - if response is not None and self.respect_retry_after_header: - retry_after = self.get_retry_after(response) - if retry_after: - return retry_after - return self._jitter() + return _jitter(len(self.history), self.policy.retry_max_wait) - def _jitter(self) -> float: - attempt = min(max(len(self.history) - 1, 0), 30) - ceiling = min(self.retry_max_wait, 2.0**attempt) - return random.uniform(min(_RETRY_MIN_WAIT, ceiling), ceiling) + def sleep(self, response: Any = None) -> None: + time.sleep(self.get_backoff_time()) class _RetryBudgetAdapter(HTTPAdapter): - """Marks when each call's first try starts, for the retry budget.""" + """Gives each request the retry state of the call it belongs to, and + says whether that request may be sent again.""" def send(self, request: Any, *args: Any, **kwargs: Any) -> requests.Response: - token = _call_started.set(_monotonic()) - try: + with _call_scope() as state: + state.can_send_again = _can_send_again(request) return super().send(request, *args, **kwargs) - finally: - _call_started.reset(token) # Extract's answer when it knows the kind of page but none of the fields came @@ -227,12 +331,18 @@ class ZenRowsClient: `base_url` defaults to the public production endpoint; override via constructor arg or `ZENROWS_SCRAPER_BASE_URL` env var. - `retries` (default 2) retries a 429, 502, 503 or 504, and a failure to - connect, never any other status or a read timeout. Between tries it waits - the server's `Retry-After`, or else a random 0.25s to `2 ** attempt` - seconds; either is capped at `retry_max_wait` (default 30). No retry starts - if its wait would end more than `retry_budget` seconds (default 90) after - the first try. Pass `retries=0` to turn retries off. + `retries` (default 2) retries a 429, 500, 502, 503 or 504, a 422 other + than Extract's `no_data_extracted`, and a failure to connect. It never + retries any other status, a read timeout, or a connection that fails + after the request was sent. A GET is always eligible; a POST or PUT only + with an `Idempotency-Key` header and a body that can be sent again. + + Between tries it waits the server's `Retry-After` when that is whole + seconds (at least 0.25, at most 30), or else a random 0.25 to + `min(retry_max_wait, 2 ** attempt)` seconds (default `retry_max_wait` + 30). No retry starts after `retry_budget` seconds (default 90) from the + call's first try. Pass `retries=0` to turn retries off. A call cannot be + cancelled, so a wait between tries runs to its end. """ # Kept as a class attribute for compatibility — pre-1.4 callers @@ -251,6 +361,7 @@ def __init__( ): if not apikey: raise ValueError("ZenRowsClient: apikey is required.") + _check_retry_options(retries, retry_max_wait, retry_budget) self.apikey = apikey self.api_url = base_url or os.environ.get("ZENROWS_SCRAPER_BASE_URL") or DEFAULT_SCRAPER_URL @@ -259,21 +370,24 @@ def __init__( self.requests_session = requests.Session() self.retry_max_wait = retry_max_wait self.retry_budget = retry_budget + # None when retries are off. `_worker` uses it for the 422 retry. + self._retry_policy: _RetryPolicy | None = None if retries > 0: - # Retries a 429/502/503/504 and a failure to connect, nothing - # else: a read error or timeout may come after the server - # accepted the request, which keeps running and may be charged. + self._retry_policy = _RetryPolicy(retries, retry_max_wait, retry_budget) + # Retries the statuses above and a failure to connect, nothing + # else: a read error, a read timeout or a connection that drops + # after the request was sent may come after the server accepted + # the request, which keeps running and may be charged. max_retries = _JitteredRetry( total=retries, connect=retries, read=False, # re-raise as is, so it stays a ReadTimeout status=retries, other=0, - allowed_methods=None, + allowed_methods=None, # `_can_send_again` decides instead status_forcelist=_RETRY_STATUSES, raise_on_status=False, - retry_max_wait=retry_max_wait, - retry_budget=retry_budget, + policy=self._retry_policy, ) adapter = _RetryBudgetAdapter(max_retries=max_retries) self.requests_session.mount("https://", adapter) @@ -342,13 +456,17 @@ def extract( final_params = dict(params) if params else {} final_params["extract"] = mode _with_adaptive_stealth(final_params, adaptive_stealth) - response = self.fetch(url, final_params, headers, **kwargs) + # One call: the Autoparse fallback gets its own retry count, but its + # budget counts from the first request's first try. + with _call_scope() as state: + response = self.fetch(url, final_params, headers, **kwargs) if mode == "auto" and fallback_to_autoparse and _is_extract_domain_unavailable(response): autoparse_params = dict(params) if params else {} autoparse_params["autoparse"] = True _with_adaptive_stealth(autoparse_params, adaptive_stealth) - return self.fetch(url, autoparse_params, headers, **kwargs) + with _call_scope(started=state.started): + return self.fetch(url, autoparse_params, headers, **kwargs) return response @@ -409,13 +527,17 @@ async def extract_async( final_params = dict(params) if params else {} final_params["extract"] = mode _with_adaptive_stealth(final_params, adaptive_stealth) - response = await self.fetch_async(url, final_params, headers, **kwargs) + # One call, as in `extract`. `_offload` carries the scope to the + # worker thread. + with _call_scope() as state: + response = await self.fetch_async(url, final_params, headers, **kwargs) if mode == "auto" and fallback_to_autoparse and _is_extract_domain_unavailable(response): autoparse_params = dict(params) if params else {} autoparse_params["autoparse"] = True _with_adaptive_stealth(autoparse_params, adaptive_stealth) - return await self.fetch_async(url, autoparse_params, headers, **kwargs) + with _call_scope(started=state.started): + return await self.fetch_async(url, autoparse_params, headers, **kwargs) return response @@ -463,9 +585,12 @@ async def _offload( **kwargs: Any, ) -> requests.Response: loop = asyncio.get_event_loop() + # Run in a copy of this context, so the worker thread sees the + # caller's retry state (see `extract_async`). + context = contextvars.copy_context() return await loop.run_in_executor( self.executor, - partial(self._worker, method, url, params, headers, **kwargs), + partial(context.run, self._worker, method, url, params, headers, **kwargs), ) def _worker( @@ -495,7 +620,8 @@ def _worker( final_headers["Connection"] = None final_headers.update(headers) - return self.requests_session.request( + send = partial( + self.requests_session.request, method, self.api_url, params=final_params, @@ -503,3 +629,20 @@ def _worker( data=data, **kwargs, ) + policy = self._retry_policy + if policy is None: + return send() + with _call_scope() as state: + response = send() + # Retry a 422 here rather than in urllib3, which can't read the + # body: Extract's `no_data_extracted` is a final answer, any other + # 422 is a failure to get the target's content and is worth + # another try. Same count, wait and budget as urllib3's retries. + # The adapter already set `state.can_send_again` for this request. + while response.status_code == 422 and not is_no_data_extracted(response): + delay = policy.next_delay(state, response.headers.get("Retry-After")) + if delay is None: + break + time.sleep(delay) + response = send() + return response diff --git a/tests/test_retries.py b/tests/test_retries.py index 7d39ae9..f33ba91 100644 --- a/tests/test_retries.py +++ b/tests/test_retries.py @@ -1,12 +1,17 @@ """The ZenRowsClient retry policy, end to end against a local HTTP server. -Retried: 429, 502, 503, 504 and a failure to connect. Never retried: any -other status (422 included) or a read timeout. Waits honour `Retry-After` -(capped at `retry_max_wait`) or else use full jitter, and no retry starts -past `retry_budget`. Sleeps are patched out; the budget runs on a fake clock. +Retried: 429, 500, 502, 503, 504, a 422 other than `no_data_extracted`, and a +failure to connect. Never retried: any other status, a read timeout, or a +connection that drops after the request was sent. A GET is always eligible; a +POST or PUT only with an `Idempotency-Key` header and a body that can be sent +again. Waits honour a whole-seconds `Retry-After` (clamped to [0.25, 30]) or +else use full jitter, and no retry starts past `retry_budget`. Sleeps are +patched out; the budget runs on a fake clock. """ import asyncio +import io +import math import socket import threading from collections.abc import Iterator @@ -19,40 +24,59 @@ from requests.adapters import HTTPAdapter from zenrows import ZenRowsClient, is_no_data_extracted -from zenrows.client import _JitteredRetry +from zenrows.client import _jitter, _JitteredRetry, _retry_after_seconds APIKEY = "APIKEY" TARGET = "http://example.com" +KEY = {"Idempotency-Key": "a-unique-key"} NO_DATA = b'{"code": "no_data_extracted", "title": "No data extracted"}' NO_CONTENT = b'{"code": "RESP001", "title": "Could not get content"}' DOMAIN_NOT_PREPARED = b'{"code": "REQS007", "title": "Domain not prepared"}' # A scripted response: (status, headers, body). HANG answers only after the -# client's read timeout has passed. +# client's read timeout has passed. DROP reads the request and closes the +# connection without an answer. HANG = (0, {}, b"") +DROP = (-1, {}, b"") + + +def _read_body(handler: BaseHTTPRequestHandler) -> bytes: + if handler.headers.get("Transfer-Encoding", "").lower() == "chunked": + body = b"" + while True: + size = int(handler.rfile.readline().strip() or b"0", 16) + if size == 0: + handler.rfile.readline() + return body + body += handler.rfile.read(size) + handler.rfile.readline() + length = int(handler.headers.get("Content-Length") or 0) + return handler.rfile.read(length) if length else b"" class _Server: """Answers each request with the next scripted response (the last one - repeats) and records the query of every request it got.""" + repeats) and records the query and body of every request it got.""" def __init__(self) -> None: self.script: list[tuple[int, dict[str, str], bytes]] = [(200, {}, b"ok")] self.queries: list[dict[str, list[str]]] = [] + self.bodies: list[bytes] = [] self.on_request = lambda: None server = self class Handler(BaseHTTPRequestHandler): def _answer(self) -> None: - length = int(self.headers.get("Content-Length") or 0) - if length: - self.rfile.read(length) + server.bodies.append(_read_body(self)) server.queries.append(parse_qs(urlparse(self.path).query)) server.on_request() status, headers, body = ( server.script.pop(0) if len(server.script) > 1 else server.script[0] ) + if status == -1: + self.close_connection = True + return # no status line: the client sees the connection close if status == 0: threading.Event().wait(0.5) # not time.sleep: tests patch it status = 200 @@ -66,7 +90,7 @@ def _answer(self) -> None: except OSError: pass # the client gave up first - do_GET = do_POST = _answer + do_GET = do_POST = do_PUT = _answer def log_message(self, *args: object) -> None: pass @@ -97,7 +121,7 @@ def server() -> Iterator[_Server]: def sleeps() -> Iterator[list[float]]: """Every wait between tries, in seconds; none of them actually waits.""" waited: list[float] = [] - with patch("urllib3.util.retry.time.sleep", side_effect=waited.append): + with patch("time.sleep", side_effect=waited.append): yield waited @@ -105,6 +129,12 @@ def client(server: _Server, **kwargs) -> ZenRowsClient: return ZenRowsClient(APIKEY, base_url=server.url, **kwargs) +def _closed_port() -> int: + with socket.socket() as sock: + sock.bind(("127.0.0.1", 0)) + return sock.getsockname()[1] # closed once the block ends: nothing listens + + # ---- configuration ---- @@ -116,9 +146,12 @@ def test_default_is_two_retries() -> None: assert isinstance(retry, _JitteredRetry) assert (retry.total, retry.status, retry.connect, retry.other) == (2, 2, 2, 0) assert retry.read is False - assert retry.allowed_methods is None - assert set(retry.status_forcelist) == {429, 502, 503, 504} - assert (retry.retry_max_wait, retry.retry_budget) == (30.0, 90.0) + assert set(retry.status_forcelist) == {429, 500, 502, 503, 504} + assert (retry.policy.retries, retry.policy.retry_max_wait, retry.policy.retry_budget) == ( + 2, + 30.0, + 90.0, + ) def test_retries_zero_mounts_no_retry(server: _Server, sleeps: list[float]) -> None: @@ -133,22 +166,64 @@ def test_retries_zero_mounts_no_retry(server: _Server, sleeps: list[float]) -> N assert sleeps == [] +def test_retries_zero_does_not_retry_a_422(server: _Server, sleeps: list[float]) -> None: + server.script = [(422, {}, NO_CONTENT)] + + response = client(server, retries=0).fetch(TARGET) + + assert response.status_code == 422 + assert server.calls == 1 + assert sleeps == [] + + def test_default_retries_twice_then_returns_the_last_response( server: _Server, sleeps: list[float] ) -> None: - server.script = [(503, {}, b"")] + server.script = [(503, {}, b"first"), (503, {}, b"second"), (503, {}, b"last")] response = client(server).fetch(TARGET) assert response.status_code == 503 + assert response.content == b"last" assert server.calls == 3 assert len(sleeps) == 2 +@pytest.mark.parametrize( + ("option", "value"), + [ + ("retries", -1), + ("retries", 1.5), + ("retries", True), + ("retries", None), + ("retries", "2"), + ("retry_max_wait", 0), + ("retry_max_wait", -1.0), + ("retry_max_wait", math.inf), + ("retry_max_wait", math.nan), + ("retry_max_wait", "5"), + ("retry_budget", 0), + ("retry_budget", -1.0), + ("retry_budget", math.inf), + ("retry_budget", math.nan), + ("retry_budget", None), + ], +) +def test_invalid_retry_option_is_rejected(option: str, value: object) -> None: + with pytest.raises(ValueError, match=option): + ZenRowsClient(APIKEY, **{option: value}) + + +def test_smallest_valid_retry_options_are_accepted() -> None: + zr = ZenRowsClient(APIKEY, retries=0, retry_max_wait=0.1, retry_budget=0.1) + + assert (zr.retries, zr.retry_max_wait, zr.retry_budget) == (0, 0.1, 0.1) + + # ---- what is retried ---- -@pytest.mark.parametrize("status", [429, 502, 503, 504]) +@pytest.mark.parametrize("status", [429, 500, 502, 503, 504]) def test_retryable_status_is_retried_until_success( server: _Server, sleeps: list[float], status: int ) -> None: @@ -161,18 +236,9 @@ def test_retryable_status_is_retried_until_success( assert server.calls == 2 -def test_post_is_retried(server: _Server, sleeps: list[float]) -> None: - server.script = [(503, {}, b""), (200, {}, b"ok")] - - response = client(server).post(TARGET, data="payload") - - assert response.status_code == 200 - assert server.calls == 2 - - @pytest.mark.parametrize( ("status", "body"), - [(400, b""), (401, b""), (404, b""), (422, NO_CONTENT), (422, NO_DATA), (500, b"")], + [(400, b""), (401, b""), (402, b""), (403, b""), (404, b""), (422, NO_DATA)], ) def test_other_statuses_are_not_retried( server: _Server, sleeps: list[float], status: int, body: bytes @@ -186,6 +252,62 @@ def test_other_statuses_are_not_retried( assert sleeps == [] +@pytest.mark.parametrize("body", [NO_CONTENT, b"", b"not json"]) +def test_422_other_than_no_data_extracted_is_retried( + server: _Server, sleeps: list[float], body: bytes +) -> None: + server.script = [(422, {}, body), (200, {}, b"ok")] + + response = client(server).fetch(TARGET) + + assert response.status_code == 200 + assert server.calls == 2 + assert len(sleeps) == 1 + + +def test_422_is_returned_as_is_after_the_last_try(server: _Server, sleeps: list[float]) -> None: + last = b'{"code": "RESP001", "title": "last"}' + server.script = [(422, {}, NO_CONTENT), (422, {}, NO_CONTENT), (422, {}, last)] + + response = client(server).fetch(TARGET) + + assert response.status_code == 422 + assert response.content == last + assert server.calls == 3 + assert len(sleeps) == 2 + + +@pytest.mark.parametrize( + "script", + [ + [(503, {}, b""), (422, {}, NO_CONTENT), (422, {}, NO_CONTENT), (200, {}, b"ok")], + [(422, {}, NO_CONTENT), (503, {}, b""), (503, {}, b""), (200, {}, b"ok")], + [(422, {}, NO_CONTENT), (503, {}, b""), (422, {}, NO_CONTENT), (200, {}, b"ok")], + ], +) +def test_422_and_5xx_share_one_retry_count( + server: _Server, sleeps: list[float], script: list[tuple[int, dict[str, str], bytes]] +) -> None: + server.script = script + + response = client(server).fetch(TARGET) + + # Two retries in all, whichever status asked for them. + assert response.status_code != 200 + assert server.calls == 3 + assert len(sleeps) == 2 + + +def test_422_and_5xx_share_one_attempt_index(server: _Server, sleeps: list[float]) -> None: + server.script = [(503, {}, b""), (422, {}, NO_CONTENT), (200, {}, b"ok")] + + with patch("zenrows.client.random.random", return_value=1.0): + response = client(server).fetch(TARGET) + + assert response.status_code == 200 + assert sleeps == [1.0, 2.0] + + def test_413_with_retry_after_is_not_retried(server: _Server, sleeps: list[float]) -> None: server.script = [(413, {"Retry-After": "1"}, b"")] @@ -205,36 +327,172 @@ def test_read_timeout_is_not_retried(server: _Server, sleeps: list[float]) -> No assert sleeps == [] -def test_connection_refused_is_retried(sleeps: list[float]) -> None: - with socket.socket() as sock: - sock.bind(("127.0.0.1", 0)) - port = sock.getsockname()[1] # closed once the block ends: nothing listens +def test_connection_dropped_after_send_is_not_retried_for_get( + server: _Server, sleeps: list[float] +) -> None: + server.script = [DROP, (200, {}, b"ok")] + + with pytest.raises(requests.exceptions.ConnectionError): + client(server, retries=3).fetch(TARGET) + + assert server.calls == 1 + assert sleeps == [] + + +def test_connection_dropped_after_send_is_not_retried_for_post( + server: _Server, sleeps: list[float] +) -> None: + server.script = [DROP, (200, {}, b"ok")] + + with pytest.raises(requests.exceptions.ConnectionError): + client(server, retries=3).post(TARGET, headers=KEY, data="payload") + + assert server.calls == 1 + assert sleeps == [] + - zr = ZenRowsClient(APIKEY, base_url=f"http://127.0.0.1:{port}/") +def test_connection_refused_is_retried(sleeps: list[float]) -> None: + zr = ZenRowsClient(APIKEY, base_url=f"http://127.0.0.1:{_closed_port()}/") with pytest.raises(requests.exceptions.ConnectionError): zr.fetch(TARGET) assert len(sleeps) == 2 -# ---- how long it waits ---- +# ---- which requests may be sent again ---- + + +@pytest.mark.parametrize("verb", ["post", "put"]) +def test_post_and_put_without_idempotency_key_are_sent_once( + server: _Server, sleeps: list[float], verb: str +) -> None: + server.script = [(503, {}, b""), (200, {}, b"ok")] + response = getattr(client(server), verb)(TARGET, data="payload") -def test_jitter_draws_between_quarter_second_and_doubling_ceiling( + assert response.status_code == 503 + assert server.calls == 1 + assert sleeps == [] + + +def test_post_without_idempotency_key_does_not_retry_a_422( server: _Server, sleeps: list[float] ) -> None: + server.script = [(422, {}, NO_CONTENT), (200, {}, b"ok")] + + response = client(server).post(TARGET, data="payload") + + assert response.status_code == 422 + assert server.calls == 1 + + +def test_post_without_idempotency_key_does_not_retry_a_failure_to_connect( + sleeps: list[float], +) -> None: + zr = ZenRowsClient(APIKEY, base_url=f"http://127.0.0.1:{_closed_port()}/") + with pytest.raises(requests.exceptions.ConnectionError): + zr.post(TARGET, data="payload") + + assert sleeps == [] + + +@pytest.mark.parametrize( + ("verb", "data", "sent"), + [ + ("post", "payload", b"payload"), + ("post", b"payload", b"payload"), + ("put", {"a": "1"}, b"a=1"), + ("post", io.BytesIO(b"payload"), b"payload"), + ], +) +def test_idempotency_key_and_a_body_that_can_be_sent_again_are_retried( + server: _Server, sleeps: list[float], verb: str, data: object, sent: bytes +) -> None: + server.script = [(503, {}, b""), (200, {}, b"ok")] + + response = getattr(client(server), verb)(TARGET, headers=KEY, data=data) + + assert response.status_code == 200 + assert server.bodies == [sent, sent] + + +def test_idempotency_key_matches_any_case(server: _Server, sleeps: list[float]) -> None: + server.script = [(422, {}, NO_CONTENT), (200, {}, b"ok")] + + response = client(server).post(TARGET, headers={"idempotency-key": "k"}, data="payload") + + assert response.status_code == 200 + assert server.calls == 2 + + +@pytest.mark.parametrize("status", [503, 422]) +@pytest.mark.parametrize("kind", ["generator", "iterator"]) +def test_a_body_that_cannot_be_sent_again_is_never_retried( + server: _Server, sleeps: list[float], status: int, kind: str +) -> None: + server.script = [(status, {}, NO_CONTENT), (200, {}, b"ok")] + chunks = [b"chunk1-", b"chunk2"] + data = (c for c in chunks) if kind == "generator" else iter(chunks) + + response = client(server).post(TARGET, headers=KEY, data=data) + + assert response.status_code == status + assert server.bodies == [b"chunk1-chunk2"] + assert sleeps == [] + + +# ---- how long it waits ---- + + +@pytest.mark.parametrize( + ("value", "millis"), + [ + ("5", 5000), + ("0", 250), + (" 7 ", 7000), + ("99999", 30000), + ("1.5", None), + ("-1", None), + ("1e3", None), + ("0x10", None), + ("", None), + ("abc", None), + ("Wed, 21 Oct 2026 07:28:00 GMT", None), + ], +) +def test_retry_after_conformance(value: str, millis: int | None) -> None: + seconds = _retry_after_seconds(value) + + assert (None if seconds is None else round(seconds * 1000)) == millis + + +@pytest.mark.parametrize( + ("attempt", "retry_max_wait", "low", "high"), + [ + (0, 30.0, 250, 1000), + (1, 30.0, 250, 2000), + (2, 30.0, 250, 4000), + (5, 30.0, 250, 30000), + (9, 30.0, 250, 30000), + (4, 5.0, 250, 5000), + (0, 0.1, 250, 250), + (6, 0.1, 250, 250), + ], +) +def test_jitter_conformance(attempt: int, retry_max_wait: float, low: int, high: int) -> None: + with patch("zenrows.client.random.random", return_value=0.0): + assert round(_jitter(attempt, retry_max_wait) * 1000) == low + with patch("zenrows.client.random.random", return_value=1.0): + assert round(_jitter(attempt, retry_max_wait) * 1000) == high + + +def test_jitter_ceiling_doubles_per_retry(server: _Server, sleeps: list[float]) -> None: server.script = [(503, {}, b"")] * 4 + [(200, {}, b"ok")] - with patch("zenrows.client.random.uniform", side_effect=lambda lo, hi: hi) as uniform: + with patch("zenrows.client.random.random", return_value=1.0): response = client(server, retries=4).fetch(TARGET) assert response.status_code == 200 - assert [c.args for c in uniform.call_args_list] == [ - (0.25, 1.0), - (0.25, 2.0), - (0.25, 4.0), - (0.25, 8.0), - ] assert sleeps == [1.0, 2.0, 4.0, 8.0] @@ -248,48 +506,64 @@ def test_jitter_stays_in_bounds(server: _Server, sleeps: list[float]) -> None: assert 0.25 <= waited <= 2.0**attempt -def test_retry_max_wait_caps_the_jitter() -> None: - retry = _JitteredRetry(total=10, retry_max_wait=3.0) - for _ in range(6): - retry = retry.new(history=(*retry.history, None)) +def test_tiny_retry_max_wait_still_waits_a_quarter_second( + server: _Server, sleeps: list[float] +) -> None: + server.script = [(503, {}, b""), (503, {}, b""), (200, {}, b"ok")] + + client(server, retry_max_wait=0.1).fetch(TARGET) - with patch("zenrows.client.random.uniform", side_effect=lambda lo, hi: (lo, hi)): - assert retry.get_backoff_time() == (0.25, 3.0) + assert sleeps == [0.25, 0.25] -def test_retry_after_is_honoured(server: _Server, sleeps: list[float]) -> None: - server.script = [(503, {"Retry-After": "7"}, b""), (200, {}, b"ok")] +@pytest.mark.parametrize( + ("value", "waited"), [("7", 7.0), ("0", 0.25), ("99999", 30.0), (" 3 ", 3.0)] +) +def test_retry_after_is_honoured( + server: _Server, sleeps: list[float], value: str, waited: float +) -> None: + server.script = [(503, {"Retry-After": value}, b""), (200, {}, b"ok")] response = client(server).fetch(TARGET) assert response.status_code == 200 - assert sleeps == [7.0] + assert sleeps == [waited] -def test_retry_after_is_capped_at_30_seconds(server: _Server, sleeps: list[float]) -> None: - server.script = [(429, {"Retry-After": "120"}, b""), (200, {}, b"ok")] +def test_retry_after_on_a_422_is_honoured(server: _Server, sleeps: list[float]) -> None: + server.script = [(422, {"Retry-After": "4"}, NO_CONTENT), (200, {}, b"ok")] client(server).fetch(TARGET) - assert sleeps == [30.0] + assert sleeps == [4.0] -def test_retry_after_is_capped_at_retry_max_wait(server: _Server, sleeps: list[float]) -> None: - server.script = [(503, {"Retry-After": "20"}, b""), (200, {}, b"ok")] +@pytest.mark.parametrize("retry_max_wait", [5.0, 60.0]) +def test_retry_after_does_not_depend_on_retry_max_wait( + server: _Server, sleeps: list[float], retry_max_wait: float +) -> None: + server.script = [ + (429, {"Retry-After": "20"}, b""), + (429, {"Retry-After": "45"}, b""), + (200, {}, b"ok"), + ] - client(server, retry_max_wait=5.0).fetch(TARGET) + client(server, retry_max_wait=retry_max_wait).fetch(TARGET) - assert sleeps == [5.0] + assert sleeps == [20.0, 30.0] -def test_unparseable_retry_after_falls_back_to_jitter(server: _Server, sleeps: list[float]) -> None: - server.script = [(503, {"Retry-After": "soon"}, b""), (200, {}, b"ok")] +@pytest.mark.parametrize("value", ["soon", "1.5", "-5", "Wed, 21 Oct 2099 07:28:00 GMT"]) +def test_retry_after_other_than_whole_seconds_falls_back_to_jitter( + server: _Server, sleeps: list[float], value: str +) -> None: + server.script = [(503, {"Retry-After": value}, b""), (200, {}, b"ok")] - response = client(server).fetch(TARGET) + with patch("zenrows.client.random.random", return_value=0.5): + response = client(server).fetch(TARGET) assert response.status_code == 200 - assert len(sleeps) == 1 - assert 0.25 <= sleeps[0] <= 1.0 + assert sleeps == [0.625] # ---- the retry budget ---- @@ -316,7 +590,7 @@ def wait(seconds: float) -> None: with ( patch("zenrows.client._monotonic", fake), - patch("urllib3.util.retry.time.sleep", side_effect=wait), + patch("time.sleep", side_effect=wait), ): yield fake, waited @@ -374,6 +648,22 @@ def test_custom_budget(server: _Server, clock: tuple[_Clock, list[float]]) -> No assert sleeps == [] +def test_422_and_5xx_share_one_budget(server: _Server, clock: tuple[_Clock, list[float]]) -> None: + _, sleeps = clock + server.script = [ + (503, {"Retry-After": "30"}, b""), + (422, {"Retry-After": "30"}, NO_CONTENT), + (200, {}, b"ok"), + ] + + response = client(server, retries=5, retry_budget=50.0).fetch(TARGET) + + # The 503's wait ends at 30 s; the 422's would end at 60 s > 50 s. + assert response.status_code == 422 + assert server.calls == 2 + assert sleeps == [30.0] + + # ---- Extract ---- @@ -386,6 +676,15 @@ def test_no_data_extracted_is_not_retried(server: _Server, sleeps: list[float]) assert server.calls == 1 +def test_extract_retries_another_422(server: _Server, sleeps: list[float]) -> None: + server.script = [(422, {}, NO_CONTENT), (200, {}, b"{}")] + + response = client(server).extract(TARGET) + + assert response.status_code == 200 + assert server.calls == 2 + + def test_autoparse_fallback_is_unchanged(server: _Server, sleeps: list[float]) -> None: server.script = [(403, {}, DOMAIN_NOT_PREPARED), (200, {}, b"{}")] @@ -420,6 +719,49 @@ def test_extract_retries_a_503_before_the_fallback(server: _Server, sleeps: list assert server.queries[2]["autoparse"] == ["True"] +@pytest.mark.parametrize("use_async", [False, True]) +def test_autoparse_fallback_shares_the_budget( + server: _Server, clock: tuple[_Clock, list[float]], use_async: bool +) -> None: + _, sleeps = clock + server.script = [ + (503, {"Retry-After": "30"}, b""), + (403, {}, DOMAIN_NOT_PREPARED), + (503, {"Retry-After": "30"}, b""), + (200, {}, b"{}"), + ] + zr = client(server, retry_budget=50.0) + + response = asyncio.run(zr.extract_async(TARGET)) if use_async else zr.extract(TARGET) + + # The first wait ends at 30 s. A fresh budget would let the fallback + # wait until 30 s after its own start; the shared one ends it at 50 s. + assert response.status_code == 503 + assert server.calls == 3 + assert server.queries[2]["autoparse"] == ["True"] + assert sleeps == [30.0] + + +@pytest.mark.parametrize("use_async", [False, True]) +def test_autoparse_fallback_gets_its_own_retry_count( + server: _Server, sleeps: list[float], use_async: bool +) -> None: + server.script = [ + (503, {}, b""), + (503, {}, b""), + (403, {}, DOMAIN_NOT_PREPARED), + (503, {}, b""), + (200, {}, b"{}"), + ] + zr = client(server) + + response = asyncio.run(zr.extract_async(TARGET)) if use_async else zr.extract(TARGET) + + assert response.status_code == 200 + assert server.calls == 5 + assert len(sleeps) == 3 + + # ---- is_no_data_extracted ---- From 13f9940d4f330eaa75708959c2863eeb61b7fe89 Mon Sep 17 00:00:00 2001 From: gonzalo-zenrows Date: Tue, 20 Oct 2026 17:28:43 +0200 Subject: [PATCH 3/6] fix(client): never retry a post or put, and never resend a file body The API does not deduplicate Fetch or Extract requests by Idempotency-Key. It forwards that header to the target site. A retried POST or PUT with a key could therefore run twice, so the key proved nothing. GET, HEAD and OPTIONS are now the only retried methods. POST and PUT go out once, whatever the headers, the status or the failure, including a refused connection. Every method also keeps a body check. A file object, a generator, an iterator or another stream is never sent again, because the first try reads it. The 422 loop checks the data it builds the request from, and the urllib3 retry checks the prepared body. Before this, a POST with a key and an io.BytesIO body sent an empty body on its 422 retry, and the server answered 200 to it. The Idempotency-Key rule, the seek-and-rewind check and the tests for keyed retries are gone. The README now says POST and PUT are never retried, and that a caller who needs it writes that retry. --- README.md | 17 +++--- src/zenrows/client.py | 65 +++++++++++++-------- tests/test_retries.py | 129 ++++++++++++++++++++++++------------------ 3 files changed, 123 insertions(+), 88 deletions(-) diff --git a/README.md b/README.md index 1bd8996..123ff6b 100644 --- a/README.md +++ b/README.md @@ -143,7 +143,7 @@ response = client.post(url, data={ print(response.text) ``` -A POST is sent once, even when it fails with a status the client would retry for a GET. To let the client retry it, pass an `Idempotency-Key` header and a body it can send again; see [Retries](#retries). +The client never retries a POST. It sends it once, even when it fails with a status the client would retry for a GET. If you need to retry a POST, write that retry yourself; see [Retries](#retries). ### PUT Requests @@ -163,7 +163,7 @@ response = client.put(url, data={ print(response.text) ``` -A PUT follows the same rule as a POST: it is retried only with an `Idempotency-Key` header and a body the client can send again. +The client never retries a PUT either. It sends it once. ### Concurrency @@ -286,9 +286,10 @@ It never retries these: Which requests it retries: -1. A GET is always eligible. -2. A POST or PUT is eligible only when it carries an `Idempotency-Key` header and a body the client can send again: a string, bytes, a dict, or a file object that can seek. A generator, an iterator or another one-shot stream is never retried, because a second try would send it empty. Note that `headers` are forwarded to the target site, so the key reaches the target too. -3. Any other request is sent once. +1. A GET is eligible when its body is empty, a string, bytes, or a dict the client encodes. +2. A request with a file object, a generator, an iterator or another stream as its body is never retried. The first try reads it, so a second try would send it empty. +3. A POST or PUT is never retried. The client sends it once, whatever its headers, its body or the failure. If you need to retry one, write that retry yourself. +4. The API does not deduplicate a request by its `Idempotency-Key` header, so the client does not retry on that header. How long it waits between tries: @@ -306,11 +307,7 @@ from zenrows import ZenRowsClient client = ZenRowsClient("YOUR-API-KEY", retries=3, retry_max_wait=10, retry_budget=60) -response = client.post( - "https://httpbin.org/anything", - headers={"Idempotency-Key": "a-unique-key-per-order"}, - data={"key1": "value1"}, -) +response = client.fetch("https://www.zenrows.com/") ``` ## Quickstart — Batch API (`ZenRowsBatchClient`) diff --git a/src/zenrows/client.py b/src/zenrows/client.py index ecb8666..d82d81e 100644 --- a/src/zenrows/client.py +++ b/src/zenrows/client.py @@ -19,7 +19,7 @@ import re import time import warnings -from collections.abc import Iterator +from collections.abc import Iterator, Mapping from concurrent.futures import ThreadPoolExecutor from contextlib import contextmanager from dataclasses import dataclass @@ -57,8 +57,9 @@ # a sign or anything else falls back to the jittered wait. _RETRY_AFTER_SECONDS = re.compile(r"[0-9]+") -# Methods that are safe to send again. Any other method is retried only -# with an `Idempotency-Key` header. +# The only methods the client retries. Any other method is sent once: the API +# does not deduplicate a request by `Idempotency-Key`, so a second POST or PUT +# could run twice. _SAFE_METHODS = frozenset({"GET", "HEAD", "OPTIONS"}) # Clock for the retry budget; tests replace it. @@ -110,19 +111,29 @@ def _check_retry_options(retries: Any, retry_max_wait: Any, retry_budget: Any) - ) +def _method_can_be_sent_again(method: str | None) -> bool: + """True for GET, HEAD and OPTIONS, the only methods the client retries.""" + return (method or "").upper() in _SAFE_METHODS + + +def _body_can_be_sent_again(body: Any) -> bool: + """True when a request body is none, a string or bytes. A file object, a + generator, an iterator or a stream is read by the first try, so it is + never sent again.""" + return body is None or isinstance(body, (bytes, str)) + + +def _data_can_be_sent_again(data: Any) -> bool: + """True when `requests` builds the same body from `data` every time: + none, a string, bytes, or a dict or list it encodes. A file object, a + generator or an iterator is read by the first try.""" + return data is None or isinstance(data, (bytes, str, Mapping, list, tuple)) + + def _can_send_again(request: requests.PreparedRequest) -> bool: - """True when sending `request` again is safe: a GET, HEAD or OPTIONS, or - any other method with an `Idempotency-Key` header; and in both cases a - body that can be sent again. A generator, an iterator or a stream that - cannot seek is read once, so a second try would send it empty.""" - method = (request.method or "").upper() - if method not in _SAFE_METHODS and "Idempotency-Key" not in request.headers: - return False - body = request.body - if body is None or isinstance(body, (bytes, str)): - return True - seekable = getattr(body, "seekable", None) - return callable(seekable) and bool(seekable()) and hasattr(body, "tell") + """True when urllib3 may send `request` again: a GET, HEAD or OPTIONS + whose body is none, a string or bytes.""" + return _method_can_be_sent_again(request.method) and _body_can_be_sent_again(request.body) @dataclass @@ -132,7 +143,7 @@ class _CallState: started: float retries_used: int = 0 - # Whether the request being sent may be sent again; set per request. + # Whether urllib3 may send the current request again; set per request. can_send_again: bool = True @@ -168,10 +179,10 @@ class _RetryPolicy: def next_delay(self, state: _CallState, retry_after: str | None) -> float | None: """The wait before the call's next retry, or None when no retry may - start: the request cannot be sent again, the retries are used up, or - the wait would end past the budget. Counts the retry when it allows - one.""" - if not state.can_send_again or state.retries_used >= self.retries: + start: the retries are used up, or the wait would end past the + budget. Counts the retry when it allows one. The caller first checks + that the request may be sent again.""" + if state.retries_used >= self.retries: return None delay = _retry_delay(retry_after, state.retries_used, self.retry_max_wait) if _monotonic() - state.started + delay > self.retry_budget: @@ -228,7 +239,8 @@ def increment( ) -> "_JitteredRetry": retry = super().increment(method, url, response, error, _pool, _stacktrace) retry_after = response.headers.get("Retry-After") if response is not None else None - delay = retry.policy.next_delay(retry._current_state(), retry_after) + state = retry._current_state() + delay = retry.policy.next_delay(state, retry_after) if state.can_send_again else None if delay is None: reason = error or ResponseError("no retry left for this call") raise MaxRetryError(_pool, url, reason) from reason @@ -406,8 +418,10 @@ class ZenRowsClient: `retries` (default 2) retries a 429, 500, 502, 503 or 504, a 422 other than Extract's `no_data_extracted`, and a failure to connect. It never retries any other status, a read timeout, or a connection that fails - after the request was sent. A GET is always eligible; a POST or PUT only - with an `Idempotency-Key` header and a body that can be sent again. + after the request was sent. It retries only a GET, HEAD or OPTIONS whose + body is none, a string, bytes, or a dict it encodes. A POST or PUT is + always sent once: the API does not deduplicate a request by + `Idempotency-Key`, so the client does not retry on it. Between tries it waits the server's `Retry-After` when that is whole seconds (at least 0.25, at most 30), or else a random 0.25 to @@ -773,11 +787,14 @@ def _worker( return send() with _call_scope() as state: response = send() + # This loop builds the request again from `data`, so it retries + # only a method it may send again with a body built the same way. + if not (_method_can_be_sent_again(method) and _data_can_be_sent_again(data)): + return response # Retry a 422 here rather than in urllib3, which can't read the # body: Extract's `no_data_extracted` is a final answer, any other # 422 is a failure to get the target's content and is worth # another try. Same count, wait and budget as urllib3's retries. - # The adapter already set `state.can_send_again` for this request. while response.status_code == 422 and not is_no_data_extracted(response): delay = policy.next_delay(state, response.headers.get("Retry-After")) if delay is None: diff --git a/tests/test_retries.py b/tests/test_retries.py index 50ddc16..44000c5 100644 --- a/tests/test_retries.py +++ b/tests/test_retries.py @@ -2,11 +2,12 @@ Retried: 429, 500, 502, 503, 504, a 422 other than `no_data_extracted`, and a failure to connect. Never retried: any other status, a read timeout, or a -connection that drops after the request was sent. A GET is always eligible; a -POST or PUT only with an `Idempotency-Key` header and a body that can be sent -again. Waits honour a whole-seconds `Retry-After` (clamped to [0.25, 30]) or -else use full jitter, and no retry starts past `retry_budget`. Sleeps are -patched out; the budget runs on a fake clock. +connection that drops after the request was sent. Only a GET, HEAD or OPTIONS +whose body is none, a string or bytes is retried; a POST or PUT is sent once, +with or without an `Idempotency-Key` header. Waits honour a whole-seconds +`Retry-After` (clamped to [0.25, 30]) or else use full jitter, and no retry +starts past `retry_budget`. Sleeps are patched out; the budget runs on a fake +clock. """ import asyncio @@ -339,13 +340,15 @@ def test_connection_dropped_after_send_is_not_retried_for_get( assert sleeps == [] -def test_connection_dropped_after_send_is_not_retried_for_post( - server: _Server, sleeps: list[float] +@pytest.mark.parametrize("headers", [None, KEY]) +@pytest.mark.parametrize("verb", ["post", "put"]) +def test_connection_dropped_after_send_is_not_retried_for_post_and_put( + server: _Server, sleeps: list[float], verb: str, headers: dict[str, str] | None ) -> None: server.script = [DROP, (200, {}, b"ok")] with pytest.raises(requests.exceptions.ConnectionError): - client(server, retries=3).post(TARGET, headers=KEY, data="payload") + getattr(client(server, retries=3), verb)(TARGET, headers=headers, data="payload") assert server.calls == 1 assert sleeps == [] @@ -362,85 +365,103 @@ def test_connection_refused_is_retried(sleeps: list[float]) -> None: # ---- which requests may be sent again ---- +@pytest.mark.parametrize("status", [503, 429, 422, 500]) +@pytest.mark.parametrize("headers", [None, KEY, {"idempotency-key": "k"}]) @pytest.mark.parametrize("verb", ["post", "put"]) -def test_post_and_put_without_idempotency_key_are_sent_once( - server: _Server, sleeps: list[float], verb: str +def test_post_and_put_are_sent_once( + server: _Server, + sleeps: list[float], + verb: str, + headers: dict[str, str] | None, + status: int, ) -> None: - server.script = [(503, {}, b""), (200, {}, b"ok")] + server.script = [(status, {}, NO_CONTENT), (200, {}, b"ok")] - response = getattr(client(server), verb)(TARGET, data="payload") + response = getattr(client(server, retries=3), verb)(TARGET, headers=headers, data="payload") - assert response.status_code == 503 - assert server.calls == 1 + # An `Idempotency-Key` changes nothing: the API does not deduplicate on it. + assert response.status_code == status + assert server.bodies == [b"payload"] assert sleeps == [] -def test_post_without_idempotency_key_does_not_retry_a_422( - server: _Server, sleeps: list[float] -) -> None: - server.script = [(422, {}, NO_CONTENT), (200, {}, b"ok")] - - response = client(server).post(TARGET, data="payload") - - assert response.status_code == 422 - assert server.calls == 1 - - -def test_post_without_idempotency_key_does_not_retry_a_failure_to_connect( - sleeps: list[float], +@pytest.mark.parametrize("headers", [None, KEY]) +@pytest.mark.parametrize("verb", ["post", "put"]) +def test_post_and_put_are_sent_once_after_a_failure_to_connect( + sleeps: list[float], verb: str, headers: dict[str, str] | None ) -> None: - zr = ZenRowsClient(APIKEY, base_url=f"http://127.0.0.1:{_closed_port()}/") + zr = ZenRowsClient(APIKEY, retries=3, base_url=f"http://127.0.0.1:{_closed_port()}/") with pytest.raises(requests.exceptions.ConnectionError): - zr.post(TARGET, data="payload") + getattr(zr, verb)(TARGET, headers=headers, data="payload") + # A GET to the same closed port waits before each of its retries. assert sleeps == [] -@pytest.mark.parametrize( - ("verb", "data", "sent"), - [ - ("post", "payload", b"payload"), - ("post", b"payload", b"payload"), - ("put", {"a": "1"}, b"a=1"), - ("post", io.BytesIO(b"payload"), b"payload"), - ], -) -def test_idempotency_key_and_a_body_that_can_be_sent_again_are_retried( - server: _Server, sleeps: list[float], verb: str, data: object, sent: bytes -) -> None: +@pytest.mark.parametrize("verb", ["post_async", "put_async"]) +def test_post_and_put_async_are_sent_once(server: _Server, sleeps: list[float], verb: str) -> None: server.script = [(503, {}, b""), (200, {}, b"ok")] - response = getattr(client(server), verb)(TARGET, headers=KEY, data=data) + response = asyncio.run(getattr(client(server), verb)(TARGET, headers=KEY, data="payload")) - assert response.status_code == 200 - assert server.bodies == [sent, sent] + assert response.status_code == 503 + assert server.calls == 1 + assert sleeps == [] -def test_idempotency_key_matches_any_case(server: _Server, sleeps: list[float]) -> None: - server.script = [(422, {}, NO_CONTENT), (200, {}, b"ok")] +@pytest.mark.parametrize("status", [422, 503]) +def test_post_with_a_file_body_is_sent_once( + server: _Server, sleeps: list[float], status: int +) -> None: + # A 422 retry used to send this body again empty, after the first try had + # read the file to its end. + server.script = [(status, {}, NO_CONTENT), (200, {}, b"ok")] - response = client(server).post(TARGET, headers={"idempotency-key": "k"}, data="payload") + response = client(server).post(TARGET, headers=KEY, data=io.BytesIO(b"payload=1")) - assert response.status_code == 200 - assert server.calls == 2 + assert response.status_code == status + assert response.content == NO_CONTENT + assert server.bodies == [b"payload=1"] + assert sleeps == [] @pytest.mark.parametrize("status", [503, 422]) -@pytest.mark.parametrize("kind", ["generator", "iterator"]) -def test_a_body_that_cannot_be_sent_again_is_never_retried( - server: _Server, sleeps: list[float], status: int, kind: str +@pytest.mark.parametrize("kind", ["file", "generator", "iterator"]) +def test_get_with_a_one_shot_body_is_not_retried( + server: _Server, sleeps: list[float], kind: str, status: int ) -> None: server.script = [(status, {}, NO_CONTENT), (200, {}, b"ok")] chunks = [b"chunk1-", b"chunk2"] - data = (c for c in chunks) if kind == "generator" else iter(chunks) + data = { + "file": lambda: io.BytesIO(b"".join(chunks)), + "generator": lambda: (c for c in chunks), + "iterator": lambda: iter(chunks), + }[kind]() - response = client(server).post(TARGET, headers=KEY, data=data) + response = client(server).fetch(TARGET, data=data) assert response.status_code == status assert server.bodies == [b"chunk1-chunk2"] assert sleeps == [] +@pytest.mark.parametrize("status", [503, 422]) +@pytest.mark.parametrize( + ("data", "sent"), + [("payload", b"payload"), (b"payload", b"payload"), ({"a": "1"}, b"a=1")], +) +def test_get_with_a_string_or_bytes_body_is_retried( + server: _Server, sleeps: list[float], data: object, sent: bytes, status: int +) -> None: + server.script = [(status, {}, NO_CONTENT), (200, {}, b"ok")] + + response = client(server).fetch(TARGET, data=data) + + assert response.status_code == 200 + assert server.bodies == [sent, sent] + assert len(sleeps) == 1 + + # ---- how long it waits ---- From 1815f16ce57e205ac2767a39115005f6ec02c6d3 Mon Sep 17 00:00:00 2001 From: gonzalo-zenrows Date: Tue, 20 Oct 2026 17:37:51 +0200 Subject: [PATCH 4/6] fix(client): never retry a 422 for a request that uploads files The 422 retry builds the request again from its arguments. A file passed as files is read by the first try, so the retry sent an empty file part and could return a 200 for a request that never carried the file. The 422 retry now skips any request with files. A 503 retry is unchanged: it sends the body that requests already built, file included. --- src/zenrows/client.py | 12 +++++++++--- tests/test_retries.py | 22 ++++++++++++++++++++++ 2 files changed, 31 insertions(+), 3 deletions(-) diff --git a/src/zenrows/client.py b/src/zenrows/client.py index d82d81e..986a1e4 100644 --- a/src/zenrows/client.py +++ b/src/zenrows/client.py @@ -787,9 +787,15 @@ def _worker( return send() with _call_scope() as state: response = send() - # This loop builds the request again from `data`, so it retries - # only a method it may send again with a body built the same way. - if not (_method_can_be_sent_again(method) and _data_can_be_sent_again(data)): + # This loop builds the request again from `data` and `files`, so it + # retries only a method it may send again, with a body built the + # same way each time. A file passed as `files` is read by the + # first try, so it is never retried here. + if not ( + _method_can_be_sent_again(method) + and _data_can_be_sent_again(data) + and not kwargs.get("files") + ): return response # Retry a 422 here rather than in urllib3, which can't read the # body: Extract's `no_data_extracted` is a final answer, any other diff --git a/tests/test_retries.py b/tests/test_retries.py index 44000c5..3e82503 100644 --- a/tests/test_retries.py +++ b/tests/test_retries.py @@ -445,6 +445,28 @@ def test_get_with_a_one_shot_body_is_not_retried( assert sleeps == [] +@pytest.mark.parametrize("status", [422, 503]) +def test_get_with_a_file_in_files_never_sends_an_empty_file_part( + server: _Server, sleeps: list[float], status: int +) -> None: + """A 422 retry rebuilds the request from `files`; the first try has read + the file, so the retry would send an empty part. It is never sent again. + A 503 retry sends the body that `requests` already built, file included.""" + server.script = [(status, {}, NO_CONTENT), (200, {}, b"ok")] + + response = client(server).fetch(TARGET, files={"f": ("f.txt", io.BytesIO(b"FILEDATA"))}) + + if status == 422: + assert response.status_code == 422 + assert len(server.bodies) == 1 + assert b"FILEDATA" in server.bodies[0] + assert sleeps == [] + else: + assert response.status_code == 200 + assert len(server.bodies) == 2 + assert all(b"FILEDATA" in body for body in server.bodies) + + @pytest.mark.parametrize("status", [503, 422]) @pytest.mark.parametrize( ("data", "sent"), From 27fb098ba774c0fb8e7346d572f41aae9a854870 Mon Sep 17 00:00:00 2001 From: gonzalo-zenrows Date: Tue, 20 Oct 2026 17:52:17 +0200 Subject: [PATCH 5/6] fix(client): never retry a status that comes from the target site With original_status set, the API answers with the target site's own status. The client cannot tell that status from one the API sent itself, so a 503 from the target was retried as if the API were busy. A status listed in allowed_status_codes is the same: the API passes the target's answer through. The retry adapter now reads both parameters from each request's query and stores the result on the call's retry state, next to the existing per-request flag. The urllib3 retry and the 422 loop both check it. With original_status true, no status is retried. A status listed in allowed_status_codes is not retried. A failure to connect is not a status, so it is still retried. The README lists this case under the statuses the client never retries. --- README.md | 1 + src/zenrows/client.py | 61 ++++++++++++++++++++-- tests/test_retries.py | 114 +++++++++++++++++++++++++++++++++++++++++- 3 files changed, 170 insertions(+), 6 deletions(-) diff --git a/README.md b/README.md index 123ff6b..36ed36b 100644 --- a/README.md +++ b/README.md @@ -283,6 +283,7 @@ It never retries these: 1. Any other status, such as `400`, `401`, `402`, `403` or `404`. 2. A read timeout, or a connection that closes after the request was sent. The server may still be working on that request, and may charge it. 3. A call to `prepare_domain()` or `prepared_domain_status()`. Their `429` and `503` are answers to read, so these calls are sent once; see [Preparing a domain](#preparing-a-domain). +4. Any status when `params` sets `original_status`, and any status listed in `allowed_status_codes`. That status comes from the target site, so another try cannot change it. A failure to connect is still retried. Which requests it retries: diff --git a/src/zenrows/client.py b/src/zenrows/client.py index 986a1e4..26a39ea 100644 --- a/src/zenrows/client.py +++ b/src/zenrows/client.py @@ -25,7 +25,7 @@ from dataclasses import dataclass from functools import partial from typing import Any -from urllib.parse import quote +from urllib.parse import parse_qs, quote, urlsplit import requests import urllib3 @@ -136,6 +136,42 @@ def _can_send_again(request: requests.PreparedRequest) -> bool: return _method_can_be_sent_again(request.method) and _body_can_be_sent_again(request.body) +# The `original_status` values the client reads as true, in any case. The +# docs spell it "true"; requests sends a Python True as "True" and 1 as "1". +_TRUE_VALUES = frozenset({"true", "1"}) + + +@dataclass(frozen=True) +class _TargetStatuses: + """The statuses of one request that come from the target site, not from + the API. Another try cannot change them, so they are never retried. + + With `original_status` set, the API answers with the target's own + status, so every status may be the target's. With + `allowed_status_codes`, the API passes the listed target statuses + through.""" + + every: bool = False + listed: frozenset[int] = frozenset() + + def __contains__(self, status: object) -> bool: + return self.every or status in self.listed + + @classmethod + def from_url(cls, url: str | None) -> "_TargetStatuses": + query = parse_qs(urlsplit(url or "").query, keep_blank_values=True) + every = any(v.strip().lower() in _TRUE_VALUES for v in query.get("original_status", [])) + # A comma-separated list, or the key repeated (requests sends a + # list that way). A token that is not a status code is ignored. + listed = frozenset( + int(token) + for value in query.get("allowed_status_codes", []) + for token in (part.strip() for part in value.split(",")) + if token.isdigit() + ) + return cls(every, listed) + + @dataclass class _CallState: """The retry state of one call. urllib3's retries and the 422 retry in @@ -145,6 +181,9 @@ class _CallState: retries_used: int = 0 # Whether urllib3 may send the current request again; set per request. can_send_again: bool = True + # The current request's statuses that come from the target site, which + # neither urllib3 nor the 422 loop retries; set per request. + target_statuses: _TargetStatuses = _TargetStatuses() # The current call's retry state. A context variable, so concurrent calls on @@ -240,7 +279,9 @@ def increment( retry = super().increment(method, url, response, error, _pool, _stacktrace) retry_after = response.headers.get("Retry-After") if response is not None else None state = retry._current_state() - delay = retry.policy.next_delay(state, retry_after) if state.can_send_again else None + from_target = response is not None and response.status in state.target_statuses + may_retry = state.can_send_again and not from_target + delay = retry.policy.next_delay(state, retry_after) if may_retry else None if delay is None: reason = error or ResponseError("no retry left for this call") raise MaxRetryError(_pool, url, reason) from reason @@ -261,11 +302,13 @@ def sleep(self, response: Any = None) -> None: class _RetryBudgetAdapter(HTTPAdapter): """Gives each request the retry state of the call it belongs to, and - says whether that request may be sent again.""" + says whether that request may be sent again and which of its statuses + come from the target site.""" def send(self, request: Any, *args: Any, **kwargs: Any) -> requests.Response: with _call_scope() as state: state.can_send_again = _can_send_again(request) + state.target_statuses = _TargetStatuses.from_url(request.url) return super().send(request, *args, **kwargs) @@ -421,7 +464,9 @@ class ZenRowsClient: after the request was sent. It retries only a GET, HEAD or OPTIONS whose body is none, a string, bytes, or a dict it encodes. A POST or PUT is always sent once: the API does not deduplicate a request by - `Idempotency-Key`, so the client does not retry on it. + `Idempotency-Key`, so the client does not retry on it. It retries no + status when `params` sets `original_status`, and no status listed in + `allowed_status_codes`: that status comes from the target site. Between tries it waits the server's `Retry-After` when that is whole seconds (at least 0.25, at most 30), or else a random 0.25 to @@ -801,7 +846,13 @@ def _worker( # body: Extract's `no_data_extracted` is a final answer, any other # 422 is a failure to get the target's content and is worth # another try. Same count, wait and budget as urllib3's retries. - while response.status_code == 422 and not is_no_data_extracted(response): + # A 422 from the target site (`original_status`, or listed in + # `allowed_status_codes`) is the target's answer: not retried. + while ( + response.status_code == 422 + and not is_no_data_extracted(response) + and 422 not in state.target_statuses + ): delay = policy.next_delay(state, response.headers.get("Retry-After")) if delay is None: break diff --git a/tests/test_retries.py b/tests/test_retries.py index 3e82503..056616d 100644 --- a/tests/test_retries.py +++ b/tests/test_retries.py @@ -4,7 +4,9 @@ failure to connect. Never retried: any other status, a read timeout, or a connection that drops after the request was sent. Only a GET, HEAD or OPTIONS whose body is none, a string or bytes is retried; a POST or PUT is sent once, -with or without an `Idempotency-Key` header. Waits honour a whole-seconds +with or without an `Idempotency-Key` header. A status from the target site +(any status with `original_status`, or one listed in `allowed_status_codes`) +is never retried. Waits honour a whole-seconds `Retry-After` (clamped to [0.25, 30]) or else use full jitter, and no retry starts past `retry_budget`. Sleeps are patched out; the budget runs on a fake clock. @@ -484,6 +486,116 @@ def test_get_with_a_string_or_bytes_body_is_retried( assert len(sleeps) == 1 +# ---- statuses from the target site ---- + + +@pytest.mark.parametrize("status", [429, 500, 502, 503, 504, 422]) +@pytest.mark.parametrize("value", [True, "true", "TRUE", 1, "1"]) +def test_original_status_turns_off_every_status_retry( + server: _Server, sleeps: list[float], value: object, status: int +) -> None: + server.script = [(status, {}, NO_CONTENT), (200, {}, b"ok")] + + response = client(server, retries=3).fetch(TARGET, params={"original_status": value}) + + # The API passes the target's own status through, and the client + # cannot tell it from a status the API sent itself. + assert response.status_code == status + assert server.calls == 1 + assert sleeps == [] + + +@pytest.mark.parametrize("status", [503, 422]) +@pytest.mark.parametrize( + "params", + [ + None, + {"original_status": False}, + {"original_status": "false"}, + {"original_status": 0}, + {"original_status": "0"}, + {"original_status": ""}, + ], +) +def test_original_status_false_or_absent_still_retries( + server: _Server, sleeps: list[float], params: dict | None, status: int +) -> None: + server.script = [(status, {}, NO_CONTENT), (200, {}, b"ok")] + + response = client(server).fetch(TARGET, params=params) + + assert response.status_code == 200 + assert server.calls == 2 + assert len(sleeps) == 1 + + +@pytest.mark.parametrize("allowed", ["503", "404,503", " 404 , 503 ", ["404", "503"], [404, 503]]) +def test_allowed_status_code_is_not_retried( + server: _Server, sleeps: list[float], allowed: object +) -> None: + server.script = [(503, {}, b""), (200, {}, b"ok")] + + response = client(server, retries=3).fetch(TARGET, params={"allowed_status_codes": allowed}) + + assert response.status_code == 503 + assert server.calls == 1 + assert sleeps == [] + + +@pytest.mark.parametrize("status", [502, 422]) +def test_status_not_in_allowed_status_codes_is_still_retried( + server: _Server, sleeps: list[float], status: int +) -> None: + server.script = [(status, {}, NO_CONTENT), (200, {}, b"ok")] + + response = client(server).fetch(TARGET, params={"allowed_status_codes": "404,503"}) + + assert response.status_code == 200 + assert server.calls == 2 + assert len(sleeps) == 1 + + +def test_422_in_allowed_status_codes_is_not_retried(server: _Server, sleeps: list[float]) -> None: + server.script = [(422, {}, NO_CONTENT), (200, {}, b"ok")] + + response = client(server, retries=3).fetch(TARGET, params={"allowed_status_codes": "404,422"}) + + assert response.status_code == 422 + assert server.calls == 1 + assert sleeps == [] + + +def test_extract_with_original_status_is_not_retried(server: _Server, sleeps: list[float]) -> None: + server.script = [(422, {}, NO_CONTENT), (503, {}, b""), (200, {}, b"ok")] + + response = client(server).extract(TARGET, params={"original_status": "true"}) + + assert response.status_code == 422 + assert server.calls == 1 + assert sleeps == [] + + +def test_original_status_applies_to_its_own_call_only(server: _Server, sleeps: list[float]) -> None: + zr = client(server) + server.script = [(503, {}, b""), (503, {}, b""), (200, {}, b"ok")] + + first = zr.fetch(TARGET, params={"original_status": "true"}) + second = zr.fetch(TARGET) + + assert (first.status_code, second.status_code) == (503, 200) + assert server.calls == 3 + assert len(sleeps) == 1 + + +def test_failure_to_connect_with_original_status_is_still_retried(sleeps: list[float]) -> None: + zr = ZenRowsClient(APIKEY, base_url=f"http://127.0.0.1:{_closed_port()}/") + with pytest.raises(requests.exceptions.ConnectionError): + zr.fetch(TARGET, params={"original_status": "true", "allowed_status_codes": "503"}) + + # A refused connection is not a status: the request never reached the API. + assert len(sleeps) == 2 + + # ---- how long it waits ---- From 367a1fbf5ecf935b8e5d4269436c5b1e091536d4 Mon Sep 17 00:00:00 2001 From: gonzalo-zenrows Date: Tue, 20 Oct 2026 22:24:05 +0200 Subject: [PATCH 6/6] fix(client): never raise on a very long retry-after value The client parsed a whole-digits Retry-After with int(). Python refuses to convert a string of more than 4300 digits, so a 429 or 503 with such a header raised ValueError out of the retry code to the caller. Any value of three or more digits, after leading zeros, is already above the 30 s cap. The client now returns the cap for it without parsing. A value such as 0005 still waits 5 s, and 000 still waits 0.25 s. --- README.md | 2 +- src/zenrows/client.py | 7 ++++++- tests/test_retries.py | 16 ++++++++++++++++ 3 files changed, 23 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index 36ed36b..9261d2b 100644 --- a/README.md +++ b/README.md @@ -294,7 +294,7 @@ Which requests it retries: How long it waits between tries: -1. When the response has a `Retry-After` header in whole seconds, the client waits that long, at least 0.25 seconds and at most 30 seconds. The 30-second cap is fixed; `retry_max_wait` does not change it. +1. When the response has a `Retry-After` header in whole seconds, the client waits that long, at least 0.25 seconds and at most 30 seconds. The 30-second cap is fixed; `retry_max_wait` does not change it. A value of any length works: any number above 30, however long, waits 30 seconds. 2. Otherwise, or when `Retry-After` is an HTTP date, a fraction or any other value, it waits a random time between 0.25 seconds and `min(retry_max_wait, 2 ** attempt)` seconds, where `attempt` counts retries from 0. `retry_max_wait` defaults to 30 seconds. The wait is never shorter than 0.25 seconds. No retry starts after 90 s, counted from the call's first try (`retry_budget`, default 90). The budget does not cut a running try short, so a call can take longer than the budget by one try. `retries`, a `422` retry and a `5xx` retry share one count and one budget in the same call. diff --git a/src/zenrows/client.py b/src/zenrows/client.py index 26a39ea..5f68478 100644 --- a/src/zenrows/client.py +++ b/src/zenrows/client.py @@ -74,7 +74,12 @@ def _retry_after_seconds(value: str | None) -> float | None: value = value.strip() if not _RETRY_AFTER_SECONDS.fullmatch(value): return None - return min(max(float(int(value)), _RETRY_MIN_WAIT), _RETRY_AFTER_CAP) + # Three or more significant digits is already past the cap. Return the cap + # without parsing: int() raises on a string of thousands of digits. + digits = value.lstrip("0") + if len(digits) > 2: + return _RETRY_AFTER_CAP + return min(max(float(int(digits or "0")), _RETRY_MIN_WAIT), _RETRY_AFTER_CAP) def _jitter(attempt: int, retry_max_wait: float) -> float: diff --git a/tests/test_retries.py b/tests/test_retries.py index 056616d..500348f 100644 --- a/tests/test_retries.py +++ b/tests/test_retries.py @@ -606,6 +606,10 @@ def test_failure_to_connect_with_original_status_is_still_retried(sleeps: list[f ("0", 250), (" 7 ", 7000), ("99999", 30000), + pytest.param("9" * 5000, 30000, id="5000-nines-30000"), + ("100", 30000), + ("0005", 5000), + ("000", 250), ("1.5", None), ("-1", None), ("1e3", None), @@ -685,6 +689,18 @@ def test_retry_after_is_honoured( assert sleeps == [waited] +def test_retry_after_of_thousands_of_digits_waits_the_cap( + server: _Server, sleeps: list[float] +) -> None: + server.script = [(429, {"Retry-After": "9" * 5000}, b""), (200, {}, b"ok")] + + response = client(server).fetch(TARGET) + + assert response.status_code == 200 + assert server.calls == 2 + assert sleeps == [30.0] + + def test_retry_after_on_a_422_is_honoured(server: _Server, sleeps: list[float]) -> None: server.script = [(422, {"Retry-After": "4"}, NO_CONTENT), (200, {}, b"ok")]