Improve architecture health signals and dependency depth

.github/workflows/code-health.yml

@@ -51,29 +51,15 @@ jobs:
           pyscn analyze --json --no-open dataretrieval 2>&1 | tee pyscn-summary.txt
           pyscn analyze --html --no-open dataretrieval >/dev/null
 
-      - name: Check the analysis actually resolved the package
-        # pyscn infers a project root, and when it guesses wrong it silently
-        # resolves only a fraction of the imports -- which *raises* the score,
-        # because most of what it grades is dependency-derived. A degraded run
-        # therefore looks like an improved one. Record the resolved edge count
-        # next to the score so that is visible rather than flattering.
+      - name: Signals that mean something for this package
+        # pyscn grades six sub-scores and four of them do not apply here, so the
+        # composite moves for reasons that are not about this codebase. This
+        # extracts the three that do -- dead code, a NEW clone group, dependency
+        # depth -- and adds the interface cost of the public surface, which
+        # nothing else measures. See tools/health_signals.py for why each.
         if: always()
         continue-on-error: true
-        run: |
-          python - <<'PY' > pyscn-sanity.txt
-          import glob, json, os
-          reports = sorted(glob.glob(".pyscn/reports/*.json"))
-          if not reports:
-              print("no pyscn JSON report found"); raise SystemExit
-          s = json.load(open(reports[-1]))["system"]["Summary"]
-          root, deps = s["ProjectRoot"], s["TotalDependencies"]
-          print(f"modules={s['TotalModules']} resolved_dependencies={deps} root={root}")
-          if os.path.realpath(root) != os.path.realpath(os.getcwd()):
-              print(f"WARNING: project root {root!r} is not the checkout; "
-                    "import resolution is probably degraded and the scores "
-                    "above are not comparable to previous runs.")
-          PY
-          cat pyscn-sanity.txt
+        run: python tools/health_signals.py > health-signals.txt && cat health-signals.txt
 
       - name: Maintainability ranking (wily)
         # Worst-maintained files today, and how the package has moved recently.
@@ -92,15 +78,26 @@ jobs:
         if: always()
         run: |
           {
-            echo '## Structural analysis'
+            echo '## Signals'
+            echo '```'
+            cat health-signals.txt 2>/dev/null || echo 'signals unavailable'
+            echo '```'
+            echo
+            echo '<details><summary>Full pyscn scores (advisory)</summary>'
+            echo
+            echo 'The composite averages six sub-scores; four do not apply to a'
+            echo 'function-oriented package, and Architecture penalises a leaf'
+            echo 'for being depended upon. Read the signals above instead.'
+            echo
             echo '```'
             if [[ -s pyscn-summary.txt ]]; then
               cat pyscn-summary.txt
             else
               echo 'pyscn produced no output'
             fi
-            cat pyscn-sanity.txt 2>/dev/null || true
             echo '```'
+            echo '</details>'
+            echo
             cat wily-summary.txt 2>/dev/null || true
             echo
             echo 'Full reports are attached to this run as the'
@@ -117,7 +114,7 @@ jobs:
           path: |
             .pyscn/reports/
             pyscn-summary.txt
-            pyscn-sanity.txt
+            health-signals.txt
             wily-summary.txt
           retention-days: 90
           if-no-files-found: warn

dataretrieval/ngwmn.py

@@ -24,7 +24,7 @@
 
 from dataretrieval.codes.states import apply_state
 from dataretrieval.credentials import WATERDATA_BASE_URL
-from dataretrieval.ogc import OgcDialect, get_ogc_data, prepare_request_args
+from dataretrieval.ogc import OgcApi, OgcDialect, get_ogc_data, prepare_request_args
 
 if TYPE_CHECKING:
     from dataretrieval._response_metadata import BaseMetadata
@@ -89,6 +89,14 @@
     sort_cols=("sample_time", "monitoring_location_id"),
 )
 
