refactor: police the fan-out abstractions

.importlinter

@@ -10,10 +10,9 @@
 
 [importlinter]
 root_package = dataretrieval
-; Contracts describe what runs, matching the AST suite. ``ogc.interruptions``
-; and ``ogc.chunking`` reference each other's types under ``TYPE_CHECKING``;
-; that is a documentation edge, not a runtime one, and no cycle exists at
-; import time.
+; Contracts describe what runs, matching the AST suite. Type-checking imports
+; document structural protocols and callback types without creating runtime
+; dependency edges.
 exclude_type_checking_imports = True
 
 [importlinter:contract:layers]
@@ -25,9 +24,13 @@ layers =
     ngwmn | nldi | nwis | streamstats | waterdata | wateruse | wqp
     ogc
     utils
+    _querying
     transport
     progress
-    _ambient | _response_metadata | codes | combining | rdb
+; Response-format conventions sit above the pure leaves because they read the
+; code tables, and below every adapter that shapes a response with them.
+    _wqx
+    _ambient | _response_metadata | codes | combining | interruptions | rdb
     credentials
     exceptions
 ; Every top-level module must be placed in the stack deliberately. A new
@@ -55,10 +58,9 @@ allowed_importers =
     dataretrieval.ngwmn
     dataretrieval.waterdata
 ignore_imports =
-; The package __init__ re-exports the resumable-call and interruption types;
-; they are part of the documented public surface, not a service reaching in.
+; The package __init__ re-exports the parallel-chunks context manager; it is
+; part of the documented public surface, not a service reaching into OGC.
     dataretrieval -> dataretrieval.ogc.chunking
-    dataretrieval -> dataretrieval.ogc.interruptions
 
 [importlinter:contract:ogc-facade]
 name = NGWMN consumes the OGC facade only, never its internals (ADR 0007)
@@ -91,16 +93,19 @@ type = forbidden
 source_modules =
     dataretrieval.ogc
 forbidden_modules =
+    dataretrieval._querying
     dataretrieval.utils
 
 [importlinter:contract:nwis-quarantine]
 name = Deprecated NWIS has no dependents (ADR 0005)
 type = forbidden
 source_modules =
+    dataretrieval._querying
     dataretrieval.codes
     dataretrieval.combining
     dataretrieval.credentials
     dataretrieval.exceptions
+    dataretrieval.interruptions
     dataretrieval.ngwmn
     dataretrieval.nldi
     dataretrieval.ogc

.pyscn-known-clones.json

