From d8e93f6406a8e65ff8cc8539bdcab2ddbca4d59e Mon Sep 17 00:00:00 2001 From: thodson-usgs Date: Mon, 10 Aug 2026 17:29:00 -0500 Subject: [PATCH 1/3] refactor(ratings): drive both stages through the shared executor Four page walks existed in the package; three went through transport.pagination.paginate. The fourth was ratings' hand-rolled synchronous STAC loop, and its per-feature RDB downloads were a serial for-loop of N sequential GETs -- so ratings alone had no per-attempt retry, no stall budget, no progress line, no resumable interruption, and no bounded concurrency. Route both stages through the executor every other getter uses (stats.py's one-item fan-out was the template; ADR 0008's plan shape -- a plain list -- covers the downloads with no adapter class): - _search drives paginate as a one-item FanOut with STAC strategies: pages wrap raw feature dicts in a single ``feature`` column, the ``next`` href still passes through resolve_next_url's shared cross-host/credential policy, and a mid-search 429 now surfaces as a resumable QuotaExhausted instead of a raw RateLimited. - The downloads fan out over the feature list itself, bounded by API_USGS_CONCURRENT, with headers still evaluated per asset href. - The log-and-skip swallow is gone -- deliberately. A deterministic per-feature failure (missing asset, malformed RDB, 404) now surfaces typed instead of silently dropping that rating from the result dict, and a transient one is retried then raised resumable. Silent data loss under rate limiting was the old behavior's worst case. A new fitness function pins that ratings can no longer import the executing sync transport helpers, alongside the existing wateruse rule. Co-Authored-By: Claude Fable 5 --- dataretrieval/waterdata/ratings.py | 205 +++++++++++++++++++---------- tests/architecture_test.py | 25 ++++ tests/headers_host_scoping_test.py | 13 +- tests/waterdata_ratings_test.py | 34 +++++ 4 files changed, 199 insertions(+), 78 deletions(-) diff --git a/dataretrieval/waterdata/ratings.py b/dataretrieval/waterdata/ratings.py index ff852029..35e947ba 100644 --- a/dataretrieval/waterdata/ratings.py +++ b/dataretrieval/waterdata/ratings.py @@ -9,7 +9,6 @@ from __future__ import annotations -import logging import os from collections.abc import Iterable from typing import Any, Literal, get_args @@ -17,29 +16,26 @@ import httpx import pandas as pd -from dataretrieval.exceptions import DataRetrievalError from dataretrieval.ogc.dates import _DURATION_RE, _format_api_dates from dataretrieval.ogc.errors import _raise_for_non_200 from dataretrieval.ogc.filters import _quote_cql_str from dataretrieval.ogc.requests import _check_monitoring_location_id from dataretrieval.rdb import extract_rdb_comment, read_rdb -from dataretrieval.transport.http import ( - HTTPX_DEFAULTS, -) +from dataretrieval.transport.fanout import FanOut, active_client from dataretrieval.transport.http import ( default_headers as _default_headers, ) from dataretrieval.transport.http import ( - get as _get, + open_async_client as _open_async_client, ) from dataretrieval.transport.links import resolve_next_url +from dataretrieval.transport.pagination import paginate +from dataretrieval.transport.retry import RetryPolicy from dataretrieval.waterdata.endpoints import STAC_URL __all__ = ["get_ratings"] -logger = logging.getLogger(__name__) - RATING_FILE_TYPE = Literal["exsa", "base", "corr"] _VALID_FILE_TYPES = get_args(RATING_FILE_TYPE) @@ -119,8 +115,19 @@ def get_ratings( Raises ------ ValueError - For an unrecognized ``file_type`` value or an ISO 8601 duration in - ``time``. + For an unrecognized ``file_type`` value, an ISO 8601 duration in + ``time``, or a matching feature that carries no data asset. + DataRetrievalError + The typed subclass for an HTTP error response (see + :func:`transport.pagination.paginate`); + or :class:`~dataretrieval.exceptions.NetworkError` if a request + can't reach the service in a way retrying cannot fix. + FanOutInterrupted + A transient failure (429 / 5xx / timeout) survived the built-in + retries during the search or a download. ``exc.call.resume()`` + finishes the interrupted stage (see :doc:`/userguide/errors`); the + assembled per-feature dict is returned by a fresh ``get_ratings`` + call. Examples -------- @@ -186,25 +193,7 @@ def get_ratings( if file_path is not None: os.makedirs(file_path, exist_ok=True) - out: dict[str, pd.DataFrame] = {} - for feature in matching: - fid = feature["id"] - try: - out[fid] = _download_and_parse(feature, file_path, ssl_check) - # One bad feature shouldn't abort the batch: log and skip the module's - # typed errors (DataRetrievalError, e.g. an HTTPError from - # _raise_for_non_200) plus the transport / parse / file / missing-asset - # errors a single download can raise. - except ( - DataRetrievalError, - httpx.HTTPError, - LookupError, - OSError, - ValueError, - ) as e: - logger.warning("Failed to download / parse %s: %s", fid, e) - - return out + return _download_all(matching, file_path, ssl_check) def _as_list(x: str | Iterable[str]) -> list[str]: @@ -242,8 +231,16 @@ def _search( ``limit`` is the page size (clamped to the service maximum of 10,000); the STAC ``next`` link is followed until exhausted so a result set larger than one page isn't silently truncated. - """ + The page walk is :func:`dataretrieval.transport.pagination.paginate` with + STAC strategies, driven as a one-item + :class:`~dataretrieval.transport.fanout.FanOut` -- the same executor every + other getter uses -- so the search gets per-attempt retry, the stall + budget, a progress line, and the resumable interruption taxonomy instead + of a bespoke sync loop that had none of them. Pages carry features rather + than rows, so each page frame wraps the raw feature dicts in a single + ``feature`` column. + """ query_params: dict[str, Any] = {"limit": min(limit, 10000)} if filter_str is not None: query_params["filter"] = filter_str @@ -252,60 +249,122 @@ def _search( if bbox is not None: query_params["bbox"] = ",".join(map(str, bbox)) - url: str | None = f"{STAC_URL}/search" - # ``params`` is sent only on the first request; each STAC ``next`` link - # already carries the query, so it is reset to None inside the loop. - params: dict[str, Any] | None = query_params - features: list[dict[str, Any]] = [] - while url is not None: - response = _get( - url, - params=params, - headers=_default_headers(url), - verify=ssl_check, - **HTTPX_DEFAULTS, - ) - _raise_for_non_200(response) - body = response.json() - features.extend(body.get("features", [])) + url = f"{STAC_URL}/search" + req = httpx.Request("GET", url, params=query_params, headers=_default_headers(url)) + + def parse_response(resp: httpx.Response) -> tuple[pd.DataFrame, str | None]: + body = resp.json() + page = pd.DataFrame({"feature": body.get("features", [])}) # The STAC ``next`` link is a fully-formed GET href carrying the - # limit/filter/bbox and a continuation token, so follow it verbatim - # (dropping our own params) until the server stops emitting one. + # limit/filter/bbox and a continuation token, so it becomes the + # cursor verbatim -- except for the shared safety policy: the href is + # response data, so it is checked before it becomes a request. A link + # to another host would carry this request's API key off the + # authorized host, and one carrying ``user:pass@`` would mint an + # ``Authorization: Basic`` header the caller never configured. href = next( (lnk["href"] for lnk in body.get("links", []) if lnk.get("rel") == "next"), None, ) - # Verbatim except for the credentials: the href is response data, so it - # is checked before it becomes a request. A link to another host would - # carry this request's API key off the authorized host, and one carrying - # ``user:pass@`` would mint an ``Authorization: Basic`` header the caller - # never configured. - url = ( - None - if href is None - else resolve_next_url(href, response, service="ratings") + cursor = ( + None if href is None else resolve_next_url(href, resp, service="ratings") + ) + return page, cursor + + async def follow_up(cursor: str, sess: httpx.AsyncClient) -> httpx.Response: + return await sess.get(cursor, headers=_default_headers(cursor)) + + async def fetch(request: httpx.Request) -> tuple[pd.DataFrame, httpx.Response]: + return await paginate( + request, + parse_response=parse_response, + follow_up=follow_up, + # Borrow the executor's shared client for every page. + client=active_client(), + raise_for_status=_raise_for_non_200, ) - params = None - return features - -def _download_and_parse( - feature: dict[str, Any], - file_path: str | None, - ssl_check: bool, -) -> pd.DataFrame: - """Fetch the feature's data asset, parse RDB, optionally persist to disk.""" - url = feature["assets"]["data"]["href"] - response = _get( - url, headers=_default_headers(url), verify=ssl_check, **HTTPX_DEFAULTS - ) + df, _ = FanOut( + [req], + fetch, + RetryPolicy.from_env(), + client_options={"verify": ssl_check}, + canonical_url=str(req.url), + service="ratings", + ).resume() + return [] if df.empty else list(df["feature"]) + + +async def _fetch_rating( + feature: dict[str, Any], file_path: str | None +) -> tuple[pd.DataFrame, httpx.Response]: + """Fetch one feature's data asset, parse RDB, optionally persist to disk. + + Headers are evaluated against each asset href -- assets can live on a + different host than the catalog, and must not inherit its auth. Borrows + the executor's shared client; outside a drive (a direct unit call) it + opens a short-lived one. + """ + fid = feature["id"] + href = feature.get("assets", {}).get("data", {}).get("href") + if not href: + raise ValueError(f"STAC feature {fid!r} carries no data asset href.") + headers = _default_headers(href) + session = active_client() + if session is not None: + response = await session.get(href, headers=headers) + else: + async with _open_async_client() as own: + response = await own.get(href, headers=headers) _raise_for_non_200(response) if file_path is not None: - with open(os.path.join(file_path, feature["id"]), "w") as f: + with open(os.path.join(file_path, fid), "w") as f: f.write(response.text) df = read_rdb(response.text) df.attrs["comment"] = extract_rdb_comment(response.text) - df.attrs["url"] = url - return df + df.attrs["url"] = href + return df, response + + +def _download_all( + features: list[dict[str, Any]], + file_path: str | None, + ssl_check: bool, +) -> dict[str, pd.DataFrame]: + """Download every feature's rating over the shared fan-out executor. + + The plan is the feature list itself -- ``FanOut`` asks a plan only to be + sized and iterable -- so the downloads get bounded concurrency, + per-attempt retry, the progress line, and the resumable interruption + taxonomy in place of the previous serial loop, which had none of them and + logged-and-skipped every failure. A deterministic per-feature failure (a + missing asset, a malformed RDB) now surfaces typed instead of silently + dropping that rating from the result; a transient one is retried and, if + retries are exhausted, raised as a resumable interruption. + + The public result is a dict keyed by feature id, so the fetch closure + accumulates it; the executor's combined frame is not the return shape and + is discarded. + """ + out: dict[str, pd.DataFrame] = {} + if not features: + return out + + async def fetch(feature: dict[str, Any]) -> tuple[pd.DataFrame, httpx.Response]: + df, response = await _fetch_rating(feature, file_path) + out[feature["id"]] = df + return df, response + + FanOut( + features, + fetch, + RetryPolicy.from_env(), + client_options={"verify": ssl_check}, + # No single URL expresses "all of these assets" -- the aggregate + # reports the first, matching what a single-feature call would show. + canonical_url=features[0].get("assets", {}).get("data", {}).get("href"), + service="ratings", + ).resume() + return out diff --git a/tests/architecture_test.py b/tests/architecture_test.py index 0fff59ef..727ccc82 100644 --- a/tests/architecture_test.py +++ b/tests/architecture_test.py @@ -538,6 +538,31 @@ def test_wateruse_does_not_reimplement_fan_out_orchestration() -> None: ) +def test_ratings_drives_http_through_the_shared_executor() -> None: + """Ratings must not issue one-off synchronous HTTP requests. + + Its STAC page walk and per-feature downloads previously ran hand-rolled + sync loops over ``transport.http.get`` -- no retry, no stall budget, no + progress line, no resume, and N serial downloads. Both stages now drive + ``paginate``/``FanOut`` like every other multi-request path; assert the + direct sync entry points cannot quietly return. + """ + path = PACKAGE_ROOT / "waterdata" / "ratings.py" + tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) + transport_names = { + alias.name + for node in ast.walk(tree) + if isinstance(node, ast.ImportFrom) + and node.module == "dataretrieval.transport.http" + for alias in node.names + } + offenders = transport_names - {"default_headers", "open_async_client"} + assert not offenders, ( + "ratings imports executing sync transport helpers instead of driving " + f"the shared executor: {sorted(offenders)}" + ) + + def test_fan_out_plans_are_sized_and_repeatably_iterable() -> None: """What ``FanOut`` needs of a plan, checked on both real plan types. diff --git a/tests/headers_host_scoping_test.py b/tests/headers_host_scoping_test.py index f5f31cd0..37f5c7c0 100644 --- a/tests/headers_host_scoping_test.py +++ b/tests/headers_host_scoping_test.py @@ -125,23 +125,26 @@ def test_generic_ogc_request_excludes_key_for_custom_host(self): def test_rating_download_scopes_headers_to_asset_url(self): """The ratings adapter evaluates auth against each asset href.""" + import asyncio + import pandas as pd import dataretrieval.waterdata.ratings as ratings asset_url = "https://objects.example.org/ratings/site.rdb" feature = {"id": "site.rdb", "assets": {"data": {"href": asset_url}}} - response = mock.Mock(text="rating body") + client = mock.AsyncMock(spec=httpx.AsyncClient) + client.get.return_value = mock.Mock(text="rating body") with ( - mock.patch.object(ratings, "_get", return_value=response) as get, + mock.patch.object(ratings, "active_client", return_value=client), mock.patch.object(ratings, "_raise_for_non_200"), mock.patch.object(ratings, "read_rdb", return_value=pd.DataFrame()), mock.patch.object(ratings, "extract_rdb_comment", return_value=""), ): - ratings._download_and_parse(feature, file_path=None, ssl_check=True) + asyncio.run(ratings._fetch_rating(feature, file_path=None)) - assert get.call_args.args[0] == asset_url - assert "X-Api-Key" not in get.call_args.kwargs["headers"] + assert client.get.call_args.args[0] == asset_url + assert "X-Api-Key" not in client.get.call_args.kwargs["headers"] def test_sync_redirect_strips_key_before_cross_host_request(self): """The synchronous transport guard runs again for redirects.""" diff --git a/tests/waterdata_ratings_test.py b/tests/waterdata_ratings_test.py index bd1d8e43..a3acd111 100644 --- a/tests/waterdata_ratings_test.py +++ b/tests/waterdata_ratings_test.py @@ -201,6 +201,40 @@ def test_get_ratings_multi_type_filters_via_property(httpx_mock, tmp_path): assert "file_type" not in qs["filter"][0] +def test_get_ratings_search_429_is_resumable(httpx_mock): + """A rate-limited search surfaces as a resumable interruption — parity + with the other getters, which drive the same executor — instead of a raw + ``RateLimited``; resuming finishes the interrupted stage.""" + from dataretrieval.interruptions import QuotaExhausted + + httpx_mock.add_response(method="GET", url=STAC_SEARCH_RE, status_code=429) + httpx_mock.add_response( + method="GET", url=STAC_SEARCH_RE, json=_stub_search_response() + ) + + with pytest.raises(QuotaExhausted) as excinfo: + get_ratings(monitoring_location_id="USGS-01104475", download_and_parse=False) + + df, _ = excinfo.value.call.resume() + assert list(df["feature"])[0]["id"] == "USGS-01104475.exsa.rdb" + + +def test_get_ratings_download_failure_surfaces_typed(httpx_mock): + """A failing download surfaces the module's typed error instead of being + logged and silently dropped from the result dict.""" + httpx_mock.add_response( + method="GET", url=STAC_SEARCH_RE, json=_stub_search_response() + ) + httpx_mock.add_response( + method="GET", + url="https://api.waterdata.usgs.gov/stac-files/ratings/USGS.01104475.exsa.rdb", + status_code=404, + ) + + with pytest.raises(DataRetrievalError): + get_ratings(monitoring_location_id="USGS-01104475") + + def test_stac_next_link_refuses_another_host(httpx_mock): """The STAC page walk must not follow a link off the ratings host. From 47c1751125c31889464bfbd3c50284a8c927d12a Mon Sep 17 00:00:00 2001 From: thodson-usgs Date: Wed, 12 Aug 2026 15:31:58 -0500 Subject: [PATCH 2/3] fix(ratings): skip deterministic per-feature failures with a visible warning Split the download stage's failure policy by whether retrying could help, instead of treating every failure as fatal to the batch: - A transient failure (429 / 5xx / timeout / connection drop) keeps the executor semantics: retried per API_USGS_RETRIES, then raised as a resumable interruption. Skipping these would silently drop every remaining feature under rate limiting -- the old serial loop's worst case, and the reason this branch retired log-and-skip. - A deterministic per-feature failure (stale catalog entry 404ing, a feature with no data asset, a malformed RDB) now skips that feature under a new SkippedRatingWarning naming it, and returns the rest of the batch. Aborting would have discarded every other site's rating over one bad catalog entry; retrying would reproduce the failure. - OSError writing file_path propagates: a local disk problem is not a per-feature condition (the old loop's skip was overbroad here). SkippedRatingWarning is a UserWarning (visible by default, unlike the old logger.warning, which was silent unless logging was configured) in dataretrieval.exceptions, re-exported at the package root. Strict all-or-nothing behavior is one filter away: warnings.filterwarnings("error", category=SkippedRatingWarning). A skipped feature hands the executor an inert 204 placeholder pair so the item is marked complete and a later resume() continues past it. A site with no published rating still warns nothing: it matches no feature in the search, so there is nothing to skip. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01BTaSm7HmVb94RSJiKW4WAS --- dataretrieval/__init__.py | 2 + dataretrieval/exceptions.py | 32 +++++++++- dataretrieval/waterdata/ratings.py | 69 +++++++++++++++++---- tests/waterdata_ratings_test.py | 97 +++++++++++++++++++++++++++--- 4 files changed, 178 insertions(+), 22 deletions(-) diff --git a/dataretrieval/__init__.py b/dataretrieval/__init__.py index 29e288d7..28a36f01 100644 --- a/dataretrieval/__init__.py +++ b/dataretrieval/__init__.py @@ -41,6 +41,7 @@ RateLimited, RequestTooLarge, ServiceUnavailable, + SkippedRatingWarning, TransientError, Unchunkable, URLTooLong, @@ -95,6 +96,7 @@ "RateLimited", "RequestTooLarge", "ServiceUnavailable", + "SkippedRatingWarning", "TransientError", "URLTooLong", "Unchunkable", diff --git a/dataretrieval/exceptions.py b/dataretrieval/exceptions.py index 79160e9e..d020e78b 100644 --- a/dataretrieval/exceptions.py +++ b/dataretrieval/exceptions.py @@ -13,7 +13,9 @@ aren't a plain status: :class:`RequestTooLarge` (with :class:`URLTooLong` / :class:`Unchunkable`), :class:`NetworkError` (a failed connection, per above), :class:`NoSitesError`, and :class:`ConfigurationError` for an unusable setting. -:func:`error_for_status` maps a status to its type. +:func:`error_for_status` maps a status to its type. The one *warning* +category lives here too: :class:`SkippedRatingWarning`, a per-feature skip +inside a rating batch. This module has no third-party runtime dependencies -- ``httpx`` is imported only for type checking. Any module can therefore import it without pulling in pandas @@ -42,6 +44,7 @@ "NetworkError", "NoSitesError", "ConfigurationError", + "SkippedRatingWarning", "error_for_status", "parse_retry_after", ] @@ -289,6 +292,33 @@ def __str__(self) -> str: ) +# --- Skipped work --------------------------------------------------------- + + +class SkippedRatingWarning(UserWarning): + """One feature of a rating batch was skipped; the rest were returned. + + Emitted by :func:`dataretrieval.waterdata.get_ratings` when a single + feature fails *deterministically* -- a stale catalog entry (404 on its + data asset), a feature carrying no data asset, a malformed RDB file. The + failed feature's id is absent from the returned dict. Skipping is the + right default for this class of failure: retrying would reproduce it, and + aborting would discard every other site's rating over one bad catalog + entry. + + Transient failures (429 / 5xx / timeouts / connection drops) are never + skipped -- they are retried and, if retries run out, raised as a resumable + interruption. Rate limiting in particular is systematic, so skipping + there would silently drop most of a batch; that silent loss is the + failure mode this policy exists to prevent. + + A warning rather than a log line so it is visible by default. To make a + skip fatal (strict all-or-nothing behavior):: + + warnings.filterwarnings("error", category=SkippedRatingWarning) + """ + + def error_for_status( status: int, message: str, *, retry_after: float | None = None ) -> DataRetrievalError: diff --git a/dataretrieval/waterdata/ratings.py b/dataretrieval/waterdata/ratings.py index 35e947ba..8b0a5710 100644 --- a/dataretrieval/waterdata/ratings.py +++ b/dataretrieval/waterdata/ratings.py @@ -10,12 +10,15 @@ from __future__ import annotations import os +import warnings from collections.abc import Iterable +from datetime import timedelta from typing import Any, Literal, get_args import httpx import pandas as pd +from dataretrieval.exceptions import DataRetrievalError, SkippedRatingWarning from dataretrieval.ogc.dates import _DURATION_RE, _format_api_dates from dataretrieval.ogc.errors import _raise_for_non_200 from dataretrieval.ogc.filters import _quote_cql_str @@ -115,10 +118,10 @@ def get_ratings( Raises ------ ValueError - For an unrecognized ``file_type`` value, an ISO 8601 duration in - ``time``, or a matching feature that carries no data asset. + For an unrecognized ``file_type`` value or an ISO 8601 duration in + ``time``. DataRetrievalError - The typed subclass for an HTTP error response (see + The typed subclass for an HTTP error response during the search (see :func:`transport.pagination.paginate`); or :class:`~dataretrieval.exceptions.NetworkError` if a request can't reach the service in a way retrying cannot fix. @@ -129,6 +132,19 @@ def get_ratings( assembled per-feature dict is returned by a fresh ``get_ratings`` call. + Warns + ----- + SkippedRatingWarning + One feature of the batch failed *deterministically* -- a stale + catalog entry (404 on its data asset), a feature with no data asset, + a malformed RDB file. That feature is skipped and its id is absent + from the returned dict; the rest of the batch is unaffected. A site + with no published rating never warns -- it matches no feature in the + search, so there is nothing to skip. Transient failures are never + skipped (they retry, then raise resumable). To make a skip fatal:: + + warnings.filterwarnings("error", category=SkippedRatingWarning) + Examples -------- .. code:: @@ -338,23 +354,54 @@ def _download_all( The plan is the feature list itself -- ``FanOut`` asks a plan only to be sized and iterable -- so the downloads get bounded concurrency, per-attempt retry, the progress line, and the resumable interruption - taxonomy in place of the previous serial loop, which had none of them and - logged-and-skipped every failure. A deterministic per-feature failure (a - missing asset, a malformed RDB) now surfaces typed instead of silently - dropping that rating from the result; a transient one is retried and, if - retries are exhausted, raised as a resumable interruption. + taxonomy in place of the previous serial loop, which had none of them. + + Failure policy -- split by whether retrying could help. A *transient* + failure (429 / 5xx / timeout / connection drop) is retried and, if + retries run out, raised as a resumable interruption: rate limiting is + systematic, so skipping it would silently drop every remaining feature, + which was the old serial loop's worst case. A *deterministic* per-feature + failure (a stale catalog entry 404ing, a feature with no data asset, a + malformed RDB) is that feature's problem, not the batch's: it is skipped + under a :class:`~dataretrieval.exceptions.SkippedRatingWarning` naming + the feature, visible by default and escalatable to an error for strict + all-or-nothing behavior. ``OSError`` writing ``file_path`` propagates -- + a local disk problem is not a per-feature condition. Raw ``httpx`` + errors pass through untouched so the executor can classify and retry + them. The public result is a dict keyed by feature id, so the fetch closure accumulates it; the executor's combined frame is not the return shape and - is discarded. + is discarded. A skipped feature returns an inert placeholder pair so the + executor marks it complete -- a later ``resume()`` continues past it + rather than re-attempting the skip. """ out: dict[str, pd.DataFrame] = {} if not features: return out async def fetch(feature: dict[str, Any]) -> tuple[pd.DataFrame, httpx.Response]: - df, response = await _fetch_rating(feature, file_path) - out[feature["id"]] = df + fid = feature.get("id", "") + try: + df, response = await _fetch_rating(feature, file_path) + except (DataRetrievalError, LookupError, ValueError) as e: + if isinstance(e, DataRetrievalError) and e.retryable: + raise # transient: the executor retries, then raises resumable + warnings.warn( + f"Skipping rating {fid!r}: {e}", + SkippedRatingWarning, + stacklevel=2, + ) + href = feature.get("assets", {}).get("data", {}).get("href") + # The executor aggregates each completed item's response + # (headers / elapsed / url), so the skip must hand back a real, + # inert response rather than None. 204: completed, no content. + placeholder = httpx.Response( + 204, request=httpx.Request("GET", href or f"{STAC_URL}/search") + ) + placeholder.elapsed = timedelta(0) + return pd.DataFrame(), placeholder + out[fid] = df return df, response FanOut( diff --git a/tests/waterdata_ratings_test.py b/tests/waterdata_ratings_test.py index a3acd111..1c373ed7 100644 --- a/tests/waterdata_ratings_test.py +++ b/tests/waterdata_ratings_test.py @@ -1,10 +1,11 @@ import re +import warnings from urllib.parse import parse_qs, urlsplit import pandas as pd import pytest -from dataretrieval.exceptions import DataRetrievalError +from dataretrieval.exceptions import DataRetrievalError, SkippedRatingWarning from dataretrieval.waterdata import get_ratings from dataretrieval.waterdata.ratings import _build_filter @@ -219,20 +220,96 @@ def test_get_ratings_search_429_is_resumable(httpx_mock): assert list(df["feature"])[0]["id"] == "USGS-01104475.exsa.rdb" -def test_get_ratings_download_failure_surfaces_typed(httpx_mock): - """A failing download surfaces the module's typed error instead of being - logged and silently dropped from the result dict.""" +_GOOD_ASSET = "https://api.waterdata.usgs.gov/stac-files/ratings/USGS.01104475.exsa.rdb" +_BAD_ASSET = "https://api.waterdata.usgs.gov/stac-files/ratings/USGS.99999999.exsa.rdb" + + +def _two_feature_search_response(): + """One feature whose asset will fail, one that will succeed.""" + return { + "features": [ + { + "id": "USGS-99999999.exsa.rdb", + "properties": {"file_type": "exsa"}, + "assets": {"data": {"href": _BAD_ASSET}}, + }, + { + "id": "USGS-01104475.exsa.rdb", + "properties": {"file_type": "exsa"}, + "assets": {"data": {"href": _GOOD_ASSET}}, + }, + ] + } + + +def test_get_ratings_deterministic_download_failure_warns_and_skips(httpx_mock): + """A stale catalog entry (404 on its asset) costs only that feature: the + skip is announced with ``SkippedRatingWarning`` naming the feature, and + every other rating in the batch is still returned.""" httpx_mock.add_response( - method="GET", url=STAC_SEARCH_RE, json=_stub_search_response() + method="GET", url=STAC_SEARCH_RE, json=_two_feature_search_response() ) + httpx_mock.add_response(method="GET", url=_BAD_ASSET, status_code=404) + httpx_mock.add_response(method="GET", url=_GOOD_ASSET, text=_SAMPLE_RDB) + + with pytest.warns(SkippedRatingWarning, match="USGS-99999999"): + out = get_ratings(monitoring_location_id=["USGS-99999999", "USGS-01104475"]) + + assert sorted(out) == ["USGS-01104475.exsa.rdb"] + assert len(out["USGS-01104475.exsa.rdb"]) == 3 + + +def test_get_ratings_feature_without_asset_warns_and_skips(httpx_mock): + """A catalog feature carrying no data asset is a per-feature data problem: + skipped with a warning, without costing the rest of the batch.""" + body = _two_feature_search_response() + body["features"][0]["assets"] = {} + + httpx_mock.add_response(method="GET", url=STAC_SEARCH_RE, json=body) + httpx_mock.add_response(method="GET", url=_GOOD_ASSET, text=_SAMPLE_RDB) + + with pytest.warns(SkippedRatingWarning, match="no data asset"): + out = get_ratings(monitoring_location_id=["USGS-99999999", "USGS-01104475"]) + + assert sorted(out) == ["USGS-01104475.exsa.rdb"] + + +def test_get_ratings_skip_warning_escalates_to_error(httpx_mock): + """``filterwarnings("error", ...)`` restores strict all-or-nothing: the + escalated skip surfaces as an exception instead of a silent gap.""" httpx_mock.add_response( - method="GET", - url="https://api.waterdata.usgs.gov/stac-files/ratings/USGS.01104475.exsa.rdb", - status_code=404, + method="GET", url=STAC_SEARCH_RE, json=_two_feature_search_response() + ) + httpx_mock.add_response(method="GET", url=_BAD_ASSET, status_code=404) + httpx_mock.add_response( + method="GET", url=_GOOD_ASSET, text=_SAMPLE_RDB, is_optional=True + ) + + with warnings.catch_warnings(): + warnings.filterwarnings("error", category=SkippedRatingWarning) + with pytest.raises(SkippedRatingWarning): + get_ratings(monitoring_location_id=["USGS-99999999", "USGS-01104475"]) + + +def test_get_ratings_download_429_is_resumable_not_skipped(httpx_mock): + """A rate-limited download must never be skipped -- it is raised as a + resumable interruption, and resuming completes the batch. The escalation + filter proves no ``SkippedRatingWarning`` fires along the way.""" + from dataretrieval.interruptions import QuotaExhausted + + httpx_mock.add_response( + method="GET", url=STAC_SEARCH_RE, json=_stub_search_response() ) + httpx_mock.add_response(method="GET", url=_GOOD_ASSET, status_code=429) + httpx_mock.add_response(method="GET", url=_GOOD_ASSET, text=_SAMPLE_RDB) - with pytest.raises(DataRetrievalError): - get_ratings(monitoring_location_id="USGS-01104475") + with warnings.catch_warnings(): + warnings.filterwarnings("error", category=SkippedRatingWarning) + with pytest.raises(QuotaExhausted) as excinfo: + get_ratings(monitoring_location_id="USGS-01104475") + + df, _ = excinfo.value.call.resume() + assert len(df) == 3 def test_stac_next_link_refuses_another_host(httpx_mock): From 585ba6c470637e737579f0d02081cfa532bd38a0 Mon Sep 17 00:00:00 2001 From: thodson-usgs Date: Wed, 12 Aug 2026 15:44:18 -0500 Subject: [PATCH 3/3] refactor(ratings): apply /simplify findings Four parallel review agents (reuse, simplification, efficiency, altitude) over the branch diff; applied the converging findings: - Generalize the warning taxonomy: SkippedItemWarning carries the batch-skip policy once; SkippedRatingWarning subclasses it, so users can escalate every future skip with one filter while still targeting ratings alone. Both exported at the package root. The get_ratings and _download_all docstrings now cross-reference the policy instead of restating it in full (three copies -> one). - Delete the dead borrow-or-open client branch in _fetch_rating: FanOut publishes the shared client before any fetch runs, so active_client() is never None inside a drive; the fallback executed nowhere. The open_async_client import goes with it, and the architecture test's allowlist tightens to default_headers alone. - Drop the redundant placeholder.elapsed = timedelta(0): the aggregate path reads elapsed only through combining._safe_elapsed, which already zero-fills hand-built responses. The timedelta import goes with it. - Return body-less responses to the executor (_inert_response) for successes as well as skips: the executor keeps every completed item's response until the drive ends but aggregates only status, headers, and URL, so retaining full RDB bodies pinned ~N x tens-of-KB for the whole retry-prone drive. Real status and quota headers are kept. - Deduplicate the thrice-spelled assets.data.href traversal into _asset_href. - Drop the derivable empty-frame guard in _search: every page frame is built with a feature column and the combine helpers preserve it. - Tests: hoist the QuotaExhausted import and the _GOOD_ASSET/_BAD_ASSET constants (the 429-resume test silently depended on two spellings of the same URL agreeing); extract _imports_from in architecture_test so the three import-surface rules share one mechanism, standardized on ast.walk (function-local imports can no longer slip past the two older rules). Skipped as out of scope, for a follow-up issue: a shared one-request paginate-through-FanOut driver (second copy of the stats.py scaffold), and a FanOut per-item result mode (native skip outcome; resume() returning the public dict shape instead of requiring a fresh call, which currently re-downloads the whole batch after an interruption). Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01BTaSm7HmVb94RSJiKW4WAS --- dataretrieval/__init__.py | 2 + dataretrieval/exceptions.py | 52 ++++++++------ dataretrieval/waterdata/ratings.py | 108 ++++++++++++++++------------- tests/architecture_test.py | 48 ++++++------- tests/waterdata_ratings_test.py | 32 +++------ 5 files changed, 123 insertions(+), 119 deletions(-) diff --git a/dataretrieval/__init__.py b/dataretrieval/__init__.py index 28a36f01..8ecfb7bf 100644 --- a/dataretrieval/__init__.py +++ b/dataretrieval/__init__.py @@ -41,6 +41,7 @@ RateLimited, RequestTooLarge, ServiceUnavailable, + SkippedItemWarning, SkippedRatingWarning, TransientError, Unchunkable, @@ -96,6 +97,7 @@ "RateLimited", "RequestTooLarge", "ServiceUnavailable", + "SkippedItemWarning", "SkippedRatingWarning", "TransientError", "URLTooLong", diff --git a/dataretrieval/exceptions.py b/dataretrieval/exceptions.py index d020e78b..a6033893 100644 --- a/dataretrieval/exceptions.py +++ b/dataretrieval/exceptions.py @@ -13,9 +13,9 @@ aren't a plain status: :class:`RequestTooLarge` (with :class:`URLTooLong` / :class:`Unchunkable`), :class:`NetworkError` (a failed connection, per above), :class:`NoSitesError`, and :class:`ConfigurationError` for an unusable setting. -:func:`error_for_status` maps a status to its type. The one *warning* -category lives here too: :class:`SkippedRatingWarning`, a per-feature skip -inside a rating batch. +:func:`error_for_status` maps a status to its type. The *warning* side of the +taxonomy lives here too: :class:`SkippedItemWarning` (specialized by +:class:`SkippedRatingWarning`), a per-item skip inside a batched retrieval. This module has no third-party runtime dependencies -- ``httpx`` is imported only for type checking. Any module can therefore import it without pulling in pandas @@ -44,6 +44,7 @@ "NetworkError", "NoSitesError", "ConfigurationError", + "SkippedItemWarning", "SkippedRatingWarning", "error_for_status", "parse_retry_after", @@ -295,27 +296,38 @@ def __str__(self) -> str: # --- Skipped work --------------------------------------------------------- -class SkippedRatingWarning(UserWarning): - """One feature of a rating batch was skipped; the rest were returned. - - Emitted by :func:`dataretrieval.waterdata.get_ratings` when a single - feature fails *deterministically* -- a stale catalog entry (404 on its - data asset), a feature carrying no data asset, a malformed RDB file. The - failed feature's id is absent from the returned dict. Skipping is the - right default for this class of failure: retrying would reproduce it, and - aborting would discard every other site's rating over one bad catalog - entry. +class SkippedItemWarning(UserWarning): + """One item of a batched retrieval was skipped; the rest were returned. + The policy for batch getters whose items are independent documents: an + item that fails *deterministically* -- so retrying would reproduce the + failure -- is dropped from the result under a warning naming it, because + aborting would discard every other item's data over one bad entry. Transient failures (429 / 5xx / timeouts / connection drops) are never - skipped -- they are retried and, if retries run out, raised as a resumable - interruption. Rate limiting in particular is systematic, so skipping - there would silently drop most of a batch; that silent loss is the - failure mode this policy exists to prevent. + skipped -- they are retried and, if retries run out, raised as a + resumable interruption. Rate limiting in particular is systematic, so + skipping there would silently drop most of a batch; that silent loss is + the failure mode this policy exists to prevent. + + A warning rather than a log line so it is visible by default. To make + any skip fatal (strict all-or-nothing behavior):: + + warnings.filterwarnings("error", category=SkippedItemWarning) + + Getters emit a subclass naming their surface (e.g. + :class:`SkippedRatingWarning`), so a filter can also target one getter. + """ + - A warning rather than a log line so it is visible by default. To make a - skip fatal (strict all-or-nothing behavior):: +class SkippedRatingWarning(SkippedItemWarning): + """A rating feature was skipped by + :func:`dataretrieval.waterdata.get_ratings`. - warnings.filterwarnings("error", category=SkippedRatingWarning) + Emitted when a single STAC feature fails deterministically -- a stale + catalog entry (404 on its data asset), a feature carrying no data asset, + a malformed RDB file. The failed feature's id is absent from the returned + dict. See :class:`SkippedItemWarning` for the policy and how to escalate + a skip to an error. """ diff --git a/dataretrieval/waterdata/ratings.py b/dataretrieval/waterdata/ratings.py index 8b0a5710..5aa979b9 100644 --- a/dataretrieval/waterdata/ratings.py +++ b/dataretrieval/waterdata/ratings.py @@ -12,7 +12,6 @@ import os import warnings from collections.abc import Iterable -from datetime import timedelta from typing import Any, Literal, get_args import httpx @@ -28,9 +27,6 @@ from dataretrieval.transport.http import ( default_headers as _default_headers, ) -from dataretrieval.transport.http import ( - open_async_client as _open_async_client, -) from dataretrieval.transport.links import resolve_next_url from dataretrieval.transport.pagination import paginate from dataretrieval.transport.retry import RetryPolicy @@ -140,10 +136,10 @@ def get_ratings( a malformed RDB file. That feature is skipped and its id is absent from the returned dict; the rest of the batch is unaffected. A site with no published rating never warns -- it matches no feature in the - search, so there is nothing to skip. Transient failures are never - skipped (they retry, then raise resumable). To make a skip fatal:: - - warnings.filterwarnings("error", category=SkippedRatingWarning) + search, so there is nothing to skip. See + :class:`~dataretrieval.exceptions.SkippedItemWarning` for the policy + (transients never skip) and the ``filterwarnings`` recipe that makes + a skip fatal. Examples -------- @@ -250,12 +246,10 @@ def _search( The page walk is :func:`dataretrieval.transport.pagination.paginate` with STAC strategies, driven as a one-item - :class:`~dataretrieval.transport.fanout.FanOut` -- the same executor every - other getter uses -- so the search gets per-attempt retry, the stall - budget, a progress line, and the resumable interruption taxonomy instead - of a bespoke sync loop that had none of them. Pages carry features rather - than rows, so each page frame wraps the raw feature dicts in a single - ``feature`` column. + :class:`~dataretrieval.transport.fanout.FanOut` -- the same executor and + semantics (retry, stall budget, progress line, resumable interruption) as + every other getter. Pages carry features rather than rows, so each page + frame wraps the raw feature dicts in a single ``feature`` column. """ query_params: dict[str, Any] = {"limit": min(limit, 10000)} if filter_str is not None: @@ -308,7 +302,29 @@ async def fetch(request: httpx.Request) -> tuple[pd.DataFrame, httpx.Response]: canonical_url=str(req.url), service="ratings", ).resume() - return [] if df.empty else list(df["feature"]) + # Every page frame is built with a ``feature`` column, and the combine + # helpers preserve it, so the empty case needs no special branch. + return list(df["feature"]) + + +def _asset_href(feature: dict[str, Any]) -> str | None: + """The feature's data-asset href, or ``None`` when the catalog omits it.""" + href: str | None = feature.get("assets", {}).get("data", {}).get("href") + return href + + +def _inert_response( + status: int, url: str, headers: httpx.Headers | None = None +) -> httpx.Response: + """A body-less stand-in the executor can aggregate. + + The executor keeps every completed item's response until the drive ends, + but its aggregation reads only status, headers, and URL -- never the + body. Handing it a stand-in keeps a large batch from pinning every + downloaded file in memory for the whole drive. ``elapsed`` is left + unset; the aggregate's ``_safe_elapsed`` treats that as zero. + """ + return httpx.Response(status, headers=headers, request=httpx.Request("GET", url)) async def _fetch_rating( @@ -317,21 +333,19 @@ async def _fetch_rating( """Fetch one feature's data asset, parse RDB, optionally persist to disk. Headers are evaluated against each asset href -- assets can live on a - different host than the catalog, and must not inherit its auth. Borrows - the executor's shared client; outside a drive (a direct unit call) it - opens a short-lived one. + different host than the catalog, and must not inherit its auth. Runs + inside a drive: the executor publishes the shared client before any + fetch starts. """ fid = feature["id"] - href = feature.get("assets", {}).get("data", {}).get("href") + href = _asset_href(feature) if not href: raise ValueError(f"STAC feature {fid!r} carries no data asset href.") headers = _default_headers(href) session = active_client() - if session is not None: - response = await session.get(href, headers=headers) - else: - async with _open_async_client() as own: - response = await own.get(href, headers=headers) + if session is None: + raise RuntimeError("_fetch_rating must run inside a FanOut drive.") + response = await session.get(href, headers=headers) _raise_for_non_200(response) if file_path is not None: @@ -356,25 +370,22 @@ def _download_all( per-attempt retry, the progress line, and the resumable interruption taxonomy in place of the previous serial loop, which had none of them. - Failure policy -- split by whether retrying could help. A *transient* - failure (429 / 5xx / timeout / connection drop) is retried and, if - retries run out, raised as a resumable interruption: rate limiting is - systematic, so skipping it would silently drop every remaining feature, - which was the old serial loop's worst case. A *deterministic* per-feature - failure (a stale catalog entry 404ing, a feature with no data asset, a - malformed RDB) is that feature's problem, not the batch's: it is skipped - under a :class:`~dataretrieval.exceptions.SkippedRatingWarning` naming - the feature, visible by default and escalatable to an error for strict - all-or-nothing behavior. ``OSError`` writing ``file_path`` propagates -- - a local disk problem is not a per-feature condition. Raw ``httpx`` - errors pass through untouched so the executor can classify and retry - them. + Failure policy (rationale on + :class:`~dataretrieval.exceptions.SkippedItemWarning`): a *transient* + failure (429 / 5xx / timeout / connection drop) re-raises so the executor + retries and then raises resumable; a *deterministic* per-feature failure + warns with :class:`~dataretrieval.exceptions.SkippedRatingWarning` and + skips the feature. ``OSError`` writing ``file_path`` propagates -- a + local disk problem is not a per-feature condition. Raw ``httpx`` errors + pass through untouched so the executor can classify and retry them. The public result is a dict keyed by feature id, so the fetch closure accumulates it; the executor's combined frame is not the return shape and - is discarded. A skipped feature returns an inert placeholder pair so the - executor marks it complete -- a later ``resume()`` continues past it - rather than re-attempting the skip. + is discarded. Both outcomes hand the executor a body-less + :func:`_inert_response` -- a skip so the item counts as complete (a later + ``resume()`` continues past it rather than re-attempting), a success so + the drive doesn't pin every downloaded file in memory while keeping the + real status and quota headers for aggregation. """ out: dict[str, pd.DataFrame] = {} if not features: @@ -392,17 +403,14 @@ async def fetch(feature: dict[str, Any]) -> tuple[pd.DataFrame, httpx.Response]: SkippedRatingWarning, stacklevel=2, ) - href = feature.get("assets", {}).get("data", {}).get("href") - # The executor aggregates each completed item's response - # (headers / elapsed / url), so the skip must hand back a real, - # inert response rather than None. 204: completed, no content. - placeholder = httpx.Response( - 204, request=httpx.Request("GET", href or f"{STAC_URL}/search") + # 204: completed, no content. + return pd.DataFrame(), _inert_response( + 204, _asset_href(feature) or f"{STAC_URL}/search" ) - placeholder.elapsed = timedelta(0) - return pd.DataFrame(), placeholder out[fid] = df - return df, response + return df, _inert_response( + response.status_code, str(response.url), response.headers + ) FanOut( features, @@ -411,7 +419,7 @@ async def fetch(feature: dict[str, Any]) -> tuple[pd.DataFrame, httpx.Response]: client_options={"verify": ssl_check}, # No single URL expresses "all of these assets" -- the aggregate # reports the first, matching what a single-feature call would show. - canonical_url=features[0].get("assets", {}).get("data", {}).get("href"), + canonical_url=_asset_href(features[0]), service="ratings", ).resume() return out diff --git a/tests/architecture_test.py b/tests/architecture_test.py index 727ccc82..88606b76 100644 --- a/tests/architecture_test.py +++ b/tests/architecture_test.py @@ -143,6 +143,21 @@ def _literal_exports(path: Path) -> set[str]: raise AssertionError(f"{path.relative_to(PACKAGE_ROOT.parent)} has no __all__") +def _imports_from(path: Path, module: str) -> set[str]: + """Names *path* imports from *module*, wherever the import appears. + + Walks the whole tree rather than only ``tree.body`` so a function-local + import can't slip past an import-surface rule. + """ + tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) + return { + alias.name + for node in ast.walk(tree) + if isinstance(node, ast.ImportFrom) and node.module == module + for alias in node.names + } + + def test_exceptions_has_no_runtime_third_party_dependency() -> None: """The shared error-policy leaf must remain cheap and cycle-safe.""" imports = _runtime_imports(PACKAGE_ROOT / "exceptions.py") @@ -170,19 +185,13 @@ def test_engine_request_import_surface_does_not_grow() -> None: means request construction is migrating back into engine. """ path = PACKAGE_ROOT / "ogc" / "engine.py" - tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) - imported = { - alias.name - for node in tree.body - if isinstance(node, ast.ImportFrom) - and node.module == "dataretrieval.ogc.requests" - for alias in node.names - } + imported = _imports_from(path, "dataretrieval.ogc.requests") assert len(imported) <= _MAX_ENGINE_REQUEST_IMPORTS, ( "ogc.engine imports more request names than before; use the canonical " "requests module instead of expanding engine's request surface.\n" f"limit={_MAX_ENGINE_REQUEST_IMPORTS}\nobserved={sorted(imported)}" ) + tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) referenced = { node.id for node in ast.walk(tree) @@ -485,14 +494,7 @@ def test_ogc_request_construction_does_not_execute_http() -> None: be either vacuous or a list of exceptions. """ path = PACKAGE_ROOT / "ogc" / "requests.py" - tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) - transport_names = { - alias.name - for node in tree.body - if isinstance(node, ast.ImportFrom) - and node.module == "dataretrieval.transport.http" - for alias in node.names - } + transport_names = _imports_from(path, "dataretrieval.transport.http") assert transport_names == {"default_headers"}, ( f"ogc.requests imports executing transport helpers: {sorted(transport_names)}" ) @@ -547,16 +549,10 @@ def test_ratings_drives_http_through_the_shared_executor() -> None: ``paginate``/``FanOut`` like every other multi-request path; assert the direct sync entry points cannot quietly return. """ - path = PACKAGE_ROOT / "waterdata" / "ratings.py" - tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) - transport_names = { - alias.name - for node in ast.walk(tree) - if isinstance(node, ast.ImportFrom) - and node.module == "dataretrieval.transport.http" - for alias in node.names - } - offenders = transport_names - {"default_headers", "open_async_client"} + transport_names = _imports_from( + PACKAGE_ROOT / "waterdata" / "ratings.py", "dataretrieval.transport.http" + ) + offenders = transport_names - {"default_headers"} assert not offenders, ( "ratings imports executing sync transport helpers instead of driving " f"the shared executor: {sorted(offenders)}" diff --git a/tests/waterdata_ratings_test.py b/tests/waterdata_ratings_test.py index 1c373ed7..5341ed99 100644 --- a/tests/waterdata_ratings_test.py +++ b/tests/waterdata_ratings_test.py @@ -6,6 +6,7 @@ import pytest from dataretrieval.exceptions import DataRetrievalError, SkippedRatingWarning +from dataretrieval.interruptions import QuotaExhausted from dataretrieval.waterdata import get_ratings from dataretrieval.waterdata.ratings import _build_filter @@ -69,17 +70,17 @@ def test_build_filter_escapes_quotes(): """ +_GOOD_ASSET = "https://api.waterdata.usgs.gov/stac-files/ratings/USGS.01104475.exsa.rdb" +_BAD_ASSET = "https://api.waterdata.usgs.gov/stac-files/ratings/USGS.99999999.exsa.rdb" + + def _stub_search_response(): return { "features": [ { "id": "USGS-01104475.exsa.rdb", "properties": {"file_type": "exsa"}, - "assets": { - "data": { - "href": "https://api.waterdata.usgs.gov/stac-files/ratings/USGS.01104475.exsa.rdb" - } - }, + "assets": {"data": {"href": _GOOD_ASSET}}, } ] } @@ -92,11 +93,7 @@ def test_get_ratings_mocked_search_and_download(httpx_mock, tmp_path): url=STAC_SEARCH_RE, json=_stub_search_response(), ) - httpx_mock.add_response( - method="GET", - url="https://api.waterdata.usgs.gov/stac-files/ratings/USGS.01104475.exsa.rdb", - text=_SAMPLE_RDB, - ) + httpx_mock.add_response(method="GET", url=_GOOD_ASSET, text=_SAMPLE_RDB) out = get_ratings( monitoring_location_id="USGS-01104475", @@ -123,10 +120,7 @@ def test_get_ratings_attaches_rdb_comment_and_url(httpx_mock, tmp_path): url=STAC_SEARCH_RE, json=_stub_search_response(), ) - asset_url = ( - "https://api.waterdata.usgs.gov/stac-files/ratings/USGS.01104475.exsa.rdb" - ) - httpx_mock.add_response(method="GET", url=asset_url, text=_SAMPLE_RDB) + httpx_mock.add_response(method="GET", url=_GOOD_ASSET, text=_SAMPLE_RDB) out = get_ratings( monitoring_location_id="USGS-01104475", @@ -139,7 +133,7 @@ def test_get_ratings_attaches_rdb_comment_and_url(httpx_mock, tmp_path): "# header line one", "# header line two", ] - assert df.attrs["url"] == asset_url + assert df.attrs["url"] == _GOOD_ASSET def test_get_ratings_download_and_parse_false_returns_features(httpx_mock): @@ -206,8 +200,6 @@ def test_get_ratings_search_429_is_resumable(httpx_mock): """A rate-limited search surfaces as a resumable interruption — parity with the other getters, which drive the same executor — instead of a raw ``RateLimited``; resuming finishes the interrupted stage.""" - from dataretrieval.interruptions import QuotaExhausted - httpx_mock.add_response(method="GET", url=STAC_SEARCH_RE, status_code=429) httpx_mock.add_response( method="GET", url=STAC_SEARCH_RE, json=_stub_search_response() @@ -220,10 +212,6 @@ def test_get_ratings_search_429_is_resumable(httpx_mock): assert list(df["feature"])[0]["id"] == "USGS-01104475.exsa.rdb" -_GOOD_ASSET = "https://api.waterdata.usgs.gov/stac-files/ratings/USGS.01104475.exsa.rdb" -_BAD_ASSET = "https://api.waterdata.usgs.gov/stac-files/ratings/USGS.99999999.exsa.rdb" - - def _two_feature_search_response(): """One feature whose asset will fail, one that will succeed.""" return { @@ -295,8 +283,6 @@ def test_get_ratings_download_429_is_resumable_not_skipped(httpx_mock): """A rate-limited download must never be skipped -- it is raised as a resumable interruption, and resuming completes the batch. The escalation filter proves no ``SkippedRatingWarning`` fires along the way.""" - from dataretrieval.interruptions import QuotaExhausted - httpx_mock.add_response( method="GET", url=STAC_SEARCH_RE, json=_stub_search_response() )