From 0d0d6dabe24d3d750c1e30d5f0f1d23faca596ec Mon Sep 17 00:00:00 2001 From: Dave Boyd Date: Mon, 22 Jun 2026 10:49:18 -0400 Subject: [PATCH] Add transient-network retry with exponential backoff - bls_client/retry.py: with_retries() retries timeouts/connection drops and 429/5xx with exponential backoff; honors Retry-After; never retries 4xx or BLSQuotaError (those won't fix themselves) - BLSClient(retries=2, backoff=0.5) wraps the data POST and discovery GETs; qcew CSV fetch wrapped too. retries=0 disables. - tests/test_retry.py: 6 offline tests (injectable sleep, no real delays) - README: retry behavior documented; drop the now-resolved limitation Offline suite 125 passing; live smoke green. Co-Authored-By: Claude Opus 4.8 (1M context) --- README.md | 8 +++- bls_client/client.py | 22 ++++++++-- bls_client/qcew.py | 7 ++- bls_client/retry.py | 58 +++++++++++++++++++++++++ tests/test_retry.py | 101 +++++++++++++++++++++++++++++++++++++++++++ 5 files changed, 188 insertions(+), 8 deletions(-) create mode 100644 bls_client/retry.py create mode 100644 tests/test_retry.py diff --git a/README.md b/README.md index 8de460c..3009da6 100644 --- a/README.md +++ b/README.md @@ -68,6 +68,7 @@ bls_client/ ├── series.py low-level series-ID builders (LAUS, CES, CPI, PPI, OES, JOLTS, ECI, ECEC, QCEW, productivity) ├── qcew.py QCEW Open Data CSV client (county/industry detail; no key, no quota) ├── cache.py on-disk response cache (FileCache) +├── retry.py exponential-backoff retry for transient network errors ├── errors.py BLSError / BLSQuotaError / BLSRequestError └── queries/ ├── employment.py payrolls, unemployment (LAUS), JOLTS, QCEW totals @@ -114,6 +115,11 @@ client.queries_used # network calls made this session (ca A blown daily quota raises `BLSQuotaError` (vs `BLSRequestError` for a bad request), so "am I throttled or is my series ID wrong?" is no longer ambiguous. +Transient network failures (timeouts, dropped connections, `429`/`5xx`) are retried with +exponential backoff — `BLSClient(API_KEY, retries=2, backoff=0.5)` (set `retries=0` to +disable). A server `Retry-After` header is honored. The QCEW CSV client retries too. Note +that retries deliberately exclude `BLSQuotaError` and 4xx — those won't fix themselves. + ## QCEW county/industry detail The timeseries helpers give QCEW national/state totals; the `qcew` module reaches the full @@ -128,8 +134,6 @@ hospitals = qcew.industry("622", 2024, "a") # one NAICS across all a ## Known limitations -- **No automatic retry/backoff** on transient network errors (a `requests` failure surfaces - directly). Caching mitigates repeat load but there's no rate-limit pacing. - **Coverage is the headline cut** of each survey, not an exhaustive mirror — e.g. CES is national supersectors + a couple of states; CPI is the common items; OES ships 15 named occupations (any SOC works via `occupation_*`). Broaden the helper dicts as needed. diff --git a/bls_client/client.py b/bls_client/client.py index deee7b6..92af4ff 100644 --- a/bls_client/client.py +++ b/bls_client/client.py @@ -12,6 +12,7 @@ from datetime import datetime from .cache import FileCache from .errors import BLSError, BLSRequestError, BLSQuotaError +from .retry import with_retries class BLSClient: @@ -19,7 +20,8 @@ class BLSClient: MAX_SERIES_PER_CALL = 50 MAX_YEARS_PER_CALL = 20 - def __init__(self, api_key: str, cache=False, cache_dir=None, cache_ttl=86400): + def __init__(self, api_key: str, cache=False, cache_dir=None, cache_ttl=86400, + retries=2, backoff=0.5): """ Args: api_key: BLS registration key. @@ -27,6 +29,9 @@ class BLSClient: or pass a custom cache object with get(key)/set(key, value). cache_dir: Override the default cache directory. cache_ttl: Cache entry lifetime in seconds (default 1 day). + retries: retries after the first attempt on transient network errors + (timeouts, connection drops, 429/5xx). 0 disables retrying. + backoff: base exponential-backoff interval in seconds. """ self.api_key = api_key if cache is True: @@ -35,8 +40,15 @@ class BLSClient: self.cache = cache # caller-supplied cache object else: self.cache = None + self.retries = retries + self.backoff = backoff self.queries_used = 0 # network POSTs spent this session (cache hits don't count) + def _get(self, url, **kwargs): + """GET with retry/backoff.""" + return with_retries(lambda: requests.get(url, **kwargs), + retries=self.retries, backoff=self.backoff) + # ------------------------------------------------------------------ # Core fetch # ------------------------------------------------------------------ @@ -97,7 +109,9 @@ class BLSClient: def _fetch_batch(self, payload: dict) -> dict: """Execute one (uncached) API call and return {series_id: {data, catalog}}.""" - r = requests.post(f"{self.BASE_URL}/timeseries/data/", json=payload, timeout=30) + url = f"{self.BASE_URL}/timeseries/data/" + r = with_retries(lambda: requests.post(url, json=payload, timeout=30), + retries=self.retries, backoff=self.backoff) r.raise_for_status() self.queries_used += 1 body = r.json() @@ -155,13 +169,13 @@ class BLSClient: # ------------------------------------------------------------------ def surveys(self) -> list[dict]: """Return all BLS surveys with abbreviation and name.""" - r = requests.get(f"{self.BASE_URL}/surveys", timeout=15) + r = self._get(f"{self.BASE_URL}/surveys", timeout=15) r.raise_for_status() return r.json()["Results"]["survey"] def popular(self, survey: str) -> list[str]: """Return popular series IDs for a given survey abbreviation.""" - r = requests.get( + r = self._get( f"{self.BASE_URL}/timeseries/popular", params={"survey": survey}, timeout=15, diff --git a/bls_client/qcew.py b/bls_client/qcew.py index 7d5f8f6..12e16f1 100644 --- a/bls_client/qcew.py +++ b/bls_client/qcew.py @@ -21,6 +21,8 @@ import io import requests +from .retry import with_retries + QCEW_BASE = "https://data.bls.gov/cew/data/api" _HEADERS = {"User-Agent": "bls-data-library (https://gitea.doesworks.net/giteaadmin/bls-data)"} @@ -35,8 +37,9 @@ OWNERSHIP = { } -def _fetch_csv(url: str) -> list[dict]: - r = requests.get(url, timeout=60, headers=_HEADERS) +def _fetch_csv(url: str, retries=2, backoff=0.5) -> list[dict]: + r = with_retries(lambda: requests.get(url, timeout=60, headers=_HEADERS), + retries=retries, backoff=backoff) r.raise_for_status() return parse_csv(r.text) diff --git a/bls_client/retry.py b/bls_client/retry.py new file mode 100644 index 0000000..540af79 --- /dev/null +++ b/bls_client/retry.py @@ -0,0 +1,58 @@ +"""Retry transient network failures with exponential backoff. + +Shared by the JSON client (POST) and the QCEW CSV client (GET). Retries only +genuinely transient conditions — connection drops, timeouts, and 429/5xx — never +4xx client errors or a BLS quota rejection (those won't fix themselves). +""" +import time + +import requests + +# HTTP statuses worth retrying (429 = throttled, 5xx = server-side transient). +RETRY_STATUS = {429, 500, 502, 503, 504} + +# Network-level exceptions worth retrying. +RETRY_EXCEPTIONS = ( + requests.exceptions.ConnectionError, + requests.exceptions.Timeout, + requests.exceptions.ChunkedEncodingError, +) + + +def _retry_after(resp) -> float | None: + """Honor a numeric Retry-After header if the server sent one.""" + ra = getattr(resp, "headers", {}).get("Retry-After") if resp is not None else None + if ra and str(ra).isdigit(): + return float(ra) + return None + + +def with_retries(send, *, retries=2, backoff=0.5, backoff_max=8.0, sleep=time.sleep): + """ + Call ``send()`` (which returns a requests.Response), retrying transient + failures. Backoff is ``backoff * 2**attempt`` seconds, capped at ``backoff_max``. + + Args: + send: zero-arg callable issuing the request and returning a Response. + retries: max retries after the first attempt (so retries=2 -> 3 tries). + backoff: base backoff in seconds. + backoff_max: cap on a single backoff interval. + sleep: sleep function (injectable for tests). + """ + attempt = 0 + while True: + try: + resp = send() + except RETRY_EXCEPTIONS: + if attempt >= retries: + raise + sleep(min(backoff * (2 ** attempt), backoff_max)) + attempt += 1 + continue + + if getattr(resp, "status_code", None) in RETRY_STATUS and attempt < retries: + sleep(_retry_after(resp) or min(backoff * (2 ** attempt), backoff_max)) + attempt += 1 + continue + + return resp diff --git a/tests/test_retry.py b/tests/test_retry.py new file mode 100644 index 0000000..2a635cf --- /dev/null +++ b/tests/test_retry.py @@ -0,0 +1,101 @@ +"""Offline tests for the retry/backoff layer. No network, no real sleeping.""" +import pytest +import requests + +import bls_client.client as client_mod +from bls_client import BLSClient +from bls_client.retry import with_retries + + +class Resp: + def __init__(self, status_code=200, headers=None): + self.status_code = status_code + self.headers = headers or {} + + def raise_for_status(self): + pass + + def json(self): + return { + "status": "REQUEST_SUCCEEDED", + "Results": {"series": [{"seriesID": "X", "data": [{"year": "2024", "period": "M01", "value": "1"}]}]}, + } + + +def _recorder(): + delays = [] + return delays, (lambda d: delays.append(d)) + + +def test_retries_transient_exception_then_succeeds(): + calls = {"n": 0} + delays, sleep = _recorder() + + def send(): + calls["n"] += 1 + if calls["n"] < 3: + raise requests.exceptions.ConnectionError("boom") + return Resp(200) + + resp = with_retries(send, retries=3, backoff=0.5, sleep=sleep) + assert resp.status_code == 200 + assert calls["n"] == 3 + assert delays == [0.5, 1.0] # exponential backoff between the 3 attempts + + +def test_gives_up_after_max_retries(): + delays, sleep = _recorder() + + def send(): + raise requests.exceptions.Timeout("slow") + + with pytest.raises(requests.exceptions.Timeout): + with_retries(send, retries=2, backoff=0.5, sleep=sleep) + assert len(delays) == 2 # retried twice, then re-raised + + +def test_retries_5xx_but_not_4xx(): + delays, sleep = _recorder() + seq = [Resp(503), Resp(200)] + resp = with_retries(lambda: seq.pop(0), retries=3, backoff=0.5, sleep=sleep) + assert resp.status_code == 200 and len(delays) == 1 + + delays2, sleep2 = _recorder() + resp2 = with_retries(lambda: Resp(404), retries=3, backoff=0.5, sleep=sleep2) + assert resp2.status_code == 404 and delays2 == [] # client error: not retried + + +def test_honors_retry_after_header(): + delays, sleep = _recorder() + seq = [Resp(429, headers={"Retry-After": "7"}), Resp(200)] + with_retries(lambda: seq.pop(0), retries=2, backoff=0.5, sleep=sleep) + assert delays == [7.0] + + +def test_client_retries_then_succeeds(monkeypatch): + # no real sleeping + monkeypatch.setattr("bls_client.retry.time.sleep", lambda d: None) + state = {"n": 0} + + def fake_post(url, json=None, timeout=None): + state["n"] += 1 + if state["n"] == 1: + raise requests.exceptions.ConnectionError("transient") + return Resp(200) + + monkeypatch.setattr(client_mod.requests, "post", fake_post) + c = BLSClient("key", retries=2) + out = c.fetch(["CES0000000001"], 2024, 2024) + assert out and state["n"] == 2 # failed once, retried, succeeded + assert c.queries_used == 1 # counter only counts the successful POST + + +def test_retries_zero_disables(monkeypatch): + monkeypatch.setattr("bls_client.retry.time.sleep", lambda d: None) + + def fake_post(url, json=None, timeout=None): + raise requests.exceptions.ConnectionError("transient") + + monkeypatch.setattr(client_mod.requests, "post", fake_post) + with pytest.raises(requests.exceptions.ConnectionError): + BLSClient("key", retries=0).fetch(["CES0000000001"], 2024, 2024)