feat(transport): bounded retry for active services, over an API-neutral layer

.github/workflows/python-package.yml

@@ -53,6 +53,7 @@ jobs:
           from pathlib import Path
 
           import dataretrieval
+          import dataretrieval.transport
           from dataretrieval import ngwmn, waterdata, wateruse
           from dataretrieval.ogc import engine
 
@@ -105,6 +106,12 @@ jobs:
           python -m pip install --upgrade pip
           pip install .[test,nldi]
       - name: Test with pytest and report coverage
+        # Pinned to bash on every OS. The default Windows shell is PowerShell,
+        # which does not stop on a failing native command and takes the step's
+        # exit code from the last one -- so a pytest failure was masked by the
+        # coverage report that followed it, and the Windows matrix reported
+        # success while tests were red.
+        shell: bash
         run: |
           coverage run -m pytest tests/
           coverage report -m

NEWS.md

@@ -1,3 +1,5 @@
+**08/02/2026:** Added an internal API-neutral transport layer for guarded HTTP clients, host-scoped authentication, cursor pagination, bounded retry, response aggregation, progress, and sync-over-async dispatch. Water Use and the non-OGC Statistics API now consume transport directly instead of private OGC execution helpers; WQP, NLDI, and StreamStats opt into bounded transient retry while deprecated NWIS behavior remains unchanged. OGC retains CQL2, request construction, feature shaping, chunk planning, resumable calls, and interruption types, with compatibility imports at previous private paths. Failed pagination and fan-out still raise rather than returning partial data, and no public signatures or return shapes changed.
+
 **08/02/2026:** Phase 1 OGC boundary stabilization: the `dataretrieval.ogc` package now exposes a deliberate, small facade (`OgcDialect`, `prepare_request_args`, `get_ogc_data`, `fetch_ogc_request`) that service adapters (NGWMN, Water Data's generic wrapper) import. Internal request-construction helpers moved to a new `ogc.requests` module; the dialect type and endpoint constants live in the leaf `ogc.policy` module. `ogc.shaping` no longer depends on `ogc.engine` at all, and the complete runtime OGC import graph is now acyclic. `_default_headers` now accepts a target URL and adds `X-Api-Key` only for `api.waterdata.usgs.gov`; shared sync and async clients also strip the key before following any cross-host redirect, including external rating-asset downloads. `waterdata.utils` no longer bulk-re-exports private OGC symbols, and consumers import implementation helpers from their canonical modules. No public API, return-value, or deprecation changes.
 
 **08/02/2026:** Fixed source-distribution and wheel package discovery so the `dataretrieval.ogc` and `dataretrieval.waterdata` subpackages are included in installed artifacts. CI now builds and installs the wheel outside the source checkout before importing the core service modules. Added an architecture baseline, initial decision records, and executable dependency-direction guardrails for the existing modular-monolith boundaries.

dataretrieval/__init__.py

@@ -32,6 +32,7 @@
     __version__ = "version-unknown"
 
 from dataretrieval.exceptions import (
+    ConfigurationError,
     DataRetrievalError,
     HTTPError,
     NetworkError,
@@ -84,6 +85,7 @@
     # error taxonomy (canonical home: ``dataretrieval.exceptions``), re-exported
     # so callers can ``except dataretrieval.DataRetrievalError``
     "exceptions",
+    "ConfigurationError",
     "DataRetrievalError",
     "HTTPError",
     "NetworkError",

dataretrieval/exceptions.py

@@ -11,7 +11,8 @@
 of which :class:`TransientError` (429 / 5xx) is the retryable subset. The rest
 aren't a plain status: :class:`RequestTooLarge` (with :class:`URLTooLong` /
 :class:`Unchunkable`), :class:`NetworkError` (a failed connection, per above),
-and :class:`NoSitesError`. :func:`error_for_status` maps a status to its type.
+:class:`NoSitesError`, and :class:`ConfigurationError` for an unusable setting.
+:func:`error_for_status` maps a status to its type.
 
 This module has no third-party runtime dependencies -- ``httpx`` is imported only
 for type checking -- so any module can import it without pulling in pandas / httpx
@@ -36,6 +37,7 @@
     "Unchunkable",
     "NetworkError",
     "NoSitesError",
+    "ConfigurationError",
     "error_for_status",
 ]
 
@@ -240,6 +242,20 @@ class NetworkError(DataRetrievalError):
     retryable: ClassVar[bool] = True
 
 
+# --- Bad configuration ---------------------------------------------------
+
+
+class ConfigurationError(DataRetrievalError, ValueError):
+    """A ``dataretrieval`` setting -- an environment variable, a policy field --
+    holds a value that can't be used, so no request was issued.
+
+    It is a :class:`DataRetrievalError` so ``except`` around a retrieval catches
+    it rather than letting a bare ``ValueError`` escape a request path, and a
+    :class:`ValueError` so code that already treats a bad setting as one keeps
+    working.
+    """
+
+
 # --- Empty result --------------------------------------------------------
 
 

dataretrieval/nldi.py

@@ -3,7 +3,7 @@
 from json import JSONDecodeError
 from typing import Any, Literal, cast
 
-from dataretrieval.utils import query
+from dataretrieval.utils import _query_with_retry
 
 try:
     import geopandas as gpd
@@ -23,7 +23,7 @@ def _query_nldi(
     # A helper function to query the NLDI API. ``query()`` already raises a
     # typed ``DataRetrievalError`` for any HTTP error response, so a returned
     # response is a success that we only need to parse.
-    response = query(url, payload=query_params)
+    response = _query_with_retry(url, payload=query_params)
     response_data: dict[str, Any] | list[Any] = {}
     try:
         response_data = response.json()

dataretrieval/ogc/__init__.py

@@ -8,10 +8,9 @@
 - :func:`fetch_ogc_request` — execute a pre-built request with pagination.
 
 Service adapters (NGWMN, Water Data's generic wrapper) import from this
-facade rather than reaching into engine internals. The engine module remains
-available for lower-level orchestration needs (e.g. ``_paginate``,
-``_run_sync``) that sibling modules like ``wateruse`` use under the accepted
-temporary variance.
+facade rather than reaching into engine internals. Generic execution policy
+lives in :mod:`dataretrieval.transport`; the engine retains compatibility
+wrappers at previous private paths.
 """
 
 from dataretrieval.ogc.engine import fetch_ogc_request, get_ogc_data

dataretrieval/ogc/chunking.py

@@ -17,15 +17,13 @@
 
 This module owns the *execution* half — the event loop and bounded
 concurrency that drive a plan to completion (``ChunkedCall``) plus the
-public ``multi_value_chunked`` decorator. The neighboring concerns live in
-sibling modules it imports, each with its own reason to change:
-:mod:`~dataretrieval.ogc.planning` builds the
-:class:`~dataretrieval.ogc.planning.ChunkPlan` and recombines per-chunk
-frames and responses (pure, no I/O); :mod:`~dataretrieval.ogc.retry` holds
-the transient-classification and exponential-backoff policy; and
+public ``multi_value_chunked`` decorator. The neighboring concerns remain
+separate: :mod:`~dataretrieval.ogc.planning` builds the
+:class:`~dataretrieval.ogc.planning.ChunkPlan`;
+:mod:`~dataretrieval.transport.combining` assembles results;
+:mod:`~dataretrieval.transport.retry` owns bounded retry policy; and
 :mod:`~dataretrieval.ogc.interruptions` defines the resumable
-:class:`~dataretrieval.ogc.interruptions.ChunkInterrupted` exception
-contract.
+:class:`~dataretrieval.ogc.interruptions.ChunkInterrupted` contract.
 
 Concurrency: ``multi_value_chunked`` fans every pending sub-request out
 under one ``asyncio.gather`` sharing a single ``httpx.AsyncClient``. An
@@ -83,23 +81,20 @@
 import pandas as pd
 from anyio.from_thread import start_blocking_portal
 
-from dataretrieval.utils import HTTPX_ASYNC_DEFAULTS, Ambient, _require_positive_int
-
-from . import progress as _progress
-from .combining import (
+from dataretrieval.exceptions import ConfigurationError
+from dataretrieval.transport import progress as _progress
+from dataretrieval.transport.combining import (
     _combine_chunk_frames,
     _combine_chunk_responses,
 )
-from .interruptions import (
-    ChunkInterrupted,
-)
+from dataretrieval.transport.http import open_async_client
+from dataretrieval.transport.retry import _NO_RETRY, RetryPolicy
+from dataretrieval.transport.retry import retry_async as _retry
+from dataretrieval.utils import Ambient, _require_positive_int
+
+from .interruptions import ChunkInterrupted
 from .planning import ChunkPlan
-from .retry import (
-    _NO_RETRY,
-    RetryPolicy,
-    _classify_chunk_error,
-    _retry,
-)
+from .retry import _classify_chunk_error
 
 # Empirically the API replies HTTP 414 above ~8200 bytes of full URL —
 # matches nginx's default ``large_client_header_buffers`` of 8 KB. 8000
@@ -140,12 +135,12 @@ def _read_concurrency_env() -> int | None:
     try:
         value = int(raw)
     except ValueError as exc:
-        raise ValueError(
+        raise ConfigurationError(
             f"{_CONCURRENCY_ENV} must be a positive integer or "
             f"'{_CONCURRENCY_UNBOUNDED}'; got {raw!r}."
         ) from exc
     if value < 1:
-        raise ValueError(
+        raise ConfigurationError(
             f"{_CONCURRENCY_ENV} must be >= 1 (got {value}); use "
             f"'{_CONCURRENCY_UNBOUNDED}' to disable the cap."
         )
@@ -650,31 +645,19 @@ async def _run(self, max_concurrent: int | None) -> tuple[pd.DataFrame, Any]:
             self.plan.total if max_concurrent is None else max_concurrent
         )
 
-        async with httpx.AsyncClient(limits=limits, **HTTPX_ASYNC_DEFAULTS) as client:
+        async with open_async_client(limits=limits) as client:
             with _chunked_client(client):
                 reporter = _progress.current()
                 if reporter is not None:
                     reporter.set_chunks(self.plan.total)
 
-                async def fetch_gated(
-                    args: dict[str, Any],
-                ) -> tuple[pd.DataFrame, httpx.Response]:
-                    """One fetch attempt under the concurrency gate.
-
-                    The slot is held for the attempt's full duration —
-                    every page of a paginated sub-request — but acquired
-                    per *attempt* (this is what ``_retry`` re-invokes), so
-                    a sub-request sleeping off a retry backoff isn't
-                    holding a slot while it isn't touching the server.
-                    """
-                    async with semaphore:
-                        return await self.fetch(args)
-
                 async def track(
                     index: int, args: dict[str, Any]
                 ) -> tuple[pd.DataFrame, httpx.Response]:
                     """One sub-request (with retry) + result-store + progress tick."""
-                    result = await _retry(lambda: fetch_gated(args), self.retry_policy)
+                    result = await _retry(
+                        lambda: self.fetch(args), self.retry_policy, gate=semaphore
+                    )
                     self._chunks[index] = result
                     if reporter is not None:
                         # Chunks finish out of order under gather, so tick the
@@ -683,7 +666,7 @@ async def track(
                     return result
 
                 # Dispatch every pending sub-request concurrently; the
-                # semaphore (via ``fetch_gated``) is the only throttle.
+                # semaphore (held by ``_retry`` per attempt) is the only throttle.
                 # ``return_exceptions`` keeps completed pairs after a sibling
                 # fails, so partial state stays recoverable via :meth:`resume`.
                 # Failure precedence, in order: