diff --git a/README.md b/README.md index ff259f6..9261d2b 100644 --- a/README.md +++ b/README.md @@ -31,6 +31,7 @@ This package ships three clients: - [Concurrency](#concurrency) - [Extract](#extract) - [Preparing a domain](#preparing-a-domain) + - [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) @@ -58,14 +59,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`. -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 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={ @@ -131,7 +132,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={ @@ -142,6 +143,8 @@ response = client.post(url, data={ print(response.text) ``` +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 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. @@ -149,7 +152,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={ @@ -160,6 +163,8 @@ response = client.put(url, data={ print(response.text) ``` +The client never retries a PUT either. It sends it once. + ### 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. @@ -172,7 +177,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 = [ @@ -263,6 +268,49 @@ Both methods raise `PreparedDomainError` on an error answer, with `status_code`, 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. +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: + +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: + +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. + +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.fetch("https://www.zenrows.com/") +``` + ## Quickstart — Batch API (`ZenRowsBatchClient`) > Batch is open to every account and spends the same credits as a regular diff --git a/pyproject.toml b/pyproject.toml index 11ac87f..492647e 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 bae8a5e..5f68478 100644 --- a/src/zenrows/client.py +++ b/src/zenrows/client.py @@ -12,18 +12,25 @@ """ import asyncio +import contextvars import math import os +import random +import re import time import warnings +from collections.abc import Iterator, Mapping from concurrent.futures import ThreadPoolExecutor +from contextlib import contextmanager +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 from requests.adapters import HTTPAdapter +from urllib3.exceptions import MaxRetryError, ResponseError from urllib3.util.retry import Retry from zenrows.__version__ import __version__ @@ -31,11 +38,284 @@ 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 +# 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 -# response body (see `is_no_data_extracted`). -_RETRY_STATUSES = [429, 500, 502, 503, 504] -_RETRY_BACKOFF_FACTOR = 0.5 +# body (see `is_no_data_extracted`), which urllib3 cannot read. +_RETRY_STATUSES = (429, 500, 502, 503, 504) +DEFAULT_RETRIES = 2 +# 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 +# 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 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]+") + +# 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. +_monotonic = time.monotonic + + +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 + # 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: + """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 _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 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) + + +# 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 + `_worker` share it, so they share one retry count and one budget.""" + + started: float + 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 +# the client's thread pool each see their own. +_call_state: contextvars.ContextVar[_CallState | None] = contextvars.ContextVar( + "zenrows_call_state", default=None +) + + +@contextmanager +def _call_scope() -> Iterator[_CallState]: + """The current call's retry state, or a new one when there is none yet.""" + state = _call_state.get() + if state is not None: + yield state + return + state = _CallState(started=_monotonic()) + 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 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: + return None + state.retries_used += 1 + return delay + + +class _JitteredRetry(Retry): + """urllib3 `Retry` that follows the client's `_RetryPolicy`. + + 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`). + """ + + # urllib3 also retries 413 when it carries `Retry-After`; we don't. + RETRY_AFTER_STATUS_CODES = frozenset({429, 503}) + + def __init__(self, *args: Any, policy: _RetryPolicy | None = None, **kwargs: Any): + super().__init__(*args, **kwargs) + 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.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, + 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_after = response.headers.get("Retry-After") if response is not None else None + state = retry._current_state() + 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 + 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 _jitter(len(self.history), self.policy.retry_max_wait) + + def sleep(self, response: Any = None) -> None: + time.sleep(self.get_backoff_time()) + + +class _RetryBudgetAdapter(HTTPAdapter): + """Gives each request the retry state of the call it belongs to, and + 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) + # 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. @@ -182,6 +462,26 @@ 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, 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. 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. 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 + `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. + + `prepare_domain()` and `prepared_domain_status()` never retry: their 429 + and 503 are answers to read. """ # Kept as a class attribute for compatibility — pre-1.4 callers @@ -191,27 +491,44 @@ 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.") + _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 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 + # None when retries are off. `_worker` uses it for the 422 retry. + self._retry_policy: _RetryPolicy | None = None if retries > 0: - max_retries = Retry( + 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, - 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, # `_can_send_again` decides instead status_forcelist=_RETRY_STATUSES, raise_on_status=False, + policy=self._retry_policy, ) - adapter = HTTPAdapter(max_retries=max_retries) + adapter = _RetryBudgetAdapter(max_retries=max_retries) self.requests_session.mount("https://", adapter) self.requests_session.mount("http://", adapter) @@ -506,7 +823,8 @@ def _worker( final_headers["Connection"] = None final_headers.update(headers) - response = self.requests_session.request( + send = partial( + self.requests_session.request, method, self.api_url, params=final_params, @@ -514,25 +832,35 @@ 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 + policy = self._retry_policy + if policy is None: + return send() + with _call_scope() as state: + response = send() + # 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 + # 422 is a failure to get the target's content and is worth + # another try. Same count, wait and budget as urllib3's retries. + # 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 + time.sleep(delay) + response = send() + return response diff --git a/tests/test_retries.py b/tests/test_retries.py index 0fb37d5..500348f 100644 --- a/tests/test_retries.py +++ b/tests/test_retries.py @@ -1,114 +1,941 @@ +"""The ZenRowsClient retry policy, end to end against a local HTTP server. + +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. 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. 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. +""" + import asyncio -from unittest import TestCase +import io +import math +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 import PreparedDomainError, ZenRowsClient, is_no_data_extracted +from zenrows.client import _jitter, _JitteredRetry, _retry_after_seconds -apikey = "APIKEY" -url = "http://example.com" -api_url_base = "https://api.zenrows.com/v1/" +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. 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 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: + 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 + 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 -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) + do_GET = do_POST = do_PUT = _answer + + def log_message(self, *args: object) -> None: + pass + + 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) + + 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("time.sleep", side_effect=waited.append): + yield waited + + +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 ---- + + +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 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, + ) - 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) +def test_retries_zero_mounts_no_retry(server: _Server, sleeps: list[float]) -> None: + zr = client(server, retries=0) + server.script = [(503, {}, b"")] - mock_retry.assert_called_once_with( - total=2, - backoff_factor=0.5, - status_forcelist=list(_RETRY_STATUSES), - raise_on_status=False, - ) + response = zr.fetch(TARGET) - @patch.object(Session, "mount") - def test_adapter_is_mount(self, mock_mount): - ZenRowsClient(apikey, retries=2) + assert type(zr.requests_session.get_adapter(server.url)) is HTTPAdapter + assert response.status_code == 503 + assert server.calls == 1 + assert sleeps == [] - self.assertEqual(mock_mount.call_count, 4) +def test_retries_zero_does_not_retry_a_422(server: _Server, sleeps: list[float]) -> None: + server.script = [(422, {}, NO_CONTENT)] -def _fake_response(status_code: int, body: bytes = b"") -> Response: - response = Response() - response.status_code = status_code - response._content = body - return response + response = client(server, retries=0).fetch(TARGET) + assert response.status_code == 422 + assert server.calls == 1 + assert sleeps == [] -NO_DATA = b'{"code": "no_data_extracted", "title": "No data extracted"}' -NO_CONTENT = b'{"code": "RESP001", "title": "Could not get content"}' + +def test_default_retries_twice_then_returns_the_last_response( + server: _Server, sleeps: list[float] +) -> None: + 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, 500, 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 + + +@pytest.mark.parametrize( + ("status", "body"), + [(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 +) -> None: + server.script = [(status, {}, body)] + + response = client(server, retries=3).fetch(TARGET) + + assert response.status_code == status + assert server.calls == 1 + 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"")] + + 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_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 == [] + + +@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): + getattr(client(server, retries=3), verb)(TARGET, headers=headers, data="payload") + + assert server.calls == 1 + assert sleeps == [] + + +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 + + +# ---- 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_are_sent_once( + server: _Server, + sleeps: list[float], + verb: str, + headers: dict[str, str] | None, + status: int, +) -> None: + server.script = [(status, {}, NO_CONTENT), (200, {}, b"ok")] + + response = getattr(client(server, retries=3), verb)(TARGET, headers=headers, data="payload") + + # An `Idempotency-Key` changes nothing: the API does not deduplicate on it. + assert response.status_code == status + assert server.bodies == [b"payload"] + assert sleeps == [] + + +@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, retries=3, base_url=f"http://127.0.0.1:{_closed_port()}/") + with pytest.raises(requests.exceptions.ConnectionError): + 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", ["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 = asyncio.run(getattr(client(server), verb)(TARGET, headers=KEY, data="payload")) + + assert response.status_code == 503 + assert server.calls == 1 + assert sleeps == [] + + +@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=KEY, data=io.BytesIO(b"payload=1")) + + 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", ["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 = { + "file": lambda: io.BytesIO(b"".join(chunks)), + "generator": lambda: (c for c in chunks), + "iterator": lambda: iter(chunks), + }[kind]() + + response = client(server).fetch(TARGET, data=data) + + assert response.status_code == status + assert server.bodies == [b"chunk1-chunk2"] + 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"), + [("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 + + +# ---- 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 -@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.""" +def test_422_in_allowed_status_codes_is_not_retried(server: _Server, sleeps: list[float]) -> None: + server.script = [(422, {}, NO_CONTENT), (200, {}, b"ok")] - @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 = client(server, retries=3).fetch(TARGET, params={"allowed_status_codes": "404,422"}) - response = ZenRowsClient(apikey, retries=3).extract(url) + assert response.status_code == 422 + assert server.calls == 1 + assert sleeps == [] - 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_extract_with_original_status_is_not_retried(server: _Server, sleeps: list[float]) -> None: + server.script = [(422, {}, NO_CONTENT), (503, {}, b""), (200, {}, b"ok")] - response = ZenRowsClient(apikey, retries=3).fetch(url) + response = client(server).extract(TARGET, params={"original_status": "true"}) - 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]) + assert response.status_code == 422 + assert server.calls == 1 + assert sleeps == [] - @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")] - response = ZenRowsClient(apikey, retries=3).fetch(url) +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")] - self.assertEqual(response.status_code, 200) - self.assertEqual(mock_request.call_count, 2) + first = zr.fetch(TARGET, params={"original_status": "true"}) + second = zr.fetch(TARGET) - @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) + assert (first.status_code, second.status_code) == (503, 200) + assert server.calls == 3 + assert len(sleeps) == 1 - ZenRowsClient(apikey).fetch(url) - self.assertEqual(mock_request.call_count, 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"}) - @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) + # A refused connection is not a status: the request never reached the API. + assert len(sleeps) == 2 - response = asyncio.run(ZenRowsClient(apikey, retries=3).extract_async(url)) - self.assertEqual(response.status_code, 422) - self.assertEqual(mock_request.call_count, 1) +# ---- how long it waits ---- + + +@pytest.mark.parametrize( + ("value", "millis"), + [ + ("5", 5000), + ("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), + ("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.random", return_value=1.0): + response = client(server, retries=4).fetch(TARGET) + + assert response.status_code == 200 + assert sleeps == [1.0, 2.0, 4.0, 8.0] + + +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 + + +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) + + assert sleeps == [0.25, 0.25] + + +@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 == [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")] + + client(server).fetch(TARGET) + + assert sleeps == [4.0] + + +@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=retry_max_wait).fetch(TARGET) + + assert sleeps == [20.0, 30.0] + + +@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")] + + with patch("zenrows.client.random.random", return_value=0.5): + response = client(server).fetch(TARGET) + + assert response.status_code == 200 + assert sleeps == [0.625] + + +# ---- 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("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 == [] + + +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 ---- + + +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_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_extract_returns_a_403_as_is_in_one_request(server: _Server, sleeps: list[float]) -> None: + server.script = [(403, {"Retry-After": "1"}, DOMAIN_NOT_PREPARED), (200, {}, b"{}")] + + response = client(server).extract(TARGET) + + assert response.status_code == 403 + assert server.calls == 1 + assert sleeps == [] + + +# ---- async calls ---- + + +def test_fetch_async_shares_one_retry_count(server: _Server, sleeps: list[float]) -> None: + server.script = [(503, {}, b""), (422, {}, NO_CONTENT), (422, {}, NO_CONTENT), (200, {}, b"ok")] + + response = asyncio.run(client(server).fetch_async(TARGET)) + + # Two retries in all, on the client's thread pool too. + assert response.status_code == 422 + assert server.calls == 3 + assert len(sleeps) == 2 + + +def test_fetch_async_budget_is_per_call(server: _Server, clock: tuple[_Clock, list[float]]) -> None: + _, sleeps = clock + zr = client(server, retries=1, retry_budget=40.0) + + async def two_calls() -> list[int]: + statuses = [] + for _ in range(2): + server.script = [(503, {"Retry-After": "30"}, b""), (200, {}, b"ok")] + statuses.append((await zr.fetch_async(TARGET)).status_code) + return statuses + + # A budget shared across calls would stop the second retry at 60 s > 40 s. + assert asyncio.run(two_calls()) == [200, 200] + assert sleeps == [30.0, 30.0] + + +# ---- prepared domains ---- + + +@pytest.mark.parametrize("status", [429, 503]) +@pytest.mark.parametrize("use_async", [False, True]) +@pytest.mark.parametrize("call", ["prepare_domain", "prepared_domain_status"]) +def test_prepared_domain_requests_are_never_retried( + server: _Server, sleeps: list[float], status: int, use_async: bool, call: str +) -> None: + server.script = [(status, {"Retry-After": "1"}, b'{"code": "busy"}'), (200, {}, b"{}")] + zr = client(server, retries=3) + args = ("example.com", [TARGET]) if call == "prepare_domain" else ("example.com",) + method = getattr(zr, f"{call}_async" if use_async else call) + + with pytest.raises(PreparedDomainError) as caught: + asyncio.run(method(*args)) if use_async else method(*args) + + # The first answer is the result: no second request and no wait. + assert caught.value.status_code == status + assert server.calls == 1 + assert sleeps == [] + + +# ---- 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 9d43168..f30fa27 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]