@@ -0,0 +1,108 @@
+{
+  "_comment": [
+    "Accepted pyscn clone groups. Purpose: make a NEW clone group a visible event.",
+    "The duplication sub-score is dominated by these five accepted families, so it",
+    "barely moves when real duplication is added -- this file is what the weekly",
+    "Code Health workflow diffs against instead.",
+    "",
+    "Groups are keyed by the SET OF FUNCTION NAMES they contain, not by line numbers,",
+    "so ordinary edits above a getter do not churn this file. Regenerate only when",
+    "adding or removing a getter from an accepted family, and say why in the commit.",
+    "",
+    "These are NOT a backlog. Collapsing them means **kwargs and the loss of the typed",
+    "public surface tests/contracts/public_api_test.py freezes.",
+    "Verified 2026-08-09: at similarity_threshold 0.55 (default 0.65) the package still",
+    "reports exactly these five groups, so no latent clone sits below the bar.",
+    "Verified 2026-08-09: stripping docstrings makes duplication far WORSE (60 -> 0),",
+    "so suppressing them is not an option -- they are what keeps similarity below 0.86."
+  ],
+  "pyscn_version": "1.29.0",
+  "default_thresholds": {
+    "similarity_threshold": 0.65,
+    "min_lines": 10,
+    "min_nodes": 20
+  },
+  "observed": {
+    "total_fragments": 283,
+    "cloned_fragments": 25,
+    "groups": 5
+  },
+  "groups": [
+    {
+      "members": [
+        "dataretrieval/ngwmn.py::get_lithology",
+        "dataretrieval/ngwmn.py::get_providers",
+        "dataretrieval/ngwmn.py::get_well_construction"
+      ],
+      "similarity": 0.831,
+      "clone_type": 2,
+      "rationale": "NGWMN collection getters over the OGC facade, one per collection."
+    },
+    {
+      "members": [
+        "dataretrieval/nwis.py::get_dv",
+        "dataretrieval/nwis.py::get_iv"
+      ],
+      "similarity": 0.85,
+      "clone_type": 2,
+      "rationale": "Deprecated NWIS getters sharing the waterservices call shape; bodies already delegate to _get_json_values."
+    },
+    {
+      "members": [
+        "dataretrieval/waterdata/measurements.py::get_channel",
+        "dataretrieval/waterdata/measurements.py::get_field_measurements",
+        "dataretrieval/waterdata/measurements.py::get_peaks",
+        "dataretrieval/waterdata/metadata.py::get_field_measurements_metadata",
+        "dataretrieval/waterdata/metadata.py::get_monitoring_locations",
+        "dataretrieval/waterdata/metadata.py::get_time_series_metadata",
+        "dataretrieval/waterdata/time_series.py::get_continuous",
+        "dataretrieval/waterdata/time_series.py::get_daily",
+        "dataretrieval/waterdata/time_series.py::get_latest_continuous",
+        "dataretrieval/waterdata/time_series.py::get_latest_daily"
+      ],
+      "similarity": 0.805,
+      "clone_type": 2,
+      "rationale": "Water Data collection-family getters: ~30 explicitly typed parameters and a worked-example docstring around a thin body."
+    },
+    {
+      "members": [
+        "dataretrieval/waterdata/time_series.py::get_stats_date_range",
+        "dataretrieval/waterdata/time_series.py::get_stats_por"
+      ],
+      "similarity": 0.85,
+      "clone_type": 2,
+      "rationale": "Water Data collection-family getters: ~30 explicitly typed parameters and a worked-example docstring around a thin body."
+    },
+    {
+      "members": [
+        "dataretrieval/wqp.py::what_activities",
+        "dataretrieval/wqp.py::what_activity_metrics",
+        "dataretrieval/wqp.py::what_detection_limits",
+        "dataretrieval/wqp.py::what_habitat_metrics",
+        "dataretrieval/wqp.py::what_organizations",
+        "dataretrieval/wqp.py::what_project_weights",
+        "dataretrieval/wqp.py::what_projects",
+        "dataretrieval/wqp.py::what_sites"
+      ],
+      "similarity": 0.85,
+      "clone_type": 2,
+      "rationale": "WQP what_* wrappers: one shared query per service endpoint, differing only by the service path segment and the column each documents."
+    }
+  ],
+  "sub_threshold_accepted": {
+    "_comment": "Families that are NOT reported at default thresholds (they sit under min_lines=10 / min_nodes=20) but were found by a deliberate scan at min_nodes=8 on 2026-08-09. Recorded so they are not rediscovered and re-litigated. They are NOT counted in `observed` above.",
+    "groups": [
+      {
+        "members": [
+          "dataretrieval/nwis.py::get_discharge_measurements",
+          "dataretrieval/nwis.py::get_gwlevels",
+          "dataretrieval/nwis.py::get_pmcodes",
+          "dataretrieval/nwis.py::get_qwdata",
+          "dataretrieval/nwis.py::get_water_use"
+        ],
+        "similarity": 0.83,
+        "rationale": "Deprecated NWIS getters (ADR 0005). The surface is frozen pending removal, so collapsing it buys nothing and churns a public API users are being migrated off."
+      }
+    ]
+  }
+}

CONTRIBUTING.md

@@ -113,6 +113,17 @@ about the upstream service rather than about this package.
 
 ### Coding Standards and Style
 
+**Before adding a small helper, check whether a leaf already generalizes it.**
+This package keeps its general mechanisms in dependency-free leaves --
+`_ambient.Ambient` for scoped context values, `transport.retry._read_env_number`
+for `API_USGS_*` settings, `transport.links.resolve_next_url` for pagination
+cursors. Each of those has been re-implemented at least once by someone who
+did not know it was there, and the copies drift: the same question gets a
+different cycle guard, a different error message, a different edge case. None
+of the automated checks catch it, because two eight-line helpers are below the
+clone detector's floor and neither one couples or complicates anything. A grep
+for the mechanism you are about to write is the only thing that does.
+
 The continuous integration and pre-commit configurations enforce formatting,
 linting, and strict type checking. Run the relevant checks before opening a PR:
 

NEWS.md

@@ -1,3 +1,9 @@
+**08/09/2026:** Every retrieval path now runs through one executor. `waterdata.get_cql` (via the OGC `fetch_ogc_request`) and `waterdata.get_stats_por` / `get_stats_date_range` (via the Statistics page walk) previously bypassed `dataretrieval.transport.fanout.FanOut` through a private sync bridge, which meant they were the only getters in the package with **no retry**: a mid-page-walk 429 or 503 failed the whole call while every typed getter and Water Use rode it out. Both now run as a one-item fan-out and the 25-line `transport/sync.py` is gone. **Behavior change:** those three getters now retry transient failures (`API_USGS_RETRIES`, default 4) and, when the retries are exhausted, raise the resumable `ServiceInterrupted` / `QuotaExhausted` rather than `ServiceUnavailable` / `RateLimited` / `NetworkError` — all remain `DataRetrievalError`, so broad handlers are unaffected, but narrow handlers around those calls must widen, and `.call.resume()` is now available on the interruption. A failure that retrying cannot fix (bad scheme, a hostname that does not resolve) still surfaces as `NetworkError` immediately. The progress line moved with it: `FanOut.resume()` opens the reporter it ticks into, so a driver can no longer run the shared executor and silently print nothing, and a `.call.resume()` fired long after the interruption now reports progress instead of running mute. Internal tidying with no public effect: the WQX3 / legacy-WQP CSV datetime shaping moved out of `dataretrieval.utils` (whose docstring reserves it for non-service-specific shaping) into the `dataretrieval._wqx` leaf; the five Water Data endpoint URLs are declared once in `dataretrieval.waterdata.endpoints` instead of being derived in three modules; the OGC queryables document is parsed by `dataretrieval.ogc.schema` so every OGC adapter can offer the table, with `waterdata.get_queryables` unchanged as its documented wrapper; and `ogc/engine.py` imports each symbol from the module that defines it.
+
+**08/09/2026:** Internal structure cleanup, no public API change. Validating a server-supplied next-page link is now one policy in `dataretrieval.transport.links` instead of three divergent copies (the OGC engine, the ratings STAC walk, and Water Use). Two of those copies were fixed by the merge: the OGC page walk now resolves a *relative* `next` href against the page it came from (it previously handed the unresolved reference back as the pagination cursor) and refuses an unparseable one rather than following it unchecked. Cross-host refusal, credential stripping, and Water Use's host-alias rewrite are unchanged, as is the error type each walk raises. `parse_retry_after` moved to `dataretrieval.exceptions`, next to the `DataRetrievalError.retry_after` field it exists to produce. The one-shot HTTP query path (`query`, `to_str`, and their helpers) moved out of `dataretrieval.utils` into the private `dataretrieval._querying`; `dataretrieval.utils.query` and `dataretrieval.utils.to_str` remain the documented public paths, as `Ambient` and `BaseMetadata` already do. `waterdata` profile validation moved next to the tables it validates in `waterdata.types`, and `nwis.get_dv`/`get_iv` now share one body.
+
+**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, 429, or recoverable connection failure now raises `ServiceInterrupted`/`QuotaExhausted` rather than `ServiceUnavailable`/`RateLimited`/`NetworkError` — all remain `DataRetrievalError`, so broad handlers are unaffected, but narrow handlers 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/03/2026:** Split the Water Data implementation into focused time-series, metadata, measurements, reference, samples, and CQL collection-family modules behind the unchanged `waterdata.api` facade. Active service modules now declare explicit exports; public Water Data imports, signatures, function identities, deprecations, and return contracts are protected by executable contract snapshots. OGC ambient context and schema/queryables execution are separated from request construction, adapter-to-adapter reach-through is prohibited by architecture tests, and service-specific output shapes are documented rather than forced into one model.
 
 **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.

dataretrieval/__init__.py

@@ -19,9 +19,10 @@
 
 A failed request raises a subclass of :class:`dataretrieval.DataRetrievalError`
 (the taxonomy lives in ``dataretrieval.exceptions``); connection-level failures
-(timeouts, DNS) are wrapped as :class:`dataretrieval.NetworkError`. A large
-request interrupted mid-stream raises :class:`dataretrieval.ChunkInterrupted`,
-whose ``.call.resume()`` continues from the work already completed.
+(timeouts, DNS) are wrapped as :class:`dataretrieval.NetworkError`. A fanned-out
+request interrupted mid-stream raises :class:`dataretrieval.FanOutInterrupted`
+(also available under its original ``ChunkInterrupted`` name), whose
+``.call.resume()`` continues from the work already completed.
 """
 
 from importlib.metadata import PackageNotFoundError, version
@@ -45,23 +46,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,
@@ -96,8 +98,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)