+#: The NGWMN OGC API. It carries no synthetic id columns, so only the base URL
+#: and the dialect differ from the default.
+NGWMN_API = OgcApi(
+    base_url=NGWMN_OGC_API_URL,
+    dialect=NGWMN_DIALECT,
+    output_ids=_NGWMN_OUTPUT_ID,
+)
+
 
 def _get(service: str, local_vars: dict[str, Any]) -> tuple[pd.DataFrame, BaseMetadata]:
     """Marshal a getter's arguments and dispatch to the shared OGC facade.
@@ -103,9 +111,7 @@ def _get(service: str, local_vars: dict[str, Any]) -> tuple[pd.DataFrame, BaseMe
     return get_ogc_data(
         args,
         service,
-        output_id=_NGWMN_OUTPUT_ID,
-        base_url=NGWMN_OGC_API_URL,
-        dialect=NGWMN_DIALECT,
+        api=NGWMN_API,
     )
 
 

dataretrieval/ogc/__init__.py

@@ -2,6 +2,7 @@
 
 The public facade exposes only the minimal collection-adapter seam:
 
+- :class:`OgcApi` — one OGC API's identity: base URL, dialect, id columns.
 - :class:`OgcDialect` — per-API request/response quirks.
 - :func:`prepare_request_args` — normalize caller kwargs for the engine.
 - :func:`get_ogc_data` — full orchestrated OGC fetch (chunking + pagination).
@@ -14,10 +15,11 @@
 """
 
 from dataretrieval.ogc.engine import fetch_ogc_request, get_ogc_data
-from dataretrieval.ogc.policy import OgcDialect
+from dataretrieval.ogc.policy import OgcApi, OgcDialect
 from dataretrieval.ogc.requests import prepare_request_args
 
 __all__ = [
+    "OgcApi",
     "OgcDialect",
     "fetch_ogc_request",
     "get_ogc_data",

dataretrieval/ogc/chunking.py

@@ -8,14 +8,10 @@
 chunk URL under the budget. Requests that already fit get a
 trivial single-step plan — the executor has one code path either way.
 
-This module owns the OGC-specific half: the byte budget, the
-``parallel_chunks`` dial, and the ``multi_value_chunked`` decorator that
-ties a plan to a fetcher. Driving the resulting chunks to
-completion — bounded concurrency, retry, failure precedence, resume — is
-API-neutral and belongs to
-:class:`dataretrieval.transport.fanout.FanOut`, which this module hands
-its plan to. :class:`~dataretrieval.ogc.planning.ChunkPlan` satisfies
-:class:`~dataretrieval.transport.fanout.FanOutPlan` structurally.
+This module owns the public ``parallel_chunks`` dial and preserves compatibility
+names from the former combined planner/executor. The OGC engine composes
+planning with API-neutral fan-out directly, avoiding an extra dependency hop in
+every getter call.
 
 Parallel chunks: the planner is conservative by default — it splits only as
 far as the byte limit forces. A caller who knows their result is large can opt
@@ -36,26 +32,17 @@
 
 from __future__ import annotations
 
-import functools
-from collections.abc import Callable, Iterator
+from collections.abc import Iterator
 from contextlib import contextmanager
-from typing import Any
 
-import httpx
-import pandas as pd
-
-from dataretrieval._ambient import Ambient
-from dataretrieval.transport.fanout import (
-    FanOut,
-    _active_client,
-    _Fetch,
-    _Finalize,
-    _passthrough_result,
-    active_client,
+from dataretrieval.ogc.engine import (
+    _parallel_chunks,
+)
+from dataretrieval.ogc.engine import (
+    multi_value_chunked as multi_value_chunked,
 )
-from dataretrieval.transport.retry import RetryPolicy
+from dataretrieval.transport.fanout import FanOut, _active_client, active_client
 
-from .planning import ChunkPlan
 from .policy import _require_positive_int
 
 # Compatibility aliases. ``ChunkedCall`` was this module's executor before it
@@ -70,22 +57,6 @@
 get_active_client = active_client
 _chunked_client = _active_client
 
-# Empirically the API replies HTTP 414 above ~8200 bytes of full URL —
-# matches nginx's default ``large_client_header_buffers`` of 8 KB. 8000
-# leaves ~200 bytes for request-line framing and proxy variance. The decorator
-# resolves this module-level default at call time when ``url_limit`` is None,
-# so a test can ``monkeypatch.setattr`` it on this module.
-_OGC_URL_BYTE_LIMIT = 8000
-
-
-# Parallel-chunks dial: opt-in to fan a query out *more finely* than the byte
-# limit alone requires. Scoped to a ``with parallel_chunks(...):`` block (a
-# ContextVar), deliberately NOT an env var (see :func:`parallel_chunks` for
-# why). The ambient holds ``n`` — the requested cap on the plan's total
-# chunk count; ``1`` (the default, outside any block) means "off — chunk
-# only as much as the byte limit needs, no extra fan-out".
-_parallel_chunks: Ambient[int] = Ambient("ogc_parallel_chunks", 1)
-
 
 @contextmanager
 def parallel_chunks(n: int) -> Iterator[None]:
@@ -188,92 +159,3 @@ def parallel_chunks(n: int) -> Iterator[None]:
     _require_positive_int(n, "parallel_chunks(n)", examples="2, 8, 32")
     with _parallel_chunks(n):
         yield
-
-
-def multi_value_chunked(
-    *,
-    build_request: Callable[..., httpx.Request],
-    url_limit: int | None = None,
-) -> Callable[[_Fetch[dict[str, Any]]], Callable[..., tuple[pd.DataFrame, Any]]]:
-    """
-    Decorate an async fetcher to transparently chunk over-budget requests.
-
-    Returns a callable that builds a :class:`ChunkPlan` from ``args``,
-    constructs a :class:`ChunkedCall` over the decorated
-    ``async def fetch(args) -> (df, response)``, and drives it to
-    completion via :meth:`ChunkedCall.resume`. The plan splits multi-value
-    list params and the cql-text filter so each chunk URL fits the
-    byte limit. An already-fitting request is a one-step plan, unless an
-    active :func:`parallel_chunks` block asks the plan to fan out more
-    finely. See the module docstring for the concurrency model.
-
-    Parameters
-    ----------
-    build_request : Callable[..., httpx.Request]
-        Factory that turns a kwargs dict into a sized httpx request,
-        e.g. ``_construct_api_requests``. Called during planning to
-        measure each candidate plan.
-    url_limit : int, optional
-        Byte budget for the request (URL + body). When ``None``
-        (default), the module-level ``_OGC_URL_BYTE_LIMIT`` is
-        resolved at call time so test patches via
-        ``monkeypatch.setattr`` take effect.
-
-    Returns
-    -------
-    Callable
-        A *synchronous* wrapper ``wrapper(args, *, finalize=...) ->
-        (df, response)`` that executes the underlying plan transparently
-        over the decorated async fetcher.
-
-    Raises
-    ------
-    Unchunkable
-        If no plan can fit ``url_limit``.
-    ChunkInterrupted
-        On a mid-execution transient — 429, 5xx, or a bare transport
-        error: :class:`QuotaExhausted` for 429, :class:`ServiceInterrupted`
-        for the rest. See :class:`ChunkedCall` for the resume semantics.
-
-    See Also
-    --------
-    ChunkPlan : Planning shape (axes, partitioning, passthrough).
-    ChunkedCall : Per-chunk execution and resume semantics.
-    """
-
-    def decorator(
-        fetch: _Fetch[dict[str, Any]],
-    ) -> Callable[..., tuple[pd.DataFrame, Any]]:
-        @functools.wraps(fetch)
-        def wrapper(
-            args: dict[str, Any],
-            *,
-            finalize: _Finalize = _passthrough_result,
-        ) -> tuple[pd.DataFrame, Any]:
-            limit = _OGC_URL_BYTE_LIMIT if url_limit is None else url_limit
-            # Read the parallel_chunks dial ``n`` from the ambient set by
-            # ``parallel_chunks`` (1 = off outside any such block; otherwise the
-            # requested total chunk cap). It only affects *planning*, done
-            # here up front, so a later resume — which re-issues the
-            # already-planned chunks — needs no snapshot.
-            plan = ChunkPlan(
-                args, build_request, limit, max_chunks=_parallel_chunks.get()
-            )
-            retry_policy = RetryPolicy.from_env()
-            # The concurrency cap is resolved inside ``resume()`` from
-            # ``API_USGS_CONCURRENT``; ``1`` is a sequential gather,
-            # ``total <= 1`` a one-element gather — no special branch.
-            return ChunkedCall(
-                plan,
-                fetch,
-                retry_policy,
-                finalize,
-                canonical_url=plan.canonical_url,
-                # The collection name, for the progress line the executor
-                # opens. ``get_ogc_data`` puts it in ``args``.
-                service=args.get("collection"),
-            ).resume()
-
-        return wrapper
-
-    return decorator