diff --git a/.github/workflows/code-health.yml b/.github/workflows/code-health.yml index c04b9b59..601ed47c 100644 --- a/.github/workflows/code-health.yml +++ b/.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 '
Full pyscn scores (advisory)' + 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 '
' + 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 diff --git a/dataretrieval/ngwmn.py b/dataretrieval/ngwmn.py index 154899fa..9b51ba3e 100644 --- a/dataretrieval/ngwmn.py +++ b/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, ) diff --git a/dataretrieval/ogc/__init__.py b/dataretrieval/ogc/__init__.py index 4a091e32..67cdac1d 100644 --- a/dataretrieval/ogc/__init__.py +++ b/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", diff --git a/dataretrieval/ogc/chunking.py b/dataretrieval/ogc/chunking.py index c958b49d..8e57d380 100644 --- a/dataretrieval/ogc/chunking.py +++ b/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 diff --git a/dataretrieval/ogc/engine.py b/dataretrieval/ogc/engine.py index 6b9e4a3c..04cecec9 100644 --- a/dataretrieval/ogc/engine.py +++ b/dataretrieval/ogc/engine.py @@ -35,12 +35,12 @@ import httpx import pandas as pd -import dataretrieval.ogc.chunking as chunking +from dataretrieval._ambient import Ambient from dataretrieval.ogc.context import _dialect, _ogc_base_url, _row_cap from dataretrieval.ogc.errors import _raise_for_non_200 +from dataretrieval.ogc.planning import ChunkPlan from dataretrieval.ogc.policy import ( - DEFAULT_DIALECT, - OgcDialect, + OgcApi, _require_positive_int, ) @@ -52,7 +52,13 @@ _switch_properties_id, ) from dataretrieval.ogc.shaping import GEOPANDAS, _finalize_ogc, _get_resp_data -from dataretrieval.transport.fanout import FanOut, active_client +from dataretrieval.transport.fanout import ( + FanOut, + _Fetch, + _Finalize, + _passthrough_result, + active_client, +) from dataretrieval.transport.links import resolve_next_url from dataretrieval.transport.pagination import paginate from dataretrieval.transport.retry import RetryPolicy @@ -63,8 +69,14 @@ # Set up logger for this module logger = logging.getLogger(__name__) -# Compatibility alias: the old name used internally and in tests. -_DEFAULT_DIALECT = DEFAULT_DIALECT + +# Empirically the API replies HTTP 414 above ~8200 bytes of full URL. Resolve +# this default at call time so tests and alternate deployments can patch it. +_OGC_URL_BYTE_LIMIT = 8000 + +# The public ``parallel_chunks`` context manager lives in ``ogc.chunking``; +# execution reads the shared ambient here without depending on that facade. +_parallel_chunks: Ambient[int] = Ambient("ogc_parallel_chunks", 1) def _next_req_url( @@ -217,12 +229,10 @@ async def follow_up(cursor: str, sess: httpx.AsyncClient) -> httpx.Response: def get_ogc_data( args: dict[str, Any], collection: str, - output_id: str, + output_id: str | None = None, *, - base_url: str, + api: OgcApi, max_rows: int | None = None, - extra_id_cols: frozenset[str] | set[str] = frozenset(), - dialect: OgcDialect | None = None, ) -> tuple[pd.DataFrame, BaseMetadata]: """ Retrieves OGC (Open Geospatial Consortium) data as a DataFrame with metadata. @@ -238,30 +248,22 @@ def get_ogc_data( collection : str The OGC API collection name (e.g., ``"daily"``, ``"monitoring-locations"``, ``"continuous"``). - output_id : str - The user-facing id column the wire ``id`` is renamed to. Required — - the per-API collection-to-id map lives in the caller, not here. + output_id : str, optional + The user-facing id column the wire ``id`` is renamed to. Defaults to + whatever ``api`` registers for this collection; pass one only for a + collection the API does not register, such as a reference table whose + ids follow a rule rather than a list. + api : OgcApi + Which OGC API to target: its base URL, its dialect, the synthetic id + columns its results carry, and the id it renames the wire ``id`` to. + Required, because this package is API-neutral and names no API of its + own -- each adapter declares its own (``waterdata.utils.WATERDATA_API``, + ``ngwmn.NGWMN_API``) and passes that. max_rows : int, optional Stop paginating once this many rows have been collected and truncate the result to exactly ``max_rows``. ``None`` (default) fetches the full result. Intended for cheap previews of large, un-chunked tables (e.g. :func:`get_reference_table`). - base_url : str - OGC API base URL to target. Required: this package is API-neutral and - names no API of its own, so each adapter passes its own base (e.g. - ``waterdata.utils.OGC_API_URL``, ``ngwmn.NGWMN_OGC_API_URL``). It was - once optional, falling back to whatever was in ambient scope -- which - defaults to the empty string, so omitting it built a *relative* - ``/collections/{id}/items`` that planning accepted and only httpx - rejected at send time, surfacing as a NetworkError about an unknown - service. Requiring it moves that mistake to the call site, where mypy - catches it. - extra_id_cols : set or frozenset, optional - Synthetic id columns to push to the end of a result frame (see - :func:`_arrange_cols`). Defaults to an empty set. - dialect : OgcDialect, optional - Per-API request quirks (CQL2-only collections, date-only collections). - Defaults to a plain OGC API with neither. Returns ------- @@ -283,8 +285,8 @@ def get_ogc_data( if max_rows is not None: _require_positive_int(max_rows, "max_rows") - if dialect is None: - dialect = _DEFAULT_DIALECT + if output_id is None: + output_id = api.output_id(collection) args = args.copy() args["collection"] = collection args = _switch_arg_id(args, id_name=output_id, collection=collection) @@ -311,23 +313,61 @@ def get_ogc_data( convert_type=convert_type, collection=collection, max_rows=max_rows, - extra_id_cols=extra_id_cols, - dialect=dialect, - base_url=base_url, + api=api, ) # No progress block here: the executor that emits the events owns the line # (see :meth:`~dataretrieval.transport.fanout.FanOut.resume`). - with _row_cap(max_rows), _ogc_base_url(base_url), _dialect(dialect): + with _row_cap(max_rows), _ogc_base_url(api.base_url), _dialect(api.dialect): return _fetch_once(args, finalize=finalize) -@chunking.multi_value_chunked(build_request=_construct_api_requests) +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 plan and execute chunks. + + OGC-specific planning is composed here with the API-neutral + :class:`~dataretrieval.transport.fanout.FanOut` executor. The public + :func:`dataretrieval.ogc.chunking.parallel_chunks` context manager controls + optional finer fan-out through the shared ``_parallel_chunks`` ambient. + """ + + 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 + plan = ChunkPlan( + args, build_request, limit, max_chunks=_parallel_chunks.get() + ) + return FanOut( + plan, + fetch, + RetryPolicy.from_env(), + finalize, + canonical_url=plan.canonical_url, + service=args.get("collection"), + ).resume() + + return wrapper + + return decorator + + +@multi_value_chunked(build_request=_construct_api_requests) async def _fetch_once( args: dict[str, Any], ) -> tuple[pd.DataFrame, httpx.Response]: """Send one prepared-args OGC request asynchronously; return (frame, response). - ``@chunking.multi_value_chunked`` models every multi-value list + ``@multi_value_chunked`` models every multi-value list parameter and the cql-text filter as a chunkable axis, greedy-halves the biggest chunk across all axes until each chunk URL fits, and iterates the cartesian product. With no chunkable inputs the diff --git a/dataretrieval/ogc/filters.py b/dataretrieval/ogc/filters.py index 50875409..a476d9ac 100644 --- a/dataretrieval/ogc/filters.py +++ b/dataretrieval/ogc/filters.py @@ -4,7 +4,7 @@ - ``FILTER_LANG``: the type alias used for the ``filter_lang`` kwarg. -Internal helpers used by ``chunking.multi_value_chunked``'s joint +Internal helpers used by ``engine.multi_value_chunked``'s joint planner: ``_split_top_level_or`` (clause partitioning), ``_is_chunkable`` (filter-language gate), and ``_check_numeric_filter_pitfall`` (the lexicographic-comparison guard). diff --git a/dataretrieval/ogc/policy.py b/dataretrieval/ogc/policy.py index 99498ec1..3735b1f1 100644 --- a/dataretrieval/ogc/policy.py +++ b/dataretrieval/ogc/policy.py @@ -16,6 +16,7 @@ from __future__ import annotations import numbers +from collections.abc import Mapping from dataclasses import dataclass, field @@ -73,3 +74,81 @@ class OgcDialect: # Default dialect: a plain OGC API with no CQL2-only collections and no # date-only collections (every time argument rendered as a full UTC datetime). DEFAULT_DIALECT = OgcDialect() + + +@dataclass(frozen=True) +class OgcApi: + """One OGC API's identity: where it lives and how it answers. + + These three facts always travel together and are constant for a service, + so they are one value rather than three arguments threaded through the + engine and the shaper. An adapter declares its API once + (``waterdata.utils.WATERDATA_API``, ``ngwmn.NGWMN_API``) and passes that. + + Adding a third OGC service is then one object, not three arguments to + thread through two layers and remember to keep consistent -- the mistake + the previous shape invited, and made: ``get_cql`` passed the dialect and + the id columns but let the base URL fall back to a default that happened + to be right. + + Attributes + ---------- + base_url : str + Root the collections hang off, e.g. ``.../ogcapi/v0``. + dialect : OgcDialect + Per-API quirks the request builder and shaper need. + extra_id_cols : frozenset[str] + Synthetic id columns this API's results carry, ordered to the front. + """ + + base_url: str + dialect: OgcDialect = DEFAULT_DIALECT + extra_id_cols: frozenset[str] = frozenset() + #: How the wire ``id`` is renamed for the caller: a per-collection mapping, + #: or a single name when the API applies one to every collection. Excluded + #: from hashing because a mapping is unhashable. + output_ids: Mapping[str, str] | str = field(default="id", hash=False) + + def output_id(self, collection: str) -> str: + """The user-facing name the wire ``id`` takes for ``collection``. + + The caller should not have to know this -- it is a fact about the API, + so the API answers it. Callers with a collection outside the mapping + (a reference table, whose ids follow a rule rather than a list) pass + one explicitly instead. + + Raises + ------ + KeyError + If this API registers ids per collection and does not know this + one. Deliberately loud: a silent fallback would rename the id + column to something no Water Data collection uses, mis-shaping the + result instead of reporting the unregistered collection. + """ + if isinstance(self.output_ids, str): + return self.output_ids + try: + return self.output_ids[collection] + except KeyError: + raise KeyError( + f"{collection!r} has no output id registered for this API. " + f"Known collections: {sorted(self.output_ids)}." + ) from None + + def knows(self, collection: str) -> bool: + """Whether this API recognizes ``collection``. + + Ask this rather than testing ``collection in api.output_ids``: that + works only for the mapping form, and silently becomes a substring test + when an API names one id for every collection. + """ + if isinstance(self.output_ids, str): + return True + return collection in self.output_ids + + @property + def collections(self) -> tuple[str, ...]: + """Registered collection names, or empty when ids are not per-collection.""" + if isinstance(self.output_ids, str): + return () + return tuple(sorted(self.output_ids)) diff --git a/dataretrieval/ogc/shaping.py b/dataretrieval/ogc/shaping.py index e9d26b32..48bfd907 100644 --- a/dataretrieval/ogc/shaping.py +++ b/dataretrieval/ogc/shaping.py @@ -12,14 +12,16 @@ import logging import re -from typing import Any +from typing import TYPE_CHECKING, Any import httpx import pandas as pd +if TYPE_CHECKING: + from dataretrieval.ogc.policy import OgcApi, OgcDialect + from dataretrieval._response_metadata import BaseMetadata from dataretrieval.ogc.context import _ogc_base_url -from dataretrieval.ogc.policy import DEFAULT_DIALECT, OgcDialect try: import geopandas as gpd @@ -364,10 +366,8 @@ def _finalize_ogc( output_id: str, convert_type: bool, collection: str, + api: OgcApi, max_rows: int | None = None, - extra_id_cols: frozenset[str] | set[str] = frozenset(), - dialect: OgcDialect | None = None, - base_url: str | None = None, ) -> tuple[pd.DataFrame, BaseMetadata]: """Shape a combined OGC result into the user-facing ``(df, md)``. @@ -387,14 +387,10 @@ def _finalize_ogc( rather than only per-chunk, so a chunked call's total is bounded to exactly ``max_rows`` and a resumed call honors the cap too. The per-``_paginate`` ``_row_cap`` is only an early-stop download bound. - ``base_url`` is captured with the finalizer so resumed calls query the same + ``api`` is captured with the finalizer so resumed calls query the same API's schema when their combined result is empty. """ - if dialect is None: - dialect = DEFAULT_DIALECT - if base_url is None: - base_url = _ogc_base_url.get() - frame = _deal_with_empty(frame, properties, collection, base_url=base_url) + frame = _deal_with_empty(frame, properties, collection, base_url=api.base_url) # Normalize to PEP-8 snake_case column names *first*, so the dialect's # ``time_cols``/``numerical_cols``/``sort_cols`` (all snake_case) match # regardless of whether the API returns snake_case (Water Data, where @@ -408,9 +404,9 @@ def _finalize_ogc( if renames: frame = frame.rename(columns=renames) if convert_type: - frame = _type_cols(frame, dialect) - frame = _arrange_cols(frame, properties, output_id, extra_id_cols) - frame = _sort_rows(frame, dialect) + frame = _type_cols(frame, api.dialect) + frame = _arrange_cols(frame, properties, output_id, api.extra_id_cols) + frame = _sort_rows(frame, api.dialect) if max_rows is not None: frame = frame.head(max_rows) return frame, BaseMetadata(response) diff --git a/dataretrieval/waterdata/cql.py b/dataretrieval/waterdata/cql.py index 5139b071..7c0f6863 100644 --- a/dataretrieval/waterdata/cql.py +++ b/dataretrieval/waterdata/cql.py @@ -23,10 +23,8 @@ ) from dataretrieval.ogc.shaping import _finalize_ogc from dataretrieval.waterdata.utils import ( - _EXTRA_ID_COLS, - _OUTPUT_ID_BY_COLLECTION, OGC_API_URL, - WATERDATA_DIALECT, + WATERDATA_API, _accept_legacy_kwargs, ) @@ -145,12 +143,12 @@ def get_cql( ... ' "02070010%"]}', ... ) """ - if collection not in _OUTPUT_ID_BY_COLLECTION: + if not WATERDATA_API.knows(collection): raise ValueError( f"Unknown collection {collection!r}. Valid collections: " - f"{sorted(_OUTPUT_ID_BY_COLLECTION)}." + f"{list(WATERDATA_API.collections)}." ) - output_id = _OUTPUT_ID_BY_COLLECTION[collection] + output_id = WATERDATA_API.output_id(collection) # ``dict`` is the pythonic input — serialize on the way out. ``str`` is sent # verbatim so callers who already have a CQL2 doc (e.g. imported from a @@ -186,8 +184,7 @@ def get_cql( output_id=output_id, convert_type=convert_type, collection=collection, - extra_id_cols=_EXTRA_ID_COLS, - dialect=WATERDATA_DIALECT, + api=WATERDATA_API, ) diff --git a/dataretrieval/waterdata/metadata.py b/dataretrieval/waterdata/metadata.py index 6ab681e2..880bafa8 100644 --- a/dataretrieval/waterdata/metadata.py +++ b/dataretrieval/waterdata/metadata.py @@ -336,9 +336,7 @@ def get_monitoring_locations( # Build argument dictionary, omitting None values (resolving the unified # `state` argument into the OGC `state_name` queryable). - args = _get_args( - _with_state(locals(), to="name", into="state_name"), exclude={"max_rows"} - ) + args = _get_args(_with_state(locals(), collection), exclude={"max_rows"}) return get_ogc_data(args, collection, max_rows=max_rows) @@ -587,9 +585,7 @@ def get_time_series_metadata( # Build argument dictionary, omitting None values (resolving the unified # `state` argument into the OGC `state_name` queryable). - args = _get_args( - _with_state(locals(), to="name", into="state_name"), exclude={"max_rows"} - ) + args = _get_args(_with_state(locals(), collection), exclude={"max_rows"}) return get_ogc_data(args, collection, max_rows=max_rows) @@ -845,9 +841,7 @@ def get_combined_metadata( collection = "combined-metadata" # Resolve the unified `state` argument into the OGC `state_name` queryable. - args = _get_args( - _with_state(locals(), to="name", into="state_name"), exclude={"max_rows"} - ) + args = _get_args(_with_state(locals(), collection), exclude={"max_rows"}) return get_ogc_data(args, collection, max_rows=max_rows) diff --git a/dataretrieval/waterdata/time_series.py b/dataretrieval/waterdata/time_series.py index f3c7b572..99b179d4 100644 --- a/dataretrieval/waterdata/time_series.py +++ b/dataretrieval/waterdata/time_series.py @@ -1006,14 +1006,16 @@ def get_stats_por( ... end_date="01-31", ... ) """ + resource = "observationNormals" + # Build argument dictionary, omitting None values params = _get_args( - _with_state(locals(), to="fips_us", into="state_code"), + _with_state(locals(), resource), exclude={"expand_percentiles"}, ) return stats.get_data( - args=params, service="observationNormals", expand_percentiles=expand_percentiles + args=params, service=resource, expand_percentiles=expand_percentiles ) @@ -1152,15 +1154,17 @@ def get_stats_date_range( ... computation_type=["minimum", "maximum"], ... ) """ + resource = "observationIntervals" + # Build argument dictionary, omitting None values params = _get_args( - _with_state(locals(), to="fips_us", into="state_code"), + _with_state(locals(), resource), exclude={"expand_percentiles"}, ) return stats.get_data( args=params, - service="observationIntervals", + service=resource, expand_percentiles=expand_percentiles, ) diff --git a/dataretrieval/waterdata/utils.py b/dataretrieval/waterdata/utils.py index c24624d0..fc3f7f69 100644 --- a/dataretrieval/waterdata/utils.py +++ b/dataretrieval/waterdata/utils.py @@ -24,7 +24,7 @@ import pandas as pd from dataretrieval.codes.states import apply_state -from dataretrieval.ogc import OgcDialect, prepare_request_args +from dataretrieval.ogc import OgcApi, OgcDialect, prepare_request_args from dataretrieval.ogc import get_ogc_data as _facade_get_ogc_data # Endpoint constants live in one place for the whole collection; they are re-bound @@ -96,6 +96,16 @@ sort_cols=("time", "monitoring_location_id"), ) +#: This package's view of the Water Data OGC API: where it lives, how it +#: answers, and the synthetic id columns its results carry. Declared once so +#: the engine and the shaper take one value instead of three arguments. +WATERDATA_API = OgcApi( + base_url=OGC_API_URL, + dialect=WATERDATA_DIALECT, + extra_id_cols=_EXTRA_ID_COLS, + output_ids=_OUTPUT_ID_BY_COLLECTION, +) + # The Water-Data-specific *extras* on top of the engine's own no-normalize set # (which already covers the date-range params and ``bbox``). Scalar non-string # knobs are caught by runtime type, so only iterables with special handling @@ -150,17 +160,37 @@ def _get_args( ) -def _with_state(local_vars: dict[str, Any], *, to: str, into: str) -> dict[str, Any]: - """Resolve the unified ``state`` argument into an endpoint's state queryable. +#: Which state queryable each endpoint filters on, and in which +#: representation. A fact about the endpoint, so it lives with the endpoint +#: rather than being repeated at every getter that filters by state -- the +#: shape ``ngwmn`` already uses for the same problem. +#: +#: Keyed by OGC collection *and* by Statistics API resource, because both +#: accept a state filter and they disagree about it: the OGC metadata +#: collections take a full state name, Statistics takes a FIPS code. Those are +#: different kinds of thing (see ``CONTEXT.md``) and only share this table +#: because the question is the same one. +_STATE_QUERYABLE: dict[str, dict[str, str]] = { + "monitoring-locations": {"to": "name", "into": "state_name"}, + "time-series-metadata": {"to": "name", "into": "state_name"}, + "combined-metadata": {"to": "name", "into": "state_name"}, + "observationNormals": {"to": "fips_us", "into": "state_code"}, + "observationIntervals": {"to": "fips_us", "into": "state_code"}, +} + + +def _with_state(local_vars: dict[str, Any], endpoint: str) -> dict[str, Any]: + """Resolve the unified ``state`` argument into an endpoint's queryable. Returns the (mutated) args mapping. ``state`` is the canonical, format-flexible parameter (full name / postal / FIPS); it is normalized via - :func:`~dataretrieval.codes.states.to_state` to the ``to`` representation - and stored under ``into`` (the queryable this endpoint actually filters on). + :func:`~dataretrieval.codes.states.to_state` into whichever representation + and queryable ``endpoint`` filters on -- see :data:`_STATE_QUERYABLE`. It is additive sugar over the native ``state_code`` / ``state_name`` parameters, which still accept the API's raw values (e.g. non-US FIPS); passing ``state`` together with either raises ``ValueError``. """ + queryable = _STATE_QUERYABLE[endpoint] # Flatten ``**queryables`` first so a native state param arriving that way # (e.g. ``get_time_series_metadata``'s ``state_code``, which isn't an # explicit parameter) is visible to apply_state's mutual-exclusion guard. @@ -168,7 +198,10 @@ def _with_state(local_vars: dict[str, Any], *, to: str, into: str) -> dict[str, # check and silently send both. _flatten_queryables(local_vars) return apply_state( - local_vars, to=to, into=into, reject=("state_code", "state_name") + local_vars, + to=queryable["to"], + into=queryable["into"], + reject=("state_code", "state_name"), ) @@ -180,11 +213,13 @@ def get_ogc_data( ) -> tuple[pd.DataFrame, BaseMetadata]: """Water-Data wrapper over :func:`~dataretrieval.ogc.get_ogc_data`. - Defaults ``output_id`` from the Water Data collection map when not given, - and supplies the Water Data extra-id columns and dialect, so the typed - getters in ``api.py`` call this unchanged. (Sibling OGC APIs such as - NGWMN call ``dataretrieval.ogc.get_ogc_data`` directly with their own - base URL and dialect rather than going through this Water Data wrapper.) + Supplies :data:`WATERDATA_API`, so the typed getters in ``api.py`` call + this unchanged. ``output_id`` is resolved by that value and only needs + passing for a collection outside its map (a reference table, say). + + Sibling OGC APIs such as NGWMN call ``dataretrieval.ogc.get_ogc_data`` + directly with their own :class:`~dataretrieval.ogc.OgcApi` rather than + going through this Water Data wrapper. Parameters ---------- @@ -209,16 +244,12 @@ def get_ogc_data( A metadata object with request information, including the URL and query time. """ - if output_id is None: - output_id = _OUTPUT_ID_BY_COLLECTION[collection] return _facade_get_ogc_data( args, collection, output_id, max_rows=max_rows, - base_url=OGC_API_URL, - extra_id_cols=_EXTRA_ID_COLS, - dialect=WATERDATA_DIALECT, + api=WATERDATA_API, ) @@ -291,6 +322,7 @@ def wrapper(*args: Any, **kwargs: Any) -> _R: "BASE_URL", "OGC_API_URL", "SAMPLES_URL", + "WATERDATA_API", "WATERDATA_DIALECT", "_EXTRA_ID_COLS", "_NO_NORMALIZE_PARAMS", diff --git a/docs/source/architecture/health-iterations.rst b/docs/source/architecture/health-iterations.rst new file mode 100644 index 00000000..34d3c4bd --- /dev/null +++ b/docs/source/architecture/health-iterations.rst @@ -0,0 +1,104 @@ +Architecture health iteration log +================================= + +Purpose +------- + +This log records small architecture experiments evaluated with ``pyscn +1.29.0``. It keeps rejected ideas available for future work without leaving +non-improving code in the branch. PySCN is treated as an architectural fitness +function, not as a substitute for design review: public compatibility, +dependency direction, and the package's function-oriented API remain higher +priority than optimizing a composite metric. + +Method +------ + +Each iteration starts from the latest accepted commit, runs ``pyscn analyze +--json --no-open dataretrieval``, and compares the rounded health score with the +accepted score. A code change is retained only when the target health score +increases and the test, type, lint, and dependency-contract checks pass. +Experiments that improve an unrounded or subsidiary signal but leave the target +score unchanged are reverted and recorded below. Iteration stops at 95 or after +three consecutive experiments leave the accepted score unchanged. + +Results +------- + +.. list-table:: PySCN architecture iterations + :header-rows: 1 + :widths: 8 20 12 12 48 + + * - Iteration + - Experiment + - Health score + - Decision + - Evidence and rationale + * - Baseline + - Existing ``ci/health-report-signals`` branch + - 82 + - Starting point + - Complexity 95, Dead Code 100, Duplication 70, Coupling 100, + Cohesion 100, Dependencies 80, Architecture 87; maximum dependency + depth 8 with 139 internal edges. + * - 1 + - Remove ``ogc.engine -> ogc.chunking`` from the execution path + - 83 + - **Accepted** (``0172b0af``) + - The chunk planner/executor composition moved to the OGC engine while + ``ogc.chunking`` retained the public ``parallel_chunks`` dial and + compatibility imports. Maximum depth fell from 8 to 7, dependency score + rose from 80 to 85, and edges fell from 139 to 138. The full test suite, + Ruff, mypy, and all import-linter contracts passed. + * - 2 + - Inline private ``transport.env`` into ``transport.retry`` + - 83 + - Rejected + - Modules fell from 58 to 57 and edges from 138 to 135, but every rounded + PySCN sub-score and the composite were unchanged. This remains a + reasonable cleanup if module count or edge count becomes a separately + adopted fitness function. + * - 3 + - Extract ``FanOut._run`` failure precedence + - 83 + - Rejected + - The high-risk-function count fell from 20 to 19 and 175 focused tests + passed, but Complexity remained 95 and the composite remained 83. Revisit + if the project adopts the raw high-risk count as a ratchet. + * - 4 + - Remove historical transport aliases from ``ogc.chunking`` + - 83 + - Rejected + - Architecture rose from 87 to 88 and edges fell from 138 to 137, but the + target composite did not move. The experiment also removed deliberately + retained compatibility names for a marginal metric gain, conflicting + with the project's higher-priority compatibility characteristic. + +Convergence +----------- + +The accepted score is **83**. Iterations 2, 3, and 4 each returned 83, so the +three-straight-iterations convergence condition was reached. Further attempts +were stopped rather than collapsing typed public getter families or widely +used policy leaves merely to satisfy tool heuristics. + +The remaining 70 Duplication score is dominated by intentional symmetry: +collection-specific Water Data and NGWMN getters preserve explicit typed +signatures and documentation, WQP wrappers preserve discoverable service +entry points, and deprecated NWIS signatures are frozen. The Architecture +score also treats dependency-free leaves and service-neutral orchestrators as +single-responsibility violations because they are used across several module +communities. Those findings remain useful review prompts, but changing those +boundaries solely for the score would weaken the documented architecture. + +Future experiments +------------------ + +- Reconsider the ``transport.env`` consolidation if dependency edge count is + promoted from an informational signal to an explicit fitness function. +- Reconsider the ``FanOut._run`` extraction if raw high-risk findings, rather + than the rounded Complexity score, become a ratchet. +- Remove ``ogc.chunking`` compatibility aliases only through an intentional + compatibility decision, not as an incidental score optimization. +- Prefer changes that reduce maximum dependency depth, introduce no new clone + groups, preserve typed public APIs, and keep all import-linter contracts. diff --git a/docs/source/architecture/index.rst b/docs/source/architecture/index.rst index a92f19c4..31190732 100644 --- a/docs/source/architecture/index.rst +++ b/docs/source/architecture/index.rst @@ -105,12 +105,13 @@ Shared components state; ``requests`` owns argument normalization and HTTP request construction; ``schema`` executes queryables/schema requests; ``engine`` supplies OGC cursor and response strategies to transport pagination; - ``planning`` determines chunk boundaries; ``chunking`` connects those plans - to the shared fan-out executor and retains compatibility aliases; - ``interruptions`` and ``retry`` re-export their moved compatibility - surfaces; and ``shaping``, ``dates``, ``filters``, and ``errors`` isolate - their named protocol concerns. The full runtime OGC graph, including the - facade, is acyclic -- enforced by the package-wide fitness function in + ``planning`` determines chunk boundaries; ``engine`` composes those plans + with the shared fan-out executor; ``chunking`` owns the public + ``parallel_chunks`` dial and retains compatibility aliases; ``interruptions`` + and ``retry`` re-export their moved compatibility surfaces; and ``shaping``, + ``dates``, ``filters``, and ``errors`` isolate their named protocol + concerns. The full runtime OGC graph, including the facade, is acyclic -- + enforced by the package-wide fitness function in ``tests/architecture_test.py``. ``dataretrieval.transport`` @@ -343,4 +344,5 @@ Architecturally significant changes should: .. toctree:: :maxdepth: 1 + health-iterations decisions/index diff --git a/tests/waterdata_chunking_test.py b/tests/waterdata_chunking_test.py index 19687869..cf20699f 100644 --- a/tests/waterdata_chunking_test.py +++ b/tests/waterdata_chunking_test.py @@ -45,7 +45,6 @@ TransientError, Unchunkable, ) -from dataretrieval.ogc import chunking as _chunking from dataretrieval.ogc import engine as _engine from dataretrieval.ogc.chunking import ( ChunkedCall, @@ -367,7 +366,7 @@ async def fetch(args): def test_multi_value_chunked_lazy_url_limit(monkeypatch): - """``url_limit=None`` → resolve chunking._OGC_URL_BYTE_LIMIT at call + """``url_limit=None`` → resolve engine._OGC_URL_BYTE_LIMIT at call time, so tests that patch the constant affect this decorator too.""" calls = [] @@ -378,7 +377,7 @@ async def fetch(args): elapsed=datetime.timedelta(seconds=0.1), headers={} ) - monkeypatch.setattr(_chunking, "_OGC_URL_BYTE_LIMIT", 240) + monkeypatch.setattr(_engine, "_OGC_URL_BYTE_LIMIT", 240) # 4 sites of 10 chars → exceeds 240 → planner splits. fetch({"sites": ["S" * 10 + str(i) for i in range(4)]}) assert len(calls) > 1, "patched constant should drive chunking" diff --git a/tests/waterdata_utils_test.py b/tests/waterdata_utils_test.py index e805ec92..a1cc2fc5 100644 --- a/tests/waterdata_utils_test.py +++ b/tests/waterdata_utils_test.py @@ -41,17 +41,14 @@ from dataretrieval.waterdata import get_stats_date_range, get_stats_por from dataretrieval.waterdata.stats import _handle_nesting, get_data from dataretrieval.waterdata.utils import ( - _EXTRA_ID_COLS, OGC_API_URL, - WATERDATA_DIALECT, + WATERDATA_API, _get_args, ) # The Water Data injection ``get_cql`` performs at its call site, so these tests # exercise the same result shape the typed getters produce. -_finalize_ogc = functools.partial( - _ogc_finalize, extra_id_cols=_EXTRA_ID_COLS, dialect=WATERDATA_DIALECT -) +_finalize_ogc = functools.partial(_ogc_finalize, api=WATERDATA_API) _LOGGER_NAME = _utils_module.__name__ @@ -1070,19 +1067,19 @@ def test_with_state_routes_into_native_queryable(): """``_with_state`` resolves the canonical ``state`` argument into the endpoint's native queryable (any encoding -> the requested representation) and leaves args without ``state`` untouched.""" - assert _utils_module._with_state({"state": "WI"}, to="name", into="state_name") == { + assert _utils_module._with_state({"state": "WI"}, "monitoring-locations") == { "state_name": "Wisconsin" } - assert _utils_module._with_state( - {"state": "Wisconsin"}, to="fips_us", into="state_code" - ) == {"state_code": "US:55"} + assert _utils_module._with_state({"state": "Wisconsin"}, "observationNormals") == { + "state_code": "US:55" + } # Multi-value state fans out element-wise. assert _utils_module._with_state( - {"state": ["WI", "55"]}, to="name", into="state_name" + {"state": ["WI", "55"]}, "monitoring-locations" ) == {"state_name": ["Wisconsin", "Wisconsin"]} # No ``state`` -> mapping returned unchanged. assert _utils_module._with_state( - {"state_name": "Ohio"}, to="name", into="state_name" + {"state_name": "Ohio"}, "monitoring-locations" ) == {"state_name": "Ohio"} @@ -1091,11 +1088,11 @@ def test_with_state_conflict_raises(): is ambiguous and raises.""" with pytest.raises(ValueError, match="not both"): _utils_module._with_state( - {"state": "WI", "state_code": "55"}, to="name", into="state_name" + {"state": "WI", "state_code": "55"}, "monitoring-locations" ) with pytest.raises(ValueError, match="not both"): _utils_module._with_state( - {"state": "WI", "state_name": "Wisconsin"}, to="name", into="state_name" + {"state": "WI", "state_name": "Wisconsin"}, "monitoring-locations" ) @@ -1107,8 +1104,7 @@ def test_with_state_conflict_via_queryables_raises(): with pytest.raises(ValueError, match="not both"): _utils_module._with_state( {"state": "WI", "queryables": {"state_code": "55"}}, - to="name", - into="state_name", + "monitoring-locations", ) diff --git a/tools/health_signals.py b/tools/health_signals.py new file mode 100644 index 00000000..360a61c3 --- /dev/null +++ b/tools/health_signals.py @@ -0,0 +1,143 @@ +"""Report the health signals that mean something for this package. + +pyscn grades six sub-scores; four of them do not apply here. Complexity counts +synthetic per-file rows, CBO and LCOM stay pinned at 100 in a function-oriented +package, and Architecture penalises a leaf for being depended upon -- extracting +BaseMetadata to a dependency-free leaf lowered it. The composite averages all +six, so it moves for reasons that are not about this codebase. + +What is left is genuinely useful, and is what this prints: dead code, a NEW +clone group appearing, and the dependency depth. Plus the interface cost of the +public getters, which nothing else measures and which is the closest automatable +proxy for whether a module is deep. +""" + +import ast +import glob +import json +import os +import sys +from pathlib import Path + +BASELINE = Path(".pyscn-known-clones.json") +PACKAGE = Path("dataretrieval") + + +_TREES: dict[str, ast.Module] = {} +_SPANS: dict[str, dict[int, str]] = {} + + +def tree_for(path): + """Parse each file once; both walks below want the same trees.""" + if path not in _TREES: + _TREES[path] = ast.parse(Path(path).read_text(encoding="utf-8")) + return _TREES[path] + + +def enclosing(path, start): + if path not in _SPANS: + _SPANS[path] = { + n.lineno: n.name + for n in ast.walk(tree_for(path)) + if isinstance(n, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)) + } + for line in range(start, start + 3): + if line in _SPANS[path]: + return _SPANS[path][line] + return f"" + + +def clone_groups(report): + groups = [] + for g in report["clone"]["clone_groups"]: + groups.append( + frozenset( + f"{c['location']['file_path']}::" + f"{enclosing(c['location']['file_path'], c['location']['start_line'])}" + for c in g["clones"] + ) + ) + return groups + + +def required_arguments(): + """Public functions carrying required arguments, worst module first.""" + out = [] + for path in sorted(PACKAGE.rglob("*.py")): + tree = tree_for(str(path)) + declared = None + for node in tree.body: + if isinstance(node, ast.Assign) and any( + isinstance(t, ast.Name) and t.id == "__all__" for t in node.targets + ): + declared = set(ast.literal_eval(node.value)) + required = 0 + for node in tree.body: + if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): + continue + public = ( + node.name in declared + if declared is not None + else not node.name.startswith("_") + ) + if not public: + continue + positional = node.args.posonlyargs + node.args.args + required += max(0, len(positional) - len(node.args.defaults)) + if required: + out.append((str(path), required)) + return sorted(out, key=lambda r: -r[1]) + + +reports = sorted(glob.glob(".pyscn/reports/*.json")) +if not reports: + print("no pyscn JSON report found") + sys.exit(0) +report = json.loads(Path(reports[-1]).read_text()) + +# pyscn infers a project root, and when it guesses wrong it resolves only a +# fraction of the imports -- which *raises* its scores, because most of what it +# grades is dependency-derived. A degraded run therefore looks like an improved +# one. Report the root so that is visible rather than flattering. +summary = report["system"]["Summary"] +root = summary["ProjectRoot"] +if os.path.realpath(root) != os.path.realpath(os.getcwd()): + print( + f"WARNING: pyscn resolved against {root!r}, not this checkout -- import\n" + f"resolution is probably degraded and these numbers are not comparable." + ) +print(f"modules {summary['TotalModules']}") + +dead = report["dead_code"]["summary"]["total_findings"] +deps = report["system"]["DependencyAnalysis"] +print(f"dead code {dead} findings") +print( + f"dependency depth {deps['MaxDepth']} (max path), " + f"{deps['TotalDependencies']} edges" +) + +current = clone_groups(report) +if BASELINE.exists(): + baseline = json.loads(BASELINE.read_text())["groups"] + known = {frozenset(g["members"]) for g in baseline} + new = [g for g in current if g not in known] + gone = [g for g in known if g not in current] + print( + f"clone groups {len(current)} total, " + f"{len(known) - len(gone)} known, {len(new)} NEW" + ) + for g in new: + print(" NEW GROUP -- not in .pyscn-known-clones.json:") + for member in sorted(g): + print(f" {member}") + for g in gone: + print(f" resolved (baseline is stale by one group): {sorted(g)[0]} ...") +else: + print(f"clone groups {len(current)} (no baseline file to compare against)") + +print("\ninterface cost -- public functions with required arguments:") +rows = required_arguments() +if not rows: + print(" none: every public function is fully keyword-optional") +for path, count in rows[:6]: + print(f" {path:<44} {count} required")