diff --git a/dataretrieval/__init__.py b/dataretrieval/__init__.py index 29e288d7..8ecfb7bf 100644 --- a/dataretrieval/__init__.py +++ b/dataretrieval/__init__.py @@ -41,6 +41,8 @@ RateLimited, RequestTooLarge, ServiceUnavailable, + SkippedItemWarning, + SkippedRatingWarning, TransientError, Unchunkable, URLTooLong, @@ -95,6 +97,8 @@ "RateLimited", "RequestTooLarge", "ServiceUnavailable", + "SkippedItemWarning", + "SkippedRatingWarning", "TransientError", "URLTooLong", "Unchunkable", diff --git a/dataretrieval/exceptions.py b/dataretrieval/exceptions.py index 79160e9e..a6033893 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 *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 @@ -42,6 +44,8 @@ "NetworkError", "NoSitesError", "ConfigurationError", + "SkippedItemWarning", + "SkippedRatingWarning", "error_for_status", "parse_retry_after", ] @@ -289,6 +293,44 @@ def __str__(self) -> str: ) +# --- Skipped work --------------------------------------------------------- + + +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. + + 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. + """ + + +class SkippedRatingWarning(SkippedItemWarning): + """A rating feature was skipped by + :func:`dataretrieval.waterdata.get_ratings`. + + 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. + """ + + 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 ff852029..5aa979b9 100644 --- a/dataretrieval/waterdata/ratings.py +++ b/dataretrieval/waterdata/ratings.py @@ -9,37 +9,32 @@ from __future__ import annotations -import logging import os +import warnings from collections.abc import Iterable from typing import Any, Literal, get_args import httpx import pandas as pd -from dataretrieval.exceptions import DataRetrievalError +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 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, -) 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) @@ -121,6 +116,30 @@ def get_ratings( ValueError For an unrecognized ``file_type`` value or an ISO 8601 duration in ``time``. + DataRetrievalError + 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. + 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. + + 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. See + :class:`~dataretrieval.exceptions.SkippedItemWarning` for the policy + (transients never skip) and the ``filterwarnings`` recipe that makes + a skip fatal. Examples -------- @@ -186,25 +205,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 +243,14 @@ 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 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: query_params["filter"] = filter_str @@ -252,60 +259,167 @@ 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 + df, _ = FanOut( + [req], + fetch, + RetryPolicy.from_env(), + client_options={"verify": ssl_check}, + canonical_url=str(req.url), + service="ratings", + ).resume() + # 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)) -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 - ) + +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. Runs + inside a drive: the executor publishes the shared client before any + fetch starts. + """ + fid = feature["id"] + 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 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: - 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. + + 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. 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: + return out + + async def fetch(feature: dict[str, Any]) -> tuple[pd.DataFrame, httpx.Response]: + 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, + ) + # 204: completed, no content. + return pd.DataFrame(), _inert_response( + 204, _asset_href(feature) or f"{STAC_URL}/search" + ) + out[fid] = df + return df, _inert_response( + response.status_code, str(response.url), response.headers + ) + + 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=_asset_href(features[0]), + service="ratings", + ).resume() + return out diff --git a/tests/architecture_test.py b/tests/architecture_test.py index 0fff59ef..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)}" ) @@ -538,6 +540,25 @@ 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. + """ + 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)}" + ) + + 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..5341ed99 100644 --- a/tests/waterdata_ratings_test.py +++ b/tests/waterdata_ratings_test.py @@ -1,10 +1,12 @@ 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.interruptions import QuotaExhausted from dataretrieval.waterdata import get_ratings from dataretrieval.waterdata.ratings import _build_filter @@ -68,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}}, } ] } @@ -91,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", @@ -122,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", @@ -138,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): @@ -201,6 +196,108 @@ 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.""" + 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 _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=_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=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.""" + 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 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): """The STAC page walk must not follow a link off the ratings host.