refactor(transport)!: share fan-out execution across services

.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,7 @@
+**08/06/2026:** Fan-out execution is now shared across services. Chunking is how a query is divided structurally (a Water Data/NGWMN URL over the byte limit); fan-out is how the pieces are distributed operationally. Only the first is protocol-specific, so the executor moved to `dataretrieval.transport.fanout` (`FanOut`, over a three-member `FanOutPlan` protocol) while chunk planning stays in `dataretrieval.ogc`. Water Use no longer re-implements the fan-out gather and inherits resume, progress reporting, and `API_USGS_CONCURRENT`: a multi-location pull interrupted by a rate limit now raises a resumable interruption whose `.call.resume()` re-issues only the locations that did not finish, instead of discarding every completed one. The interruption taxonomy moved to the `dataretrieval.interruptions` leaf and its base class is now `FanOutInterrupted`; **`ChunkInterrupted` is a permanent alias of the same class**, so `except ChunkInterrupted` keeps working. **Breaking change:** a Water Use fan-out interrupted by a 5xx or 429 now raises `ServiceInterrupted`/`QuotaExhausted` rather than `ServiceUnavailable`/`RateLimited` — both are still `DataRetrievalError`, so broad handlers are unaffected, but a narrow `except ServiceUnavailable` around a Water Use call must widen. **Breaking change:** `wateruse.MAX_CONCURRENT_REQUESTS` is removed; set `API_USGS_CONCURRENT` (which now outranks any service default) or read `wateruse.DEFAULT_CONCURRENT_REQUESTS`.
+
+**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,
@@ -44,23 +45,24 @@
     URLTooLong,
 )
 
-# Parallel-chunks control (a context manager). Defined with the chunker in
-# ``dataretrieval.ogc.chunking``; surfaced here for a stable public path
-# ``from dataretrieval import parallel_chunks``.
-from dataretrieval.ogc.chunking import parallel_chunks
-
-# Resumable chunk-interruption exceptions. They are defined in
-# ``dataretrieval.ogc.interruptions`` rather than ``dataretrieval.exceptions``
-# because they carry pandas/httpx state and a resumable ``ChunkedCall`` handle,
+# Resumable fan-out interruption exceptions. They are defined in
+# ``dataretrieval.interruptions`` rather than ``dataretrieval.exceptions``
+# because they carry pandas/httpx state and a resumable ``FanOut`` handle,
 # which would pull heavy dependencies into the lightweight exceptions module.
-# Surfaced here so callers get a stable public path:
-# ``from dataretrieval import ChunkInterrupted``.
-from dataretrieval.ogc.interruptions import (
+# They are not under ``ogc`` because Water Use raises them too. Surfaced here so
+# callers get a stable public path: ``from dataretrieval import ChunkInterrupted``.
+from dataretrieval.interruptions import (
     ChunkInterrupted,
+    FanOutInterrupted,
     QuotaExhausted,
     ServiceInterrupted,
 )
 
+# Parallel-chunks control (a context manager). Defined with the chunker in
+# ``dataretrieval.ogc.chunking``; surfaced here for a stable public path
+# ``from dataretrieval import parallel_chunks``.
+from dataretrieval.ogc.chunking import parallel_chunks
+
 from . import (
     exceptions,
     ngwmn,
@@ -84,6 +86,7 @@
     # error taxonomy (canonical home: ``dataretrieval.exceptions``), re-exported
     # so callers can ``except dataretrieval.DataRetrievalError``
     "exceptions",
+    "ConfigurationError",
     "DataRetrievalError",
     "HTTPError",
     "NetworkError",
@@ -94,8 +97,9 @@
     "TransientError",
     "URLTooLong",
     "Unchunkable",
-    # resumable chunk-interruption exceptions (defined in ogc.interruptions)
+    # resumable fan-out interruption exceptions (defined in interruptions)
     "ChunkInterrupted",
+    "FanOutInterrupted",
     "QuotaExhausted",
     "ServiceInterrupted",
     # parallel-chunks control (defined in ogc.chunking)

dataretrieval/ogc/combining.py → dataretrieval/combining.py

@@ -2,12 +2,16 @@
 
 These utilities assemble the output of a chunked/fan-out call from its
 individual per-sub-request results.  They have no event-loop, retry, or
-network state — they're pure data transforms imported by both the
-chunked-call execution (:mod:`dataretrieval.ogc.chunking`) and the
-per-page pagination (:mod:`dataretrieval.ogc.engine`).
+network state — they're pure data transforms shared by protocol-specific
+chunk execution, service fan-out, and
+cursor-driven pagination.
 
 Separated from :mod:`dataretrieval.ogc.planning` so that module stays
 focused on *what* to split, while this module owns *how* to reassemble.
+
+A top-level leaf rather than part of :mod:`dataretrieval.transport`: these are
+pandas transforms over already-fetched results, with no HTTP or event-loop
+concern, consumed by chunk planning and service fan-out as well as by pagination.
 """
 
 from __future__ import annotations
@@ -104,8 +108,8 @@ def _merge_response(
     ``httpx.Headers`` means downstream mutations don't back-propagate into any
     underlying response — so callers may re-fold idempotently.  This is the one
     low-level merge behind both pagination
-    (:func:`~dataretrieval.ogc.engine._paginate`) and the chunked / fan-out
-    aggregation (:func:`_combine_chunk_responses`)."""
+    (:func:`~dataretrieval.transport.pagination.paginate`) and the chunked /
+    fan-out aggregation (:func:`_combine_chunk_responses`)."""
     merged = copy.copy(base)
     merged.headers = httpx.Headers(headers_from.headers)
     merged.elapsed = elapsed

dataretrieval/credentials.py

@@ -0,0 +1,97 @@
+"""Which host honors the USGS API key, and how it is attached and withheld.
+
+One leaf owns every answer about the ``API_USGS_PAT`` credential: the host that
+accepts it, whether a given destination qualifies, and how it is stripped back
+off a request bound somewhere else. Splitting those answers across the layers
+that happen to need them is how a credential reaches a host nobody authorized:
+the code that attaches a key and the code that removes it have to agree, and the
+only way to guarantee they agree is to have them read the same predicate.
+
+This is deliberately a leaf. It sits below HTTP mechanics (which attaches the
+header) and below progress reporting (which tells an unauthenticated caller where
+to register), so neither has to depend on the other to learn the same fact.
+"""
+
+from __future__ import annotations
+
+import os
+
+import httpx
+
+#: Environment variable holding the USGS Water Data personal access token.
+API_KEY_ENV = "API_USGS_PAT"
+
+#: Where to register for a key. Surfaced once, by the progress reporter, when a
+#: query against the authorized host runs without one -- unauthenticated callers
+#: hit much lower rate limits (see the ``API_USGS_PAT`` note in the README).
+SIGNUP_URL = "https://api.waterdata.usgs.gov/signup/"
+
+#: The only host that honors the key. Every other service this package talks to
+#: ignores it, so sending it there would leak a credential for no benefit.
+_AUTHORIZED_API_KEY_HOST = "api.waterdata.usgs.gov"
+
+#: Origin of the Water Data API, built from the authorized host rather than
+#: spelled again. The host that serves these endpoints and the host that honors
+#: the key are the same fact, and the failure mode of keeping two copies is
+#: silent: the endpoint moves, the predicate does not follow, and either the key
+#: quietly stops attaching or it is sent somewhere nobody authorized. The
+#: adapters import this instead of restating the authority (enforced by
+#: ``test_credential_policy_has_one_definition``).
+WATERDATA_BASE_URL = f"https://{_AUTHORIZED_API_KEY_HOST}"
+
+
+def accepts_api_key(target_url: str | httpx.URL | None) -> bool:
+    """Whether ``target_url`` names the host that honors :data:`API_KEY_ENV`.
+
+    The single answer to "does this destination get the key" -- used when
+    attaching the credential, when stripping it back off at redirect time, and
+    when deciding whether "get an API key" is useful advice rather than noise, so
+    the three can't drift apart.
+
+    The scheme has to be ``https``, not just the host. A bearer token sent over
+    cleartext is readable by anything on the path, and the destination that would
+    receive it is reachable through data we do not control: a redirect, or a
+    server-supplied next-page link naming ``http://`` on the very host that is
+    otherwise authorized. Matching on the host alone would hand the key over in
+    the clear on the strength of a hostname the attacker chose to keep.
+    """
+    if target_url is None:
+        return False
+    try:
+        url = target_url if isinstance(target_url, httpx.URL) else httpx.URL(target_url)
+    except (httpx.InvalidURL, TypeError):
+        return False
+    return url.scheme == "https" and url.host == _AUTHORIZED_API_KEY_HOST
+
+
+def without_embedded_credentials(url: httpx.URL) -> httpx.URL:
+    """Drop any ``user:pass@`` from a URL we were *handed* rather than built.
+
+    A next-page link is data, not configuration. ``httpx`` derives an
+    ``Authorization: Basic`` header from userinfo in a URL, so a poisoned link
+    carrying ``user:pass@`` mints a credential the caller never configured and
+    sends it onward -- next to the real API key, when the host still checks out
+    and the host check therefore raises nothing. No USGS service authenticates
+    that way, so stripping it costs a legitimate caller nothing.
+    """
+    return url.copy_with(userinfo=b"") if url.userinfo else url
+
+
+def api_key() -> str | None:
+    """The configured token, or ``None``.
+
+    Read through a function rather than captured at import so a caller that sets
+    the variable after import -- or a test that patches it -- is still honored.
+    """
+    return os.getenv(API_KEY_ENV)
+
+
+def strip_api_key_from_untrusted_host(request: httpx.Request) -> None:
+    """Remove Water Data credentials before sending to any other host."""
+    if not accepts_api_key(request.url):
+        request.headers.pop("X-Api-Key", None)
+
+
+async def strip_api_key_from_untrusted_host_async(request: httpx.Request) -> None:
+    """Async-client form of :func:`strip_api_key_from_untrusted_host`."""
+    strip_api_key_from_untrusted_host(request)

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 --------------------------------------------------------