refactor(transport): extract API-neutral execution policy

Author: thodson-usgs

.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
 

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/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,19 @@
 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.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
@@ -650,7 +644,7 @@ 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:

dataretrieval/ogc/combining.py

@@ -1,206 +1,21 @@
-"""Result recombination: merge per-chunk frames and responses (no I/O).
-
-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`).
-
-Separated from :mod:`dataretrieval.ogc.planning` so that module stays
-focused on *what* to split, while this module owns *how* to reassemble.
-"""
-
-from __future__ import annotations
-
-import copy
-from datetime import timedelta
-
-import httpx
-import pandas as pd
-
-# Response header USGS uses to advertise remaining hourly quota.  Lives in this
-# module so every layer (the combine helpers below, the engine's per-page
-# progress reporter) reads it from one place rather than hard-coding the string.
-_QUOTA_HEADER = "x-ratelimit-remaining"
-
-
-def _safe_elapsed(response: httpx.Response) -> timedelta:
-    """
-    Read ``response.elapsed``, falling back to ``timedelta(0)`` when
-    the attribute hasn't been populated.
-
-    httpx only writes ``.elapsed`` when a response is closed through
-    its normal transport path. ``MockTransport`` (used by
-    ``pytest-httpx``) and hand-constructed ``httpx.Response`` objects
-    leave the attribute unset, so accessing it raises ``RuntimeError``.
-    Combining responses across chunks needs a defined duration, so we
-    treat the missing attribute as zero elapsed.
-    """
-    try:
-        elapsed: object = response.elapsed
-    except RuntimeError:
-        return timedelta(0)
-    return elapsed if isinstance(elapsed, timedelta) else timedelta(0)
-
-
-def _set_response_url(response: httpx.Response, url: str | httpx.URL) -> None:
-    """
-    Overwrite the URL surfaced by a response without back-propagating
-    the change into any aliased original.
-
-    Lightweight test doubles expose ``.url`` as a writable attribute. Real
-    :class:`httpx.Response` objects resolve it through a bound request, so swap
-    in a fresh request carrying the new URL; mutating the existing request would
-    leak through any shallow copy that shares it.
-    """
-    if not isinstance(response, httpx.Response):
-        # Lightweight test doubles expose ``url`` as a writable attribute.
-        response.url = url
-        return
-
-    target = httpx.URL(str(url))
-    try:
-        old = response.request
-    except RuntimeError:
-        # No request bound (some hand-built httpx.Response fixtures);
-        # synthesize a minimal one to hold the URL.
-        response.request = httpx.Request("GET", target)
-        return
-    response.request = httpx.Request(method=old.method, url=target, headers=old.headers)
-
-
-def _lowest_remaining(responses: list[httpx.Response]) -> httpx.Response:
-    """The response reporting the lowest ``x-ratelimit-remaining``.
-
-    Within a rate-limit window, the counter decreases monotonically, so the
-    smallest value observed is the most conservative value to surface. Under
-    concurrent fan-out, the last response *by index* need not be the one the
-    server processed last. Fall back to the last response when none reports
-    the header.
-    """
-    best: httpx.Response | None = None
-    best_remaining: int | None = None
-    for response in responses:
-        try:
-            remaining = int(response.headers[_QUOTA_HEADER])
-        except (KeyError, ValueError):
-            continue
-        if best_remaining is None or remaining < best_remaining:
-            best, best_remaining = response, remaining
-    return best if best is not None else responses[-1]
-
-
-def _merge_response(
-    base: httpx.Response,
-    *,
-    headers_from: httpx.Response,
-    elapsed: timedelta,
-    url: str | httpx.URL | None = None,
-) -> httpx.Response:
-    """Fold several responses into one: a shallow copy of ``base`` whose
-    ``.headers`` are rebuilt as a fresh ``httpx.Headers`` from ``headers_from``,
-    ``.elapsed`` set to ``elapsed``, and ``.url`` overridden when ``url`` is
-    given.  ``base`` and ``headers_from`` are never mutated, and the fresh
-    ``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`)."""
-    merged = copy.copy(base)
-    merged.headers = httpx.Headers(headers_from.headers)
-    merged.elapsed = elapsed
-    if url is not None:
-        _set_response_url(merged, url)
-    return merged
-
-
-def _combine_chunk_frames(frames: list[pd.DataFrame]) -> pd.DataFrame:
-    """Concatenate per-chunk frames and deduplicate IDs across chunks.
-
-    Empty frames are ignored before concatenation so an empty plain
-    :class:`pandas.DataFrame` cannot downgrade a real ``GeoDataFrame`` and
-    strip its geometry or CRS. When every frame is empty, the first frame is
-    returned to preserve its concrete type.
-
-    When multiple non-empty frames are combined, non-null feature IDs are
-    deduplicated regardless of the plan axis. Filter clauses can match the same
-    feature, and list inputs can contain repeated values or otherwise select
-    overlapping records. Rows without an ``id`` are preserved verbatim: pandas
-    treats null values as duplicates, so deduplicating them would silently lose
-    data.
-    """
-    non_empty = [frame for frame in frames if not frame.empty]
-    if not non_empty:
-        return frames[0] if frames else pd.DataFrame()
-    if len(non_empty) == 1:
-        return non_empty[0].copy()
-
-    combined = pd.concat(non_empty, ignore_index=True)
-    if "id" not in combined.columns:
-        return combined
-
-    has_id = combined["id"].notna()
-    if has_id.all():
-        return combined.drop_duplicates(subset="id", ignore_index=True)
-    if has_id.any():
-        id_rows = combined[has_id].drop_duplicates(subset="id")
-        no_id_rows = combined[~has_id]
-        return pd.concat([id_rows, no_id_rows], ignore_index=True)
-    return combined
-
-
-def _combine_chunk_responses(
-    responses: list[httpx.Response], canonical_url: str | None
-) -> httpx.Response:
-    """
-    Fold per-sub-request responses into a single aggregated response.
-
-    For a multi-response input, returns a shallow copy of
-    ``responses[0]`` with ``.headers`` set to those of the response reporting
-    the lowest ``x-ratelimit-remaining`` value (the most conservative quota
-    observation; see :func:`_lowest_remaining`), ``.elapsed`` set to the sum of
-    the per-response elapsed durations, and ``.url`` set to the
-    canonical original-query URL (when supplied) so ``BaseMetadata``
-    reflects the user's full request rather than the first chunk.
-
-    For a single-response input with no canonical-URL override,
-    ``responses[0]`` is returned unchanged to skip the copy on the
-    passthrough hot path.
-
-    Parameters
-    ----------
-    responses : list[httpx.Response]
-        One response per completed sub-request, in caller-provided order.
-    canonical_url : str or None
-        URL of the unchunked original request. ``None`` skips the URL
-        override — used by the passthrough path (the fetcher's
-        response already carries the original-query URL) and by the
-        worst-case overflow path (no buildable canonical URL exists).
-
-    Returns
-    -------
-    httpx.Response
-        A shallow copy of the first response with aggregated
-        ``headers``, ``elapsed``, and ``url``.  The function is
-        idempotent (the input responses' ``headers`` / ``elapsed`` /
-        ``url`` are never mutated), so it's safe to call repeatedly
-        via :attr:`ChunkedCall.partial_response` during error
-        inspection or resume retries.  ``headers`` on the returned
-        object is a fresh ``httpx.Headers``, so mutations there don't
-        back-propagate into any chunk's underlying response.
-    """
-    if len(responses) == 1 and canonical_url is None:
-        return responses[0]
-
-    # Headers come from the response with the lowest reported remaining quota;
-    # ``_lowest_remaining`` returns the lone response as-is
-    # for a single-element list).  ``_merge_response`` re-sums elapsed onto a
-    # fresh copy, so repeated calls (e.g. via ``ChunkedCall.partial_response``
-    # during resume) stay idempotent.
-    elapsed = sum((_safe_elapsed(r) for r in responses), start=timedelta())
-    return _merge_response(
-        responses[0],
-        headers_from=_lowest_remaining(responses),
-        elapsed=elapsed,
-        url=canonical_url,
-    )
+"""Compatibility imports for response aggregation now owned by transport."""
+
+from dataretrieval.transport.combining import (
+    _QUOTA_HEADER,
+    _combine_chunk_frames,
+    _combine_chunk_responses,
+    _lowest_remaining,
+    _merge_response,
+    _safe_elapsed,
+    _set_response_url,
+)
+
+__all__ = [
+    "_QUOTA_HEADER",
+    "_combine_chunk_frames",
+    "_combine_chunk_responses",
+    "_lowest_remaining",
+    "_merge_response",
+    "_safe_elapsed",
+    "_set_response_url",
+]