diff --git a/NEWS.md b/NEWS.md index 6bbb2412..861d2c5c 100644 --- a/NEWS.md +++ b/NEWS.md @@ -1,3 +1,5 @@ +**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. **06/23/2026:** **Breaking change (1.2.0):** the minimum supported Python is now **3.10** (`requires-python = ">=3.10"`). 3.9 support was already effectively broken — the `waterdata` module's dependencies (`anyio`, the test stack) require 3.10+, and the `waterdata` test modules already skipped on <3.10. `anyio` is now declared as a direct dependency (it is imported directly by `waterdata`), and the CI/ruff/mypy targets move to 3.10. Also fully removed the deprecated `variable_info` metadata property: the `NWIS_Metadata` override only warned and returned `None` (it relied on the defunct `get_pmcodes`), and the `BaseMetadata` abstract is gone too since nothing implemented it — accessing `.variable_info` now raises `AttributeError`. `site_info` is unaffected. diff --git a/dataretrieval/ngwmn.py b/dataretrieval/ngwmn.py index 554310c3..83cb9726 100644 --- a/dataretrieval/ngwmn.py +++ b/dataretrieval/ngwmn.py @@ -3,8 +3,8 @@ The NGWMN exposes its data through a dedicated OGC API (``https://api.waterdata.usgs.gov/ngwmn/ogcapi``) with five collections: ``sites``, ``waterLevelObs``, ``lithologyObs``, ``constructionObs``, and -``providers``. Each getter below delegates to the shared OGC engine -(:func:`~dataretrieval.ogc.engine.get_ogc_data`) with +``providers``. Each getter below delegates to the shared OGC facade +(:func:`~dataretrieval.ogc.get_ogc_data`) with ``base_url=NGWMN_OGC_API_URL``, so multi-value chunking, pagination, retry/resume, and result shaping all behave exactly as they do for the main Water Data getters. @@ -24,9 +24,12 @@ import pandas as pd from dataretrieval.codes.states import apply_state -from dataretrieval.ogc.engine import BASE_URL, OgcDialect, _get_args, get_ogc_data +from dataretrieval.ogc import OgcDialect, get_ogc_data, prepare_request_args from dataretrieval.utils import BaseMetadata +# The Water Data API base URL, defined locally to avoid importing policy internals. +BASE_URL = "https://api.waterdata.usgs.gov" + # The National Ground-Water Monitoring Network exposes its own OGC API at a # separate, unversioned base. NGWMN_OGC_API_URL = f"{BASE_URL}/ngwmn/ogcapi" @@ -72,7 +75,7 @@ def _get(service: str, local_vars: dict[str, Any]) -> tuple[pd.DataFrame, BaseMetadata]: - """Marshal a getter's arguments and dispatch to the shared OGC engine. + """Marshal a getter's arguments and dispatch to the shared OGC facade. Every NGWMN getter ends with this same call; centralizing it keeps the NGWMN base URL, output id, and dialect wired up in exactly one place. @@ -80,7 +83,7 @@ def _get(service: str, local_vars: dict[str, Any]) -> tuple[pd.DataFrame, BaseMe queryable = _STATE_QUERYABLE.get(service) if queryable is not None: apply_state(local_vars, to=queryable["to"], into=queryable["into"]) - args = _get_args(local_vars) + args = prepare_request_args(local_vars) return get_ogc_data( args, service, diff --git a/dataretrieval/ogc/__init__.py b/dataretrieval/ogc/__init__.py index 6e259bb5..46dd3918 100644 --- a/dataretrieval/ogc/__init__.py +++ b/dataretrieval/ogc/__init__.py @@ -1 +1,26 @@ -"""Generic OGC API engine shared by the Water Data and NGWMN getters.""" +"""Generic OGC API engine shared by the Water Data and NGWMN getters. + +The public facade exposes only the minimal service-adapter seam: + +- :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). +- :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. +""" + +from dataretrieval.ogc.engine import fetch_ogc_request, get_ogc_data +from dataretrieval.ogc.policy import OgcDialect +from dataretrieval.ogc.requests import prepare_request_args + +__all__ = [ + "OgcDialect", + "fetch_ogc_request", + "get_ogc_data", + "prepare_request_args", +] diff --git a/dataretrieval/ogc/chunking.py b/dataretrieval/ogc/chunking.py index b46c17df..79037a6b 100644 --- a/dataretrieval/ogc/chunking.py +++ b/dataretrieval/ogc/chunking.py @@ -83,7 +83,7 @@ import pandas as pd from anyio.from_thread import start_blocking_portal -from dataretrieval.utils import HTTPX_DEFAULTS, Ambient, _require_positive_int +from dataretrieval.utils import HTTPX_ASYNC_DEFAULTS, Ambient, _require_positive_int from . import progress as _progress from .combining import ( @@ -600,7 +600,7 @@ async def _run(self, max_concurrent: int | None) -> tuple[pd.DataFrame, Any]: The semaphore, not the pool, is deliberately the throttle. If the pool throttled instead, the excess sub-requests would queue *inside* httpx waiting for a connection, and that wait counts - against the pool-acquire timeout (60 s, from ``HTTPX_DEFAULTS``). + against the pool-acquire timeout (60 s, from ``HTTPX_ASYNC_DEFAULTS``). A batch of slow pages that keeps every connection busy past that window would then trip ``httpx.PoolTimeout`` on the queued tail — a purely client-side failure that consumes the retry budget and @@ -650,7 +650,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_DEFAULTS) as client: + async with httpx.AsyncClient(limits=limits, **HTTPX_ASYNC_DEFAULTS) as client: with _chunked_client(client): reporter = _progress.current() if reporter is not None: diff --git a/dataretrieval/ogc/engine.py b/dataretrieval/ogc/engine.py index 5fda67e5..31b81f5f 100644 --- a/dataretrieval/ogc/engine.py +++ b/dataretrieval/ogc/engine.py @@ -1,11 +1,11 @@ """Generic OGC API engine shared by the Water Data and NGWMN getters. -This module holds the API-agnostic core for talking to an OGC API Features -service — request construction (GET comma-joined or POST/CQL2), async -pagination, and the chunked fetch entry point :func:`get_ogc_data` that -orchestrates them. The surrounding concerns live in sibling modules it -composes, each with its own reason to change: -:mod:`~dataretrieval.ogc.dates` (time-parameter marshalling), +This module holds the API-agnostic orchestration core for talking to an OGC +API Features service — async pagination, the sync bridge, and the chunked +fetch entry point :func:`get_ogc_data` that orchestrates them. Request +construction lives in :mod:`~dataretrieval.ogc.requests`. The surrounding +concerns live in sibling modules it composes, each with its own reason to +change: :mod:`~dataretrieval.ogc.dates` (time-parameter marshalling), :mod:`~dataretrieval.ogc.errors` (HTTP error mapping), and :mod:`~dataretrieval.ogc.shaping` (GeoJSON features to DataFrame and result finalization). It is deliberately free of any Water-Data-specific constants @@ -25,38 +25,56 @@ from __future__ import annotations import functools -import json import logging -import re from collections.abc import ( AsyncIterator, Awaitable, Callable, - Iterable, - Mapping, ) from contextlib import asynccontextmanager -from dataclasses import dataclass, field from typing import Any, TypeVar, cast import httpx import pandas as pd from anyio.from_thread import start_blocking_portal +import dataretrieval.ogc.chunking as chunking +import dataretrieval.ogc.progress as _progress from dataretrieval.exceptions import DataRetrievalError -from dataretrieval.ogc import chunking -from dataretrieval.ogc import progress as _progress from dataretrieval.ogc.chunking import get_active_client from dataretrieval.ogc.combining import _QUOTA_HEADER, _merge_response, _safe_elapsed -from dataretrieval.ogc.dates import _DATE_RANGE_PARAMS, _format_api_dates from dataretrieval.ogc.errors import _paginated_failure_message, _raise_for_non_200 +from dataretrieval.ogc.policy import ( + BASE_URL, # noqa: F401 — compatibility alias + DEFAULT_DIALECT, + OGC_API_URL, + OgcDialect, +) + +# Frozen legacy compatibility surface; tests prevent new request-side re-exports. +from dataretrieval.ogc.requests import ( # noqa: F401 + _NO_NORMALIZE_PARAMS, + _as_str_list, + _check_monitoring_location_id, + _check_ogc_requests, + _construct_api_requests, + _construct_cql_request, + _cql2_param, + _dialect, + _get_args, + _normalize_str_iterable, + _ogc_base_url, + _ogc_query_params, + _row_cap, + _switch_arg_id, + _switch_properties_id, + prepare_request_args, +) from dataretrieval.ogc.shaping import GEOPANDAS, _finalize_ogc, _get_resp_data from dataretrieval.utils import ( - HTTPX_DEFAULTS, - Ambient, + HTTPX_ASYNC_DEFAULTS, BaseMetadata, - _default_headers, - _get, + _default_headers, # noqa: F401 — compatibility re-export for tests _network_error, _require_positive_int, ) @@ -64,415 +82,8 @@ # Set up logger for this module logger = logging.getLogger(__name__) -BASE_URL = "https://api.waterdata.usgs.gov" -OGC_API_VERSION = "v0" -OGC_API_URL = f"{BASE_URL}/ogcapi/{OGC_API_VERSION}" - - -@dataclass(frozen=True) -class OgcDialect: - """Per-API quirks the generic request builder needs to know about. - - Attributes - ---------- - cql2_services : frozenset[str] - Collections that don't accept comma-separated multi-value GET - parameters and so must be queried via POST with a CQL2 JSON body. - date_only_services : frozenset[str] - Collections whose time arguments are rendered date-only - (``YYYY-MM-DD``) rather than as a full UTC datetime. The - ``last_modified`` parameter is always rendered as a full datetime - regardless of this set. - time_cols : frozenset[str] - Result columns to coerce to datetime when ``convert_type`` is set. - Empty by default, so the generic engine carries no API-specific - column knowledge; each API supplies its own. - numerical_cols : frozenset[str] - Result columns to coerce to numeric when ``convert_type`` is set. - sort_cols : tuple[str, ...] - Columns to sort the combined result by, in priority order. Sorting - is applied only when the first (primary) column is present; any - later columns also present are added as secondary keys. - """ - - cql2_services: frozenset[str] = field(default_factory=frozenset) - date_only_services: frozenset[str] = field(default_factory=frozenset) - time_cols: frozenset[str] = field(default_factory=frozenset) - numerical_cols: frozenset[str] = field(default_factory=frozenset) - sort_cols: tuple[str, ...] = field(default_factory=tuple) - - -# 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() - - -def _switch_arg_id(ls: dict[str, Any], id_name: str, service: str) -> dict[str, Any]: - """ - Switch argument id from its package-specific identifier to the standardized "id" key - that the API recognizes. - - If `ls` does not already have an "id" key, sets it from either the - service-derived id key or the expected id column name. If neither key - exists, "id" is left unset. The original service-specific id keys are - removed regardless. - - Parameters - ---------- - ls : Dict[str, Any] - The dictionary containing identifier keys to be standardized. - id_name : str - The name of the specific identifier key to look for. - service : str - The service name. - - Returns - ------- - Dict[str, Any] - The modified dictionary with the "id" key set appropriately. - - Examples - -------- - For service "time-series-metadata", the function will look for either - "time_series_metadata_id" or "time_series_id" and change the key to simply - "id". - """ - - service_id = service.replace("-", "_") + "_id" - - if "id" not in ls: - if service_id in ls: - ls["id"] = ls[service_id] - elif id_name in ls: - ls["id"] = ls[id_name] - - # Remove the original keys regardless of whether they were used - ls.pop(service_id, None) - ls.pop(id_name, None) - - return ls - - -def _switch_properties_id( - properties: list[str] | None, id_name: str, service: str -) -> list[str]: - """ - Build the wire ``properties`` list, dropping every id alias and - ``geometry``. - - The feature ``id`` is always returned and is renamed to the - service-specific id column (e.g. ``daily_id``) in post-processing, so - it must not be requested as a property: several collections (e.g. - ``daily``, ``continuous``) reject ``id`` in ``properties`` with an - HTTP 400. ``geometry`` is likewise excluded because it is controlled - by ``skip_geometry``. Any service-specific id name (``daily_id``, - ``monitoring_location_id``, …) and the bare ``id`` are dropped, and - remaining hyphens are normalized to underscores. Returns an empty - list when `properties` is empty or None — the URL then omits the - ``properties`` filter and the result is shaped by :func:`_arrange_cols`. - - Parameters - ---------- - properties : Optional[List[str]] - A list containing the properties or column names to be pulled from the - service, or None. - id_name : str - The service-specific id column name to drop (e.g. ``daily_id``). - service : str - The service name. - - Returns - ------- - List[str] - The wire ``properties`` with id aliases and ``geometry`` removed - and hyphens normalized. - - Examples - -------- - For service "daily" with ``properties=["daily_id", "value", "geometry"]``, - returns ``["value"]`` — ``daily_id`` and ``geometry`` are dropped, while - the ``daily_id`` column still appears in the result, renamed from the - always-returned feature ``id``. - """ - if not properties: - return [] - service_id = service.replace("-", "_") + "_id" - # The feature ``id`` always comes back (renamed to the service id - # downstream) and several collections reject it as a selectable - # property; ``geometry`` is controlled by ``skip_geometry``. Drop both, - # plus the service-specific id column (``id_name``) and the name derived - # straight from the service (``service_id``). - drop = {"id", "geometry", id_name, service_id} - normalized = (p.replace("-", "_") for p in properties) - return [p for p in normalized if p not in drop] - - -def _cql2_param(args: dict[str, Any]) -> str: - """ - Convert query parameters to CQL2 JSON format for POST requests. - - Parameters - ---------- - args : Dict[str, Any] - Dictionary of query parameters to convert to CQL2 format. - - Returns - ------- - str - Compact JSON string representation of the CQL2 query. - - Notes - ----- - Serialized with the tightest separators (no indentation or - whitespace). The body counts against the server's ~8 KB request-size - limit and against :func:`planning._request_bytes` when planning - chunks, so every saved byte fits more values per POST: compact - encoding roughly halves the per-value cost versus pretty-printing, - which roughly doubles how many monitoring-location ids fit in one - sub-request and so halves the chunk count for large id lists. - """ - query = { - "op": "and", - "args": [ - {"op": "in", "args": [{"property": key}, values]} - for key, values in args.items() - ], - } - return json.dumps(query, separators=(",", ":")) - - -def _check_ogc_requests( - endpoint: str, req_type: str = "queryables" -) -> tuple[dict[str, Any], httpx.Response]: - """ - Sends an HTTP GET request to the specified OGC endpoint and request type, - returning the parsed JSON body alongside the raw response (so a caller - that needs response-derived metadata, e.g. :class:`BaseMetadata`, doesn't - have to re-issue the request). - - Parameters - ---------- - endpoint : str - The OGC collection endpoint to query (e.g. the service/collection id). - req_type : str, optional - The type of request to make. Must be either "queryables" or "schema" - (default is "queryables"). - - Returns - ------- - dict - The JSON response from the OGC endpoint. - httpx.Response - The raw response, for callers that need it (URL, elapsed time, - headers). - - Raises - ------ - ValueError - If req_type is not "queryables" or "schema". - DataRetrievalError - From :func:`_raise_for_non_200` on any non-200 (the typed subclass for - the status) — same typed contract as the main data path so callers can - use one ``except`` clause everywhere. - """ - if req_type not in ("queryables", "schema"): - raise ValueError(f"req_type must be 'queryables' or 'schema', got {req_type!r}") - url = f"{_ogc_base_url.get()}/collections/{endpoint}/{req_type}" - resp = _get(url, headers=_default_headers(), **HTTPX_DEFAULTS) - _raise_for_non_200(resp) - # ``Response.json`` is typed ``Any``; the OGC queryables/schema endpoints - # return a JSON object, and callers index it as a dict. - return cast("dict[str, Any]", resp.json()), resp - - -def _ogc_query_params( - params: dict[str, Any], - *, - properties: list[str] | None, - bbox: list[float] | None, - limit: int | None, - skip_geometry: bool | None, -) -> dict[str, Any]: - """Add the shared OGC query knobs to ``params`` (mutated in place). - - Factors out the ``skipGeometry``/``limit``/``bbox``/``properties`` block - common to every OGC request so the typed getters - (:func:`_construct_api_requests`) and the generalized CQL2 path - (:func:`_construct_cql_request`) build identical URL parameters. - - ``skip_geometry=None`` leaves ``skipGeometry`` unset (the server defaults to - including geometry); the typed getters always pass a bool, so their behavior - is unchanged. - """ - if skip_geometry is not None: - params["skipGeometry"] = skip_geometry - params["limit"] = 50000 if limit is None or limit > 50000 else limit - # `len()` instead of truthiness: a numpy ndarray would raise on `if bbox:`. - if bbox is not None and len(bbox) > 0: - params["bbox"] = ",".join(map(str, bbox)) - if properties: - params["properties"] = ",".join(properties) - return params - - -def _construct_api_requests( - service: str, - properties: list[str] | None = None, - bbox: list[float] | None = None, - limit: int | None = None, - skip_geometry: bool | None = None, - **kwargs: Any, -) -> httpx.Request: - """ - Constructs an HTTP request object for the specified water data API service. - - For most services, list parameters are comma-joined and sent as a single - GET request (e.g. ``parameter_code=["00060","00010"]`` becomes - ``parameter_code=00060,00010`` in the URL). For services the active dialect - flags as CQL2-only (``dialect.cql2_services``, e.g. the Water Data API's - ``monitoring-locations``), a POST request with CQL2 JSON is used instead. - - Parameters - ---------- - service : str - The name of the API service to query (e.g., "daily"). - properties : Optional[List[str]], optional - List of property names to include in the request. - bbox : Optional[List[float]], optional - Bounding box coordinates as a list of floats. - limit : Optional[int], optional - Maximum number of results to return per request. - skip_geometry : bool, optional - Whether to exclude geometry from the response (default is False). - **kwargs - Additional query parameters, including date/time filters and other - API-specific options. - - Returns - ------- - httpx.Request - The constructed HTTP request object ready to be sent. - - Notes - ----- - - Date/time parameters are automatically formatted to ISO8601. - """ - service_url = f"{_ogc_base_url.get()}/collections/{service}/items" - dialect = _dialect.get() - - # Format date/time parameters to ISO8601 first — both routing paths need it. - for key in _DATE_RANGE_PARAMS: - if key in kwargs: - kwargs[key] = _format_api_dates( - kwargs[key], - date=(service in dialect.date_only_services and key != "last_modified"), - ) - - if service in dialect.cql2_services: - # POST with CQL2 JSON: multi-value params go in the request body. - # The date-range loop above has already collapsed any _DATE_RANGE_PARAMS - # value to a string, so the list/tuple check below cannot match them. - post_params = { - k: v - for k, v in kwargs.items() - if isinstance(v, (list, tuple)) and len(v) > 1 - } - params = {k: v for k, v in kwargs.items() if k not in post_params} - else: - # GET with comma-separated values: join list/tuple values into one string. - # Skip empty lists/tuples so they're omitted rather than emitted as a - # filterless ``¶m=`` (which the server reads as "match empty"). - post_params = {} - params = { - k: ",".join(str(x) for x in v) if isinstance(v, (list, tuple)) else v - for k, v in kwargs.items() - if not (isinstance(v, (list, tuple)) and len(v) == 0) - } - - _ogc_query_params( - params, - properties=properties, - bbox=bbox, - limit=limit, - skip_geometry=skip_geometry, - ) - - # Translate CQL filter Python names to the hyphenated URL parameter that - # the OGC API expects. The Python kwarg is `filter_lang` because hyphens - # aren't valid in Python identifiers. - if "filter_lang" in params: - params["filter-lang"] = params.pop("filter_lang") - - headers = _default_headers() - - if post_params: - headers["Content-Type"] = "application/query-cql-json" - return httpx.Request( - method="POST", - url=service_url, - headers=headers, - content=_cql2_param(post_params), - params=params, - ) - return httpx.Request( - method="GET", - url=service_url, - headers=headers, - params=params, - ) - - -def _construct_cql_request( - service: str, - cql_body: str, - *, - properties: list[str] | None = None, - bbox: list[float] | None = None, - limit: int | None = None, - skip_geometry: bool | None = None, -) -> httpx.Request: - """Build a POST/CQL2 request from a verbatim CQL2 body. - - The OGC-API counterpart to :func:`_construct_api_requests` for the - generalized :func:`~dataretrieval.waterdata.api.get_cql` path: the - caller supplies an already-serialized CQL2 JSON document (any predicate the - grammar allows), sent unchanged as the request body, while - ``properties``/``bbox``/``limit``/``skip_geometry`` go on the URL via the - shared :func:`_ogc_query_params` — so a generalized query and an equivalent - typed getter produce the same URL parameters. - - Parameters - ---------- - service : str - OGC collection name (e.g. ``"daily"``). - cql_body : str - Serialized CQL2 JSON document, sent as the POST body verbatim. - properties, bbox, limit, skip_geometry - See :func:`_ogc_query_params`. ``properties`` are wire-format - (``id``-translated) names. - - Returns - ------- - httpx.Request - A POST request with ``Content-Type: application/query-cql-json``. - """ - service_url = f"{_ogc_base_url.get()}/collections/{service}/items" - params = _ogc_query_params( - {}, - properties=properties, - bbox=bbox, - limit=limit, - skip_geometry=skip_geometry, - ) - headers = _default_headers() - headers["Content-Type"] = "application/query-cql-json" - return httpx.Request( - method="POST", - url=service_url, - headers=headers, - content=cql_body, - params=params, - ) +# Compatibility alias: the old name used internally and in tests. +_DEFAULT_DIALECT = DEFAULT_DIALECT def _next_req_url( @@ -584,30 +195,12 @@ async def _client_for( if shared is not None: yield shared return - async with httpx.AsyncClient(**HTTPX_DEFAULTS) as new: + async with httpx.AsyncClient(**HTTPX_ASYNC_DEFAULTS) as new: yield new _Cursor = TypeVar("_Cursor") -# Ambient per-call state the generic chunker would otherwise have to thread -# through to the deep request builder / paginate loop. Each is read with -# ``.get()`` and scoped with ``with _x(value):``; the defaults leave every -# existing getter unaffected. (Mirrors the ``_progress`` ambient-reporter.) - -# Optional cap on the rows one paginated call accumulates before it stops -# following ``next`` links (``None`` = uncapped). Set by :func:`get_reference_table` -# to preview large tables without downloading every page. -_row_cap: Ambient[int | None] = Ambient("ogc_row_cap", None) - -# OGC base URL the shared request builder (:func:`_construct_api_requests`) -# targets — the main Water Data API or, for NGWMN collections, their own base. -_ogc_base_url: Ambient[str] = Ambient("ogc_base_url", OGC_API_URL) - -# Per-call OGC dialect the request builder reads for CQL2-vs-GET routing and -# date-only formatting (default: a plain OGC API). -_dialect: Ambient[OgcDialect] = Ambient("ogc_dialect", _DEFAULT_DIALECT) - async def _paginate( initial_req: httpx.Request, @@ -988,202 +581,35 @@ def _run_sync( ) from exc -# ``AGENCY-ID``: a hyphen-separated agency prefix and local id. The local id -# may itself contain hyphens (``\S+`` after the first separator) — NGWMN -# aggregates many non-USGS agencies whose local ids aren't bare digits, so -# only the agency prefix is constrained to be hyphen/space-free. -_MONITORING_LOCATION_ID_RE = re.compile(r"[^-\s]+-\S+") - -# Default set of iterable-shaped params that ``_get_args`` must NOT push -# through ``_normalize_str_iterable`` (date-range params may carry -# ``pd.NaT``/None or interval strings; ``bbox`` is ``list[float]``). Callers -# with extra numeric params (e.g. the Water Data API's ``water_year``, -# ``thresholds``) pass their own superset. -_NO_NORMALIZE_PARAMS = _DATE_RANGE_PARAMS | {"bbox"} - - -def _normalize_str_iterable( - value: str | Iterable[str] | None, - param_name: str = "value", -) -> str | list[str] | None: - """Validate that ``value`` is None, a string, or an iterable of strings. - - Non-string iterables (``list``, ``tuple``, ``pandas.Series``, - ``pandas.Index``, ``numpy.ndarray``, generators) are materialized to a - ``list`` so downstream code that branches on ``isinstance(v, (list, - tuple))`` keeps working. ``Mapping`` types are rejected because - iterating a mapping yields keys, not values. - - Parameters - ---------- - value : None, str, or iterable of str - param_name : str, optional - Used in error messages. Defaults to ``"value"``. - - Returns - ------- - None, str, or list of str - - Raises - ------ - TypeError - If the input isn't ``None``, ``str``, or a non-``Mapping`` - iterable; or if any iterable element isn't a string. - """ - if value is None: - return None - if isinstance(value, str): - return value - if isinstance(value, Mapping) or not isinstance(value, Iterable): - raise TypeError( - f"{param_name} must be a string or iterable of strings, " - f"not {type(value).__name__} (got {value!r})." - ) - values: list[str] = [] - for v in value: - if not isinstance(v, str): - raise TypeError( - f"{param_name} elements must be strings, " - f"not {type(v).__name__} (got {v!r})." - ) - values.append(v) - return values - - -def _as_str_list( - value: str | Iterable[str] | None, - param_name: str = "value", -) -> list[str] | None: - """Normalize ``value`` to ``list[str]`` (``None`` passes through). - - Wraps a bare ``str`` in a single-element list — so a later - ``",".join(...)`` doesn't iterate it character-by-character — and - materializes any other iterable via :func:`_normalize_str_iterable`. - """ - normalized = _normalize_str_iterable(value, param_name) - if isinstance(normalized, str): - return [normalized] - return normalized - - -def _check_monitoring_location_id( - monitoring_location_id: str | Iterable[str] | None, -) -> str | list[str] | None: - """Validate and normalize a ``monitoring_location_id`` value. +def fetch_ogc_request( + request: httpx.Request, + *, + service: str, +) -> tuple[pd.DataFrame, httpx.Response]: + """Execute a prepared OGC request with pagination, returning (df, response). - Combines :func:`_normalize_str_iterable` with the AGENCY-ID format - check that is unique to ``monitoring_location_id`` (the OGC spec - requires a hyphen separator, e.g. ``USGS-01646500``). + This is the facade-level entry point for generalized CQL requests: the + caller builds its own :class:`httpx.Request` (e.g. via + :func:`~dataretrieval.ogc.requests._construct_cql_request`) and hands it + here. Pagination, progress reporting, and error handling are identical to + the typed getters' path through :func:`_walk_pages`. Parameters ---------- - monitoring_location_id : None, str, or iterable of str - See :func:`_normalize_str_iterable`. Each string is additionally - required to match the AGENCY-ID hyphen-separated format. + request : httpx.Request + A fully-constructed OGC API request (typically a POST/CQL2). + service : str + Collection name, used only for progress-context labelling. Returns ------- - None, str, or list of str - - Raises - ------ - TypeError - If the input isn't ``None``, ``str``, or a non-``Mapping`` - iterable; or if any iterable element isn't a string. - ValueError - If any identifier doesn't contain a hyphen separator - (per the OGC API spec: AGENCY-ID format, e.g. ``USGS-01646500``). - """ - try: - value = _normalize_str_iterable( - monitoring_location_id, "monitoring_location_id" - ) - except TypeError as exc: - # Re-raise with the AGENCY-ID hint the generic helper doesn't carry. - raise TypeError( - f"{exc} Expected 'AGENCY-ID' format, e.g., 'USGS-01646500'." - ) from None - if value is None: - return None - for item in (value,) if isinstance(value, str) else value: - if not _MONITORING_LOCATION_ID_RE.fullmatch(item): - raise ValueError( - f"Invalid monitoring_location_id: {item!r}. " - f"Expected 'AGENCY-ID' format, e.g., 'USGS-01646500'." - ) - return value - - -def _get_args( - local_vars: dict[str, Any], - exclude: set[str] | None = None, - *, - no_normalize: frozenset[str] | set[str] = _NO_NORMALIZE_PARAMS, -) -> dict[str, Any]: + pd.DataFrame + Concatenated page results. + httpx.Response + Aggregated response metadata. """ - Build the API-request kwargs dict from a getter's ``locals()``. - - Drops bookkeeping keys (``service``, ``output_id``, anything in - ``exclude``) and ``None``-valued kwargs, then normalizes the - remaining values: - - - ``monitoring_location_id`` is validated against the AGENCY-ID - format (per :func:`_check_monitoring_location_id`). - - ``properties`` is materialized to ``list[str]`` (a bare string - gets wrapped in a single-element list so downstream - ``",".join(properties)`` doesn't iterate per character). - - A non-string iterable in ``no_normalize`` (numeric params - such as ``water_year``, ``bbox``, ``thresholds``) is materialized - to a ``list`` with its element types preserved (no string - normalization), so the GET comma-join and the chunker — which test - ``list``/``tuple`` — handle it instead of ``str()``-ing the whole - array. - - Any other ``Iterable[str]`` (i.e. not in ``no_normalize``) - is materialized to ``list[str]`` via - :func:`_normalize_str_iterable` so downstream code that branches - on ``isinstance(v, (list, tuple))`` works for ``pandas.Series``, - ``numpy.ndarray``, generators, etc. - - Scalars and strings pass through unchanged. - Parameters - ---------- - local_vars : dict[str, Any] - Dictionary of local variables, typically from ``locals()``. - exclude : set[str], optional - Additional keys to exclude from the resulting dictionary. - no_normalize : set[str], optional - Iterable-shaped params whose element types must be preserved - (no string normalization). Defaults to the generic date-range + - ``bbox`` set; callers with extra numeric params pass a superset. + async def _coro() -> tuple[pd.DataFrame, httpx.Response]: + return await _walk_pages(geopd=GEOPANDAS, req=request) - Returns - ------- - dict[str, Any] - Filtered and normalized arguments for API requests. - """ - to_exclude = {"service", "output_id"} - if exclude: - to_exclude.update(exclude) - - args: dict[str, Any] = {} - for k, v in local_vars.items(): - if k in to_exclude or v is None: - continue - if k == "monitoring_location_id": - args[k] = _check_monitoring_location_id(v) - elif k == "properties": - args[k] = _as_str_list(v, k) - elif k in no_normalize and isinstance(v, Iterable) and not isinstance(v, str): - # Numeric params (water_year, bbox, thresholds, …) keep their - # element types — no string-normalization — but a non-string - # iterable (numpy array, pandas Series, generator) is materialized - # to a list so the GET comma-join and the chunker, which test - # ``list``/``tuple``, handle it instead of str()-ing the whole - # array. ``.tolist()`` yields native int/float; ``list()`` covers - # generators and other iterables. Scalars/strings fall through. - args[k] = v.tolist() if hasattr(v, "tolist") else list(v) - elif isinstance(v, str) or not isinstance(v, Iterable): - args[k] = v - else: - args[k] = _normalize_str_iterable(v, k) - return args + return _run_sync(_coro, service=service) diff --git a/dataretrieval/ogc/policy.py b/dataretrieval/ogc/policy.py new file mode 100644 index 00000000..7aff5398 --- /dev/null +++ b/dataretrieval/ogc/policy.py @@ -0,0 +1,63 @@ +"""Low-level OGC policy: dialect types and default endpoint constants. + +This module is the single source of truth for the :class:`OgcDialect` type +(per-API quirks the generic request builder needs) and the default endpoint +constants used by the Water Data OGC API. It depends only on the stdlib so it +can be imported safely by any OGC submodule without creating cycles. + +It must NOT import engine, shaping, or any service adapter. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field + +# --------------------------------------------------------------------------- +# Endpoint constants +# --------------------------------------------------------------------------- + +BASE_URL = "https://api.waterdata.usgs.gov" +OGC_API_VERSION = "v0" +OGC_API_URL = f"{BASE_URL}/ogcapi/{OGC_API_VERSION}" + +# --------------------------------------------------------------------------- +# Dialect type +# --------------------------------------------------------------------------- + + +@dataclass(frozen=True) +class OgcDialect: + """Per-API quirks the generic request builder needs to know about. + + Attributes + ---------- + cql2_services : frozenset[str] + Collections that don't accept comma-separated multi-value GET + parameters and so must be queried via POST with a CQL2 JSON body. + date_only_services : frozenset[str] + Collections whose time arguments are rendered date-only + (``YYYY-MM-DD``) rather than as a full UTC datetime. The + ``last_modified`` parameter is always rendered as a full datetime + regardless of this set. + time_cols : frozenset[str] + Result columns to coerce to datetime when ``convert_type`` is set. + Empty by default, so the generic engine carries no API-specific + column knowledge; each API supplies its own. + numerical_cols : frozenset[str] + Result columns to coerce to numeric when ``convert_type`` is set. + sort_cols : tuple[str, ...] + Columns to sort the combined result by, in priority order. Sorting + is applied only when the first (primary) column is present; any + later columns also present are added as secondary keys. + """ + + cql2_services: frozenset[str] = field(default_factory=frozenset) + date_only_services: frozenset[str] = field(default_factory=frozenset) + time_cols: frozenset[str] = field(default_factory=frozenset) + numerical_cols: frozenset[str] = field(default_factory=frozenset) + sort_cols: tuple[str, ...] = field(default_factory=tuple) + + +# 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() diff --git a/dataretrieval/ogc/requests.py b/dataretrieval/ogc/requests.py new file mode 100644 index 00000000..eb5a201b --- /dev/null +++ b/dataretrieval/ogc/requests.py @@ -0,0 +1,331 @@ +"""OGC request preparation, construction, and schema/queryables lookup. + +This module owns the machinery for building OGC API requests (both GET and +POST/CQL2 paths), the ambient base-URL and dialect state that request builders +read, and the queryables/schema request helper used by empty-result shaping. + +It depends on :mod:`~dataretrieval.ogc.policy` (the dialect type and endpoint +constants), :mod:`~dataretrieval.ogc.dates`, :mod:`~dataretrieval.ogc.errors`, +and :mod:`~dataretrieval.utils` (shared HTTP primitives). It must NOT import +engine or shaping. +""" + +from __future__ import annotations + +import json +import logging +import re +from collections.abc import Iterable, Mapping +from typing import Any, cast + +import httpx + +from dataretrieval.ogc.dates import _DATE_RANGE_PARAMS, _format_api_dates +from dataretrieval.ogc.errors import _raise_for_non_200 +from dataretrieval.ogc.policy import DEFAULT_DIALECT, OGC_API_URL, OgcDialect +from dataretrieval.utils import HTTPX_DEFAULTS, Ambient, _default_headers, _get + +logger = logging.getLogger(__name__) + +# --------------------------------------------------------------------------- +# Ambient per-call state +# --------------------------------------------------------------------------- + +# Optional cap on the rows one paginated call accumulates before it stops +# following ``next`` links (``None`` = uncapped). Set by :func:`get_reference_table` +# to preview large tables without downloading every page. +_row_cap: Ambient[int | None] = Ambient("ogc_row_cap", None) + +# OGC base URL the shared request builder (:func:`_construct_api_requests`) +# targets — the main Water Data API or, for NGWMN collections, their own base. +_ogc_base_url: Ambient[str] = Ambient("ogc_base_url", OGC_API_URL) + +# Per-call OGC dialect the request builder reads for CQL2-vs-GET routing and +# date-only formatting (default: a plain OGC API). +_dialect: Ambient[OgcDialect] = Ambient("ogc_dialect", DEFAULT_DIALECT) + + +# --------------------------------------------------------------------------- +# Monitoring location ID validation +# --------------------------------------------------------------------------- + +# ``AGENCY-ID``: a hyphen-separated agency prefix and local id. The local id +# may itself contain hyphens (``\S+`` after the first separator) — NGWMN +# aggregates many non-USGS agencies whose local ids aren't bare digits, so +# only the agency prefix is constrained to be hyphen/space-free. +_MONITORING_LOCATION_ID_RE = re.compile(r"[^-\s]+-\S+") + + +# --------------------------------------------------------------------------- +# Request building helpers +# --------------------------------------------------------------------------- + + +def _switch_arg_id(ls: dict[str, Any], id_name: str, service: str) -> dict[str, Any]: + """Switch argument id from its package-specific identifier to the + standardized "id" key that the API recognizes.""" + service_id = service.replace("-", "_") + "_id" + if "id" not in ls: + if service_id in ls: + ls["id"] = ls[service_id] + elif id_name in ls: + ls["id"] = ls[id_name] + ls.pop(service_id, None) + ls.pop(id_name, None) + return ls + + +def _switch_properties_id( + properties: list[str] | None, id_name: str, service: str +) -> list[str]: + """Build the wire ``properties`` list, dropping every id alias and + ``geometry``.""" + if not properties: + return [] + service_id = service.replace("-", "_") + "_id" + drop = {"id", "geometry", id_name, service_id} + normalized = (p.replace("-", "_") for p in properties) + return [p for p in normalized if p not in drop] + + +def _cql2_param(args: dict[str, Any]) -> str: + """Convert query parameters to CQL2 JSON format for POST requests.""" + query = { + "op": "and", + "args": [ + {"op": "in", "args": [{"property": key}, values]} + for key, values in args.items() + ], + } + return json.dumps(query, separators=(",", ":")) + + +def _ogc_query_params( + params: dict[str, Any], + *, + properties: list[str] | None, + bbox: list[float] | None, + limit: int | None, + skip_geometry: bool | None, +) -> dict[str, Any]: + """Add the shared OGC query knobs to ``params`` (mutated in place).""" + if skip_geometry is not None: + params["skipGeometry"] = skip_geometry + params["limit"] = 50000 if limit is None or limit > 50000 else limit + if bbox is not None and len(bbox) > 0: + params["bbox"] = ",".join(map(str, bbox)) + if properties: + params["properties"] = ",".join(properties) + return params + + +def _construct_api_requests( + service: str, + properties: list[str] | None = None, + bbox: list[float] | None = None, + limit: int | None = None, + skip_geometry: bool | None = None, + **kwargs: Any, +) -> httpx.Request: + """Construct an HTTP request object for the specified OGC API service.""" + service_url = f"{_ogc_base_url.get()}/collections/{service}/items" + dialect = _dialect.get() + + for key in _DATE_RANGE_PARAMS: + if key in kwargs: + kwargs[key] = _format_api_dates( + kwargs[key], + date=(service in dialect.date_only_services and key != "last_modified"), + ) + + if service in dialect.cql2_services: + post_params = { + k: v + for k, v in kwargs.items() + if isinstance(v, (list, tuple)) and len(v) > 1 + } + params = {k: v for k, v in kwargs.items() if k not in post_params} + else: + post_params = {} + params = { + k: ",".join(str(x) for x in v) if isinstance(v, (list, tuple)) else v + for k, v in kwargs.items() + if not (isinstance(v, (list, tuple)) and len(v) == 0) + } + + _ogc_query_params( + params, + properties=properties, + bbox=bbox, + limit=limit, + skip_geometry=skip_geometry, + ) + + if "filter_lang" in params: + params["filter-lang"] = params.pop("filter_lang") + + headers = _default_headers(service_url) + + if post_params: + headers["Content-Type"] = "application/query-cql-json" + return httpx.Request( + method="POST", + url=service_url, + headers=headers, + content=_cql2_param(post_params), + params=params, + ) + return httpx.Request( + method="GET", + url=service_url, + headers=headers, + params=params, + ) + + +def _construct_cql_request( + service: str, + cql_body: str, + *, + properties: list[str] | None = None, + bbox: list[float] | None = None, + limit: int | None = None, + skip_geometry: bool | None = None, +) -> httpx.Request: + """Build a POST/CQL2 request from a verbatim CQL2 body.""" + service_url = f"{_ogc_base_url.get()}/collections/{service}/items" + params = _ogc_query_params( + {}, + properties=properties, + bbox=bbox, + limit=limit, + skip_geometry=skip_geometry, + ) + headers = _default_headers(service_url) + headers["Content-Type"] = "application/query-cql-json" + return httpx.Request( + method="POST", + url=service_url, + headers=headers, + content=cql_body, + params=params, + ) + + +def _check_ogc_requests( + endpoint: str, req_type: str = "queryables" +) -> tuple[dict[str, Any], httpx.Response]: + """Send an HTTP GET request to the OGC endpoint for queryables/schema.""" + if req_type not in ("queryables", "schema"): + raise ValueError(f"req_type must be 'queryables' or 'schema', got {req_type!r}") + url = f"{_ogc_base_url.get()}/collections/{endpoint}/{req_type}" + resp = _get(url, headers=_default_headers(url), **HTTPX_DEFAULTS) + _raise_for_non_200(resp) + return cast("dict[str, Any]", resp.json()), resp + + +# --------------------------------------------------------------------------- +# Argument normalization helpers +# --------------------------------------------------------------------------- + +# Default set of iterable-shaped params that ``_get_args`` must NOT push +# through ``_normalize_str_iterable`` (date-range params may carry +# ``pd.NaT``/None or interval strings; ``bbox`` is ``list[float]``). Callers +# with extra numeric params pass their own superset. +_NO_NORMALIZE_PARAMS = _DATE_RANGE_PARAMS | {"bbox"} + + +def _normalize_str_iterable( + value: str | Iterable[str] | None, + param_name: str = "value", +) -> str | list[str] | None: + """Validate that ``value`` is None, a string, or an iterable of strings.""" + if value is None: + return None + if isinstance(value, str): + return value + if isinstance(value, Mapping) or not isinstance(value, Iterable): + raise TypeError( + f"{param_name} must be a string or iterable of strings, " + f"not {type(value).__name__} (got {value!r})." + ) + values: list[str] = [] + for v in value: + if not isinstance(v, str): + raise TypeError( + f"{param_name} elements must be strings, " + f"not {type(v).__name__} (got {v!r})." + ) + values.append(v) + return values + + +def _as_str_list( + value: str | Iterable[str] | None, + param_name: str = "value", +) -> list[str] | None: + """Normalize ``value`` to ``list[str]`` (``None`` passes through).""" + normalized = _normalize_str_iterable(value, param_name) + if isinstance(normalized, str): + return [normalized] + return normalized + + +def _check_monitoring_location_id( + monitoring_location_id: str | Iterable[str] | None, +) -> str | list[str] | None: + """Validate and normalize a ``monitoring_location_id`` value.""" + try: + value = _normalize_str_iterable( + monitoring_location_id, "monitoring_location_id" + ) + except TypeError as exc: + raise TypeError( + f"{exc} Expected 'AGENCY-ID' format, e.g., 'USGS-01646500'." + ) from None + if value is None: + return None + for item in (value,) if isinstance(value, str) else value: + if not _MONITORING_LOCATION_ID_RE.fullmatch(item): + raise ValueError( + f"Invalid monitoring_location_id: {item!r}. " + f"Expected 'AGENCY-ID' format, e.g., 'USGS-01646500'." + ) + return value + + +def prepare_request_args( + local_vars: dict[str, Any], + exclude: set[str] | None = None, + *, + no_normalize: frozenset[str] | set[str] = _NO_NORMALIZE_PARAMS, +) -> dict[str, Any]: + """Build OGC request kwargs from a getter's ``locals()``. + + Internal bookkeeping keys, caller-supplied exclusions, and ``None`` values + are omitted. Identifiers and properties are validated; other iterables are + normalized unless listed in ``no_normalize``. + """ + to_exclude = {"service", "output_id"} + if exclude: + to_exclude.update(exclude) + + args: dict[str, Any] = {} + for k, v in local_vars.items(): + if k in to_exclude or v is None: + continue + if k == "monitoring_location_id": + args[k] = _check_monitoring_location_id(v) + elif k == "properties": + args[k] = _as_str_list(v, k) + elif k in no_normalize and isinstance(v, Iterable) and not isinstance(v, str): + args[k] = v.tolist() if hasattr(v, "tolist") else list(v) + elif isinstance(v, str) or not isinstance(v, Iterable): + args[k] = v + else: + args[k] = _normalize_str_iterable(v, k) + return args + + +# Compatibility alias for existing private imports from ``ogc.engine``. +_get_args = prepare_request_args diff --git a/dataretrieval/ogc/retry.py b/dataretrieval/ogc/retry.py index 37ea040a..bd45f275 100644 --- a/dataretrieval/ogc/retry.py +++ b/dataretrieval/ogc/retry.py @@ -19,8 +19,8 @@ import httpx import pandas as pd +import dataretrieval.ogc.progress as _progress from dataretrieval.exceptions import RateLimited, TransientError -from dataretrieval.ogc import progress as _progress from dataretrieval.ogc.interruptions import ( ChunkInterrupted, QuotaExhausted, diff --git a/dataretrieval/ogc/shaping.py b/dataretrieval/ogc/shaping.py index 5870c63b..383e54f1 100644 --- a/dataretrieval/ogc/shaping.py +++ b/dataretrieval/ogc/shaping.py @@ -12,16 +12,14 @@ import logging import re -from typing import TYPE_CHECKING, Any +from typing import Any import httpx import pandas as pd +from dataretrieval.ogc.policy import DEFAULT_DIALECT, OgcDialect from dataretrieval.utils import BaseMetadata -if TYPE_CHECKING: - from dataretrieval.ogc.engine import OgcDialect - try: import geopandas as gpd @@ -185,11 +183,8 @@ def _deal_with_empty( """ if return_list.empty: if not properties or all(pd.isna(properties)): - # Lazy import to avoid a cycle: ``_check_ogc_requests`` is a - # request-side helper in the engine, which imports this module. - # This rare empty-result schema lookup is the only shaping->engine - # call (it goes away once requests move to their own module). - from dataretrieval.ogc.engine import _check_ogc_requests + # Import from requests module (no engine dependency). + from dataretrieval.ogc.requests import _check_ogc_requests schema, _ = _check_ogc_requests(endpoint=service, req_type="schema") properties = list(schema.get("properties", {}).keys()) @@ -374,11 +369,7 @@ def _finalize_ogc( per-``_paginate`` ``_row_cap`` is only an early-stop download bound. """ if dialect is None: - # The default lives in the engine (the dialect type's home); import it - # lazily so this module needs no engine import at load time. - from dataretrieval.ogc.engine import _DEFAULT_DIALECT - - dialect = _DEFAULT_DIALECT + dialect = DEFAULT_DIALECT frame = _deal_with_empty(frame, properties, service) # Normalize to PEP-8 snake_case column names *first*, so the dialect's # ``time_cols``/``numerical_cols``/``sort_cols`` (all snake_case) match diff --git a/dataretrieval/utils.py b/dataretrieval/utils.py index acdf7822..7506a469 100644 --- a/dataretrieval/utils.py +++ b/dataretrieval/utils.py @@ -10,12 +10,13 @@ from collections.abc import Callable, Iterable, Iterator from contextlib import contextmanager from contextvars import ContextVar +from importlib.metadata import PackageNotFoundError +from importlib.metadata import version as _pkg_version from typing import Any, Generic, TypeVar import httpx import pandas as pd -import dataretrieval from dataretrieval.codes import tz from dataretrieval.exceptions import ( NetworkError, @@ -24,6 +25,11 @@ error_for_status, ) +try: + _PACKAGE_VERSION = _pkg_version("dataretrieval") +except PackageNotFoundError: + _PACKAGE_VERSION = "version-unknown" + # Typed as ``dict[str, Any]`` (not the inferred ``dict[str, object]``) so that # splatting it as ``**HTTPX_DEFAULTS`` into ``httpx.get`` / ``httpx.AsyncClient`` # type-checks: the values are a heterogeneous bag of httpx keyword arguments. @@ -97,35 +103,75 @@ def _require_positive_int( raise ValueError(f"{name} must be a positive integer{eg} (got {value!r}).") -def _default_headers() -> dict[str, str]: +# The single authorized host for the API key. The key is a USGS personal +# access token issued for the Water Data API and must never be forwarded to +# non-USGS hosts, lookalikes, or external endpoints (including STAC rating +# asset downloads). +_AUTHORIZED_API_KEY_HOST = "api.waterdata.usgs.gov" + + +def _default_headers(target_url: str | httpx.URL | None = None) -> dict[str, str]: """Build the default HTTP headers for a USGS web-API request. Always sets a descriptive ``User-Agent`` plus ``Accept`` / ``Accept-Encoding`` and ``lang``. If the ``API_USGS_PAT`` environment - variable is set, its value is added as the ``X-Api-Key`` header — a USGS - personal access token raises the request rate limit. + variable is set AND ``target_url`` points to the explicitly authorized + Water Data host (``api.waterdata.usgs.gov``), its value is added as the + ``X-Api-Key`` header. The key is never sent to other hosts. - Shared by the OGC engine (:mod:`dataretrieval.ogc`), the Water Data getters - (:mod:`dataretrieval.waterdata`), and :mod:`dataretrieval.wateruse`, so the - request identity is consistent across every USGS API the package talks to. + Parameters + ---------- + target_url : str or httpx.URL or None + The URL the request will be sent to. When provided, the API key is + included only if the host matches the authorized Water Data host. + When ``None`` (legacy callers), the key is NOT included — callers + must pass the concrete URL. Returns ------- dict[str, str] - Headers suitable for an ``httpx`` request against a USGS API. + Headers suitable for an ``httpx`` request. """ headers = { "Accept-Encoding": "compress, gzip", "Accept": "application/json", - "User-Agent": f"python-dataretrieval/{dataretrieval.__version__}", + "User-Agent": f"python-dataretrieval/{_PACKAGE_VERSION}", "lang": "en-US", } token = os.getenv("API_USGS_PAT") - if token: - headers["X-Api-Key"] = token + if token and target_url is not None: + try: + host = httpx.URL(str(target_url)).host + except (httpx.InvalidURL, TypeError): + host = None + if host == _AUTHORIZED_API_KEY_HOST: + headers["X-Api-Key"] = token return headers +def _strip_api_key_from_untrusted_host(request: httpx.Request) -> None: + """Remove Water Data credentials from any request to another host. + + HTTPX retains arbitrary custom headers across cross-origin redirects. This + hook runs for the initial request and every redirect, making host scoping an + execution-time invariant rather than relying only on initial header + construction. + """ + if request.url.host != _AUTHORIZED_API_KEY_HOST: + request.headers.pop("X-Api-Key", None) + + +async def _strip_api_key_from_untrusted_host_async(request: httpx.Request) -> None: + """Async-client form of :func:`_strip_api_key_from_untrusted_host`.""" + _strip_api_key_from_untrusted_host(request) + + +HTTPX_ASYNC_DEFAULTS: dict[str, Any] = { + **HTTPX_DEFAULTS, + "event_hooks": {"request": [_strip_api_key_from_untrusted_host_async]}, +} + + def to_str(listlike: object, delimiter: str = ",") -> str | None: """Translates list-like objects into strings. @@ -391,11 +437,21 @@ def _network_error(url: str | httpx.URL, exc: httpx.TransportError) -> NetworkEr def _get(url: str | httpx.URL, **kwargs: Any) -> httpx.Response: - """``httpx.get`` for the single-shot paths, surfacing a transport failure as - a typed :class:`~dataretrieval.exceptions.NetworkError` (the chunker wraps its - own as resumable interruptions, so it stays off this wrapper).""" + """Issue one guarded synchronous GET and map transport failures. + + A short-lived client supplies a request hook for both the initial request + and every redirect. The hook removes ``X-Api-Key`` unless the destination + host is the authorized Water Data API host. + """ + client_options: dict[str, Any] = { + key: kwargs.pop(key) + for key in ("follow_redirects", "timeout", "transport", "verify") + if key in kwargs + } + client_options["event_hooks"] = {"request": [_strip_api_key_from_untrusted_host]} try: - return httpx.get(url, **kwargs) + with httpx.Client(**client_options) as client: + return client.get(url, **kwargs) except httpx.TransportError as exc: raise _network_error(url, exc) from exc @@ -479,7 +535,7 @@ def query( # Drop them. (``to_str`` returns None for non-iterable scalars like bools.) payload = {k: v for k, v in payload.items() if v is not None} - user_agent = {"user-agent": f"python-dataretrieval/{dataretrieval.__version__}"} + user_agent = {"user-agent": f"python-dataretrieval/{_PACKAGE_VERSION}"} try: response = _get( diff --git a/dataretrieval/waterdata/api.py b/dataretrieval/waterdata/api.py index 54785691..7073d183 100644 --- a/dataretrieval/waterdata/api.py +++ b/dataretrieval/waterdata/api.py @@ -16,11 +16,20 @@ import httpx import pandas as pd +from dataretrieval.ogc import fetch_ogc_request +from dataretrieval.ogc.errors import _raise_for_non_200 from dataretrieval.ogc.filters import FILTER_LANG +from dataretrieval.ogc.requests import ( + _as_str_list, + _check_ogc_requests, + _construct_cql_request, + _switch_properties_id, +) from dataretrieval.utils import ( HTTPX_DEFAULTS, BaseMetadata, _attach_datetime_columns, + _default_headers, _get, to_str, ) @@ -34,20 +43,11 @@ ) from dataretrieval.waterdata.utils import ( _OUTPUT_ID_BY_SERVICE, - GEOPANDAS, SAMPLES_URL, _accept_legacy_kwargs, - _as_str_list, - _check_ogc_requests, _check_profiles, - _construct_cql_request, - _default_headers, _finalize_ogc, _get_args, - _raise_for_non_200, - _run_sync, - _switch_properties_id, - _walk_pages, _with_state, get_ogc_data, ) @@ -2433,7 +2433,7 @@ def get_codes(code_service: CODE_SERVICES) -> tuple[pd.DataFrame, BaseMetadata]: url = f"{SAMPLES_URL}/codeservice/{code_service}?mimeType=application%2Fjson" - response = _get(url, headers=_default_headers(), **HTTPX_DEFAULTS) + response = _get(url, headers=_default_headers(url), **HTTPX_DEFAULTS) _raise_for_non_200(response) @@ -2461,7 +2461,7 @@ def _get_samples_csv( url, params=params, verify=ssl_check, - headers=_default_headers(), + headers=_default_headers(url), **HTTPX_DEFAULTS, ) _raise_for_non_200(response) @@ -3427,10 +3427,7 @@ def get_cql( skip_geometry=skip_geometry, ) - async def _run() -> tuple[pd.DataFrame, httpx.Response]: - return await _walk_pages(geopd=GEOPANDAS, req=req) - - df, response = _run_sync(_run, service=service) + df, response = fetch_ogc_request(req, service=service) return _finalize_ogc( df, diff --git a/dataretrieval/waterdata/ratings.py b/dataretrieval/waterdata/ratings.py index de4ba0d4..cbaab057 100644 --- a/dataretrieval/waterdata/ratings.py +++ b/dataretrieval/waterdata/ratings.py @@ -18,18 +18,14 @@ import pandas as pd from dataretrieval.exceptions import DataRetrievalError +from dataretrieval.ogc.dates import _DURATION_RE, _format_api_dates +from dataretrieval.ogc.errors import _raise_for_non_200 from dataretrieval.ogc.filters import _quote_cql_str +from dataretrieval.ogc.requests import _check_monitoring_location_id from dataretrieval.rdb import extract_rdb_comment, read_rdb -from dataretrieval.utils import HTTPX_DEFAULTS, _get +from dataretrieval.utils import HTTPX_DEFAULTS, _default_headers, _get -from .utils import ( - _DURATION_RE, - BASE_URL, - _check_monitoring_location_id, - _default_headers, - _format_api_dates, - _raise_for_non_200, -) +from .utils import BASE_URL logger = logging.getLogger(__name__) @@ -255,7 +251,7 @@ def _search( response = _get( url, params=params, - headers=_default_headers(), + headers=_default_headers(url), verify=ssl_check, **HTTPX_DEFAULTS, ) @@ -280,7 +276,9 @@ def _download_and_parse( ) -> pd.DataFrame: """Fetch the feature's data asset, parse RDB, optionally persist to disk.""" url = feature["assets"]["data"]["href"] - response = _get(url, headers=_default_headers(), verify=ssl_check, **HTTPX_DEFAULTS) + response = _get( + url, headers=_default_headers(url), verify=ssl_check, **HTTPX_DEFAULTS + ) _raise_for_non_200(response) if file_path is not None: diff --git a/dataretrieval/waterdata/stats.py b/dataretrieval/waterdata/stats.py index 2aa357de..abba5deb 100644 --- a/dataretrieval/waterdata/stats.py +++ b/dataretrieval/waterdata/stats.py @@ -17,7 +17,6 @@ import pandas as pd from dataretrieval.ogc.engine import ( - BASE_URL, _paginate, _run_sync, ) @@ -28,6 +27,7 @@ _empty_feature_frame, ) from dataretrieval.utils import BaseMetadata, _default_headers +from dataretrieval.waterdata.utils import BASE_URL # ``_handle_nesting``'s geopandas branch calls ``gpd.GeoDataFrame.from_features`` # directly, so this module needs its own bound ``gpd`` name. Import it under the @@ -261,7 +261,7 @@ def get_data( req = httpx.Request( method="GET", url=url, - headers=_default_headers(), + headers=_default_headers(url), params=args, ) method = req.method diff --git a/dataretrieval/waterdata/utils.py b/dataretrieval/waterdata/utils.py index ba99e941..a6d91272 100644 --- a/dataretrieval/waterdata/utils.py +++ b/dataretrieval/waterdata/utils.py @@ -1,16 +1,17 @@ -"""Water Data API layer over the generic OGC engine. - -The API-agnostic OGC machinery (request construction, pagination, response -shaping, the chunked ``get_ogc_data`` entry point) lives in the -:mod:`dataretrieval.ogc` package — :mod:`~dataretrieval.ogc.engine` and its -sibling modules (``dates``, ``errors``, ``shaping``, ``chunking``). This -module is the Water-Data-specific layer -on top of it: it supplies the service-to-id map, the CQL2/date-only dialect, -profile validation, and a thin ``get_ogc_data`` wrapper that injects the -Water Data defaults. (The statistics path lives in its own -:mod:`dataretrieval.waterdata.stats` module.) Every engine symbol the Water Data -getters (``api.py``, ``ratings.py``, ``nearest.py``) and the test suite import -from here is re-exported below. +"""Water Data API layer over the generic OGC facade. + +This module is the Water-Data-specific adapter: it supplies the +service-to-id map, the CQL2/date-only dialect, profile validation, and a +thin ``get_ogc_data`` wrapper that injects the Water Data defaults. The +statistics path lives in its own :mod:`dataretrieval.waterdata.stats` +module. + +OGC machinery (request construction, pagination, response shaping, the +chunked ``get_ogc_data`` entry point) lives in :mod:`dataretrieval.ogc` +and its implementation submodules. This adapter consumes the public facade +for dialects, argument normalization, and retrieval; callers that need an OGC +implementation helper import its canonical module directly rather than using +this module as a re-export layer. """ from __future__ import annotations @@ -23,56 +24,24 @@ import httpx import pandas as pd +import dataretrieval.ogc.dates as _ogc_dates +import dataretrieval.ogc.shaping as _ogc_shaping from dataretrieval.codes.states import apply_state -from dataretrieval.ogc import engine -from dataretrieval.ogc.dates import ( - _DATE_RANGE_PARAMS, - _DURATION_RE, - _format_api_dates, -) -from dataretrieval.ogc.engine import ( - BASE_URL, - OGC_API_URL, - OgcDialect, - _as_str_list, - _check_monitoring_location_id, - _check_ogc_requests, - _construct_api_requests, - _construct_cql_request, - _next_req_url, - _normalize_str_iterable, - _paginate, - _row_cap, - _run_sync, - _switch_properties_id, - _walk_pages, -) -from dataretrieval.ogc.engine import ( - _get_args as _engine_get_args, -) -from dataretrieval.ogc.errors import ( - _error_body, - _paginated_failure_message, - _parse_retry_after, - _raise_for_non_200, -) -from dataretrieval.ogc.shaping import ( - GEOPANDAS, - _arrange_cols, - _deal_with_empty, - _get_resp_data, - _to_snake_case, -) -from dataretrieval.ogc.shaping import ( - _finalize_ogc as _engine_finalize_ogc, -) -from dataretrieval.utils import BaseMetadata, _default_headers +from dataretrieval.ogc import OgcDialect, prepare_request_args +from dataretrieval.ogc import get_ogc_data as _facade_get_ogc_data +from dataretrieval.utils import BaseMetadata from dataretrieval.waterdata.types import ( PROFILE_LOOKUP, PROFILES, SERVICES, ) +# --------------------------------------------------------------------------- +# Water Data endpoint constants (defined locally, not imported from OGC policy) +# --------------------------------------------------------------------------- + +BASE_URL = "https://api.waterdata.usgs.gov" +OGC_API_URL = f"{BASE_URL}/ogcapi/v0" SAMPLES_URL = f"{BASE_URL}/samples-data" # Maps each OGC waterdata service to its user-facing ``id`` column (the name the @@ -143,7 +112,7 @@ # - ``bbox``/``boundingBox`` are ``list[float]``, sometimes ``numpy.ndarray`` # - ``get_peaks``'s int-valued filters (``water_year`` etc.) are ``list[int]`` # - ``get_combined_metadata``'s ``thresholds`` is ``list[float]`` -_NO_NORMALIZE_PARAMS = _DATE_RANGE_PARAMS | { +_NO_NORMALIZE_PARAMS = _ogc_dates._DATE_RANGE_PARAMS | { "bbox", "boundingBox", "water_year", @@ -175,16 +144,15 @@ def _flatten_queryables(local_vars: dict[str, Any]) -> dict[str, Any]: def _get_args( local_vars: dict[str, Any], exclude: set[str] | None = None ) -> dict[str, Any]: - """Water-Data wrapper over :func:`engine._get_args`. + """Water-Data wrapper over :func:`~dataretrieval.ogc.prepare_request_args`. Supplies the Water Data API's extended ``no_normalize`` set (numeric params such as ``water_year``, ``thresholds``, ``boundingBox``) so they - keep their element types. See :func:`engine._get_args` for the full - normalization contract. Also flattens any ``**queryables`` passthrough + keep their element types. Also flattens any ``**queryables`` passthrough (see :func:`_flatten_queryables`). """ _flatten_queryables(local_vars) - return _engine_get_args(local_vars, exclude, no_normalize=_NO_NORMALIZE_PARAMS) + return prepare_request_args(local_vars, exclude, no_normalize=_NO_NORMALIZE_PARAMS) def _with_state(local_vars: dict[str, Any], *, to: str, into: str) -> dict[str, Any]: @@ -216,13 +184,13 @@ def get_ogc_data( output_id: str | None = None, max_rows: int | None = None, ) -> tuple[pd.DataFrame, BaseMetadata]: - """Water-Data wrapper over :func:`engine.get_ogc_data`. + """Water-Data wrapper over :func:`~dataretrieval.ogc.get_ogc_data`. Defaults ``output_id`` from the Water Data service 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 ``engine.get_ogc_data`` directly with their own base URL and - dialect rather than going through this Water Data wrapper.) + NGWMN call ``dataretrieval.ogc.get_ogc_data`` directly with their own + base URL and dialect rather than going through this Water Data wrapper.) Parameters ---------- @@ -248,7 +216,7 @@ def get_ogc_data( """ if output_id is None: output_id = _OUTPUT_ID_BY_SERVICE[service] - return engine.get_ogc_data( + return _facade_get_ogc_data( args, service, output_id, @@ -277,7 +245,7 @@ def _finalize_ogc( :func:`~dataretrieval.ogc.shaping._finalize_ogc` for the full result-shaping contract. """ - return _engine_finalize_ogc( + return _ogc_shaping._finalize_ogc( frame, response, properties=properties, @@ -374,40 +342,16 @@ def wrapper(*args: Any, **kwargs: Any) -> _R: __all__ = [ "BASE_URL", - "GEOPANDAS", "OGC_API_URL", "SAMPLES_URL", "WATERDATA_DIALECT", - "_DATE_RANGE_PARAMS", - "_DURATION_RE", "_EXTRA_ID_COLS", "_NO_NORMALIZE_PARAMS", "_OUTPUT_ID_BY_SERVICE", "_accept_legacy_kwargs", - "_arrange_cols", - "_as_str_list", - "_check_monitoring_location_id", - "_check_ogc_requests", "_check_profiles", - "_construct_api_requests", - "_construct_cql_request", - "_deal_with_empty", - "_default_headers", - "_error_body", "_finalize_ogc", - "_format_api_dates", "_get_args", - "_get_resp_data", - "_next_req_url", - "_normalize_str_iterable", - "_paginate", - "_paginated_failure_message", - "_parse_retry_after", - "_raise_for_non_200", - "_row_cap", - "_run_sync", - "_switch_properties_id", - "_to_snake_case", - "_walk_pages", + "_with_state", "get_ogc_data", ] diff --git a/dataretrieval/wateruse.py b/dataretrieval/wateruse.py index 5dd5cb9c..53de7597 100644 --- a/dataretrieval/wateruse.py +++ b/dataretrieval/wateruse.py @@ -56,7 +56,7 @@ from dataretrieval.ogc.combining import _combine_chunk_frames, _combine_chunk_responses from dataretrieval.ogc.engine import _paginate, _run_sync from dataretrieval.utils import ( - HTTPX_DEFAULTS, + HTTPX_ASYNC_DEFAULTS, BaseMetadata, _default_headers, _raise_for_status, @@ -224,7 +224,7 @@ def get_wateruse( # The NWDC queries one location per request, so fan a multi-value selector # out into one request per location, each paginated by the OGC engine's # shared pager (``_paginate``), and concatenate the results. - headers = _default_headers() + headers = _default_headers(WATERUSE_URL) requests = [ httpx.Request( "GET", @@ -349,7 +349,7 @@ async def follow(cursor: str, sess: httpx.AsyncClient) -> httpx.Response: def raise_for_status(response: httpx.Response) -> None: _raise_for_status(response, detail_from=_nwdc_error_detail) - async with httpx.AsyncClient(verify=ssl_check, **HTTPX_DEFAULTS) as client: + async with httpx.AsyncClient(verify=ssl_check, **HTTPX_ASYNC_DEFAULTS) as client: semaphore = asyncio.Semaphore(max(1, MAX_CONCURRENT_REQUESTS)) async def _one(request: httpx.Request) -> tuple[pd.DataFrame, httpx.Response]: diff --git a/docs/source/architecture/decisions/0003-dependency-direction.rst b/docs/source/architecture/decisions/0003-dependency-direction.rst index 78094b58..3f7d8ec9 100644 --- a/docs/source/architecture/decisions/0003-dependency-direction.rst +++ b/docs/source/architecture/decisions/0003-dependency-direction.rst @@ -47,6 +47,11 @@ that hold today. Its allowlist is the authoritative inventory of exact temporary cross-boundary imports; this ADR owns the direction and rationale rather than a second copy of that mutable inventory. -The allowlist should shrink as private seams move. Any growth requires explicit -architecture review, and a change to the dependency policy requires this ADR to -be superseded. +Focused fitness functions verify the current boundaries: NGWMN's only OGC +dependency is the facade, ``waterdata.utils`` does not bulk re-export private +OGC helpers, ``ogc.shaping`` does not depend on ``ogc.engine``, and the full OGC +runtime graph is acyclic. + +The exact allowlist should shrink as private seams move. Any growth requires +explicit architecture review, and a change to the dependency policy requires +this ADR to be superseded. diff --git a/docs/source/architecture/index.rst b/docs/source/architecture/index.rst index 41b5670d..906c07a4 100644 --- a/docs/source/architecture/index.rst +++ b/docs/source/architecture/index.rst @@ -67,13 +67,16 @@ Public service facades ^^^^^^^^^^^^^^^^^^^^^^ ``dataretrieval.waterdata`` - Modern USGS Water Data API facade. Typed getters delegate to the shared OGC - subsystem, with separate modules for statistics, ratings, and nearest-value - operations. + Modern USGS Water Data API facade. Its generic adapter uses the four-symbol + OGC facade; internal Water Data modules import protocol helpers from their + canonical OGC modules. Water-Data-specific utilities own service policy and + wrappers without re-exporting private OGC helpers. Statistics, ratings, and + nearest-value operations live in separate modules. ``dataretrieval.ngwmn`` - NGWMN facade. Reuses the OGC subsystem with an NGWMN-specific base URL, - output identifiers, state translation, and :class:`OgcDialect`. + NGWMN facade. Its only OGC dependency is the public OGC facade, which it + configures with an NGWMN-specific base URL, output identifiers, state + translation, and :class:`OgcDialect`. ``dataretrieval.wateruse`` NWDC Water Use facade. Builds CSV requests and follows ``Link`` headers. @@ -93,12 +96,19 @@ Shared components ^^^^^^^^^^^^^^^^^ ``dataretrieval.ogc`` - Protocol subsystem for Water Data and NGWMN. ``engine`` orchestrates - requests and pagination; ``planning`` determines chunk boundaries; - ``chunking`` executes plans and retains resumable state; ``retry`` owns the - bounded retry policy; ``combining`` assembles results; and ``shaping``, - ``dates``, ``filters``, ``errors``, and ``progress`` isolate their named - concerns. + Protocol subsystem for Water Data and NGWMN. A small facade + (``__init__.py``) exposes the service-adapter seam: ``OgcDialect``, + ``prepare_request_args``, ``get_ogc_data``, and ``fetch_ogc_request``. + Internally, ``policy`` defines the dialect type and endpoint constants + (depends only on stdlib); ``requests`` owns request construction, argument + normalization, and queryables/schema lookup; ``engine`` orchestrates + pagination and sync-from-async; ``planning`` determines chunk boundaries; + ``chunking`` executes plans and retains resumable state; ``interruptions`` + defines the resumable failure contract; ``retry`` owns the bounded retry + policy; ``combining`` assembles results; and ``shaping``, ``dates``, + ``filters``, ``errors``, and ``progress`` isolate their named concerns. The + full runtime OGC graph, including the facade, is acyclic — + enforced by ``tests/architecture_test.py``. ``dataretrieval.exceptions`` Stable error-policy leaf. It has no runtime third-party dependency and may @@ -168,8 +178,10 @@ Resource and configuration view ------------------------------- ``API_USGS_PAT`` - Optional USGS API token. Authentication must be scoped to appropriate USGS - hosts and never forwarded to WQP or another unrelated endpoint. + Optional USGS API token. It is attached only to requests for + ``api.waterdata.usgs.gov``. Shared synchronous and asynchronous clients + re-check every redirected request and strip the token before following a + link to any other host, including external rating assets. ``API_USGS_CONCURRENT`` OGC subrequest concurrency cap; defaults to 32, ``1`` is sequential, and @@ -195,11 +207,8 @@ This view records categories and representative locations of debt. The fitness functions in ``tests/architecture_test.py`` are authoritative for exact current dependency allowlists. -- ``waterdata.utils`` re-exports many underscore-prefixed OGC helpers. - ``wateruse`` depends on private generic helpers located under ``ogc`` even though NWDC is not an OGC service. -- ``ogc.shaping`` uses lazy engine imports for empty-result schema lookup and a - default dialect, creating a logical cycle. - ``waterdata/api.py`` and ``ogc/engine.py`` contain multiple reasons to change. - Active non-OGC services do not yet share OGC's retry/resume capabilities. - ``utils.py`` combines metadata, shaping, configuration, and transport duties. diff --git a/tests/architecture_test.py b/tests/architecture_test.py index 3d3fd2be..8ba2d864 100644 --- a/tests/architecture_test.py +++ b/tests/architecture_test.py @@ -18,18 +18,40 @@ "dataretrieval.wqp", ) -# These top-level modules currently reach into OGC. NGWMN is an OGC adapter; -# Water Use's imports are an accepted temporary variance under ADR 0003. This -# allowlist is the authoritative exact inventory; the ADR owns the policy and -# rationale. Exact equality makes either growth or removal intentional. +# These top-level modules currently reach into OGC. NGWMN is an OGC adapter +# that uses the small facade (``dataretrieval.ogc``) exclusively. Water Use's +# imports are an accepted temporary variance under ADR 0003. This allowlist is +# the authoritative exact inventory; the ADR owns the policy and rationale. +# Exact equality makes either growth or removal intentional. _ALLOWED_TOP_LEVEL_OGC_IMPORTS = { - "dataretrieval.ngwmn": {"dataretrieval.ogc.engine"}, + "dataretrieval.ngwmn": { + "dataretrieval.ogc", + }, "dataretrieval.wateruse": { "dataretrieval.ogc.combining", "dataretrieval.ogc.engine", }, } +_ENGINE_REQUEST_IMPORTS = { + "_NO_NORMALIZE_PARAMS", + "_as_str_list", + "_check_monitoring_location_id", + "_check_ogc_requests", + "_construct_api_requests", + "_construct_cql_request", + "_cql2_param", + "_dialect", + "_get_args", + "_normalize_str_iterable", + "_ogc_base_url", + "_ogc_query_params", + "_row_cap", + "_switch_arg_id", + "_switch_properties_id", + "prepare_request_args", +} + def _module_name(path: Path) -> str: """Return the import name for one Python file below ``PACKAGE_ROOT``.""" @@ -166,3 +188,178 @@ def test_top_level_ogc_consumers_match_documented_variances() -> None: "policy changes.\n" f"expected={_ALLOWED_TOP_LEVEL_OGC_IMPORTS!r}\nobserved={observed!r}" ) + + +def test_engine_request_import_surface_is_frozen() -> None: + """Engine may preserve legacy request names but may not grow a new hub.""" + path = PACKAGE_ROOT / "ogc" / "engine.py" + tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) + imported = { + alias.name + for node in tree.body + if isinstance(node, ast.ImportFrom) + and node.module == "dataretrieval.ogc.requests" + for alias in node.names + } + assert imported == _ENGINE_REQUEST_IMPORTS, ( + "ogc.engine request imports changed; use the canonical requests module " + "instead of expanding compatibility exports.\n" + f"expected={sorted(_ENGINE_REQUEST_IMPORTS)}\nobserved={sorted(imported)}" + ) + + +# --- Strengthened OGC boundary tests --- + + +def test_ogc_runtime_graph_is_acyclic() -> None: + """The OGC runtime import graph (including the facade) has no cycles. + + Now that no implementation module imports ``dataretrieval.ogc`` (the facade + ``__init__.py``), the full OGC graph — facade included — forms a DAG. + This is enforced without any documented exclusion. + """ + ogc_modules: dict[str, set[str]] = {} + for module, imports in _package_import_graph().items(): + if module == "dataretrieval.ogc" or module.startswith("dataretrieval.ogc."): + # Filter to intra-OGC dependencies + ogc_deps = { + dep + for dep in imports + if dep == "dataretrieval.ogc" or dep.startswith("dataretrieval.ogc.") + } + ogc_modules[module] = ogc_deps + + # DFS cycle detection + WHITE, GRAY, BLACK = 0, 1, 2 + color: dict[str, int] = {m: WHITE for m in ogc_modules} + path: list[str] = [] + + def dfs(node: str) -> list[str] | None: + color[node] = GRAY + path.append(node) + for dep in ogc_modules.get(node, set()): + if dep not in color: + continue + if color[dep] == GRAY: + cycle_start = path.index(dep) + return path[cycle_start:] + [dep] + if color[dep] == WHITE: + result = dfs(dep) + if result: + return result + path.pop() + color[node] = BLACK + return None + + for module in ogc_modules: + if color[module] == WHITE: + cycle = dfs(module) + if cycle: + raise AssertionError( + f"Cycle in OGC runtime graph: {' -> '.join(cycle)}" + ) + + +def test_shaping_has_no_engine_dependency() -> None: + """ogc.shaping must not import ogc.engine, even lazily.""" + shaping_imports = _runtime_imports(PACKAGE_ROOT / "ogc" / "shaping.py") + engine_deps = {dep for dep in shaping_imports if dep == "dataretrieval.ogc.engine"} + assert not engine_deps, ( + f"ogc.shaping must not depend on ogc.engine. Found: {engine_deps}" + ) + + +def test_ngwmn_uses_ogc_facade() -> None: + """NGWMN must use ONLY the ogc facade, not engine or other internals.""" + ngwmn_imports = _runtime_imports(PACKAGE_ROOT / "ngwmn.py") + ogc_deps = { + dep + for dep in ngwmn_imports + if dep == "dataretrieval.ogc" or dep.startswith("dataretrieval.ogc.") + } + # Exact equality: the ONLY OGC dependency is the facade package itself. + assert ogc_deps == {"dataretrieval.ogc"}, ( + "NGWMN must use ONLY the OGC facade (dataretrieval.ogc), not internals. " + f"Found: {ogc_deps}" + ) + + +def test_waterdata_utils_is_not_an_ogc_reexport_hub() -> None: + """Water Data policy wrappers may not bulk re-export OGC internals.""" + path = PACKAGE_ROOT / "waterdata" / "utils.py" + ogc_deps = { + dependency + for dependency in _runtime_imports(path) + if dependency == "dataretrieval.ogc" + or dependency.startswith("dataretrieval.ogc.") + } + assert ogc_deps == { + "dataretrieval.ogc", + "dataretrieval.ogc.dates", + "dataretrieval.ogc.shaping", + }, f"Water Data utils crossed its intended OGC seam: {ogc_deps}" + + tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) + exports: set[str] | None = None + for node in tree.body: + if not isinstance(node, ast.Assign): + continue + if any( + isinstance(target, ast.Name) and target.id == "__all__" + for target in node.targets + ): + exports = set(ast.literal_eval(node.value)) + break + assert exports is not None, "waterdata.utils must declare its intentional exports" + + old_reexports = { + "GEOPANDAS", + "_as_str_list", + "_check_monitoring_location_id", + "_check_ogc_requests", + "_construct_api_requests", + "_construct_cql_request", + "_default_headers", + "_format_api_dates", + "_paginate", + "_raise_for_non_200", + "_run_sync", + "_switch_properties_id", + "_walk_pages", + "fetch_ogc_request", + } + assert exports.isdisjoint(old_reexports), ( + "waterdata.utils regained private OGC re-exports: " + f"{sorted(exports & old_reexports)}" + ) + + +def test_default_header_calls_are_target_scoped() -> None: + """Every production header construction must name the destination URL.""" + violations: list[str] = [] + for path in sorted(PACKAGE_ROOT.rglob("*.py")): + tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) + for node in ast.walk(tree): + if not isinstance(node, ast.Call): + continue + function_name = ( + node.func.id + if isinstance(node.func, ast.Name) + else node.func.attr + if isinstance(node.func, ast.Attribute) + else None + ) + if function_name != "_default_headers": + continue + has_target = bool(node.args) or any( + keyword.arg == "target_url" for keyword in node.keywords + ) + if not has_target: + violations.append( + f"{path.relative_to(PACKAGE_ROOT.parent)}:{node.lineno}" + ) + + assert not violations, ( + "_default_headers calls without destination URL context:\n" + + "\n".join(violations) + ) diff --git a/tests/headers_host_scoping_test.py b/tests/headers_host_scoping_test.py new file mode 100644 index 00000000..85dfc270 --- /dev/null +++ b/tests/headers_host_scoping_test.py @@ -0,0 +1,160 @@ +"""Tests for API-key host scoping in _default_headers.""" + +from __future__ import annotations + +import asyncio +from unittest import mock + +import httpx +import pytest + +from dataretrieval.utils import ( + HTTPX_ASYNC_DEFAULTS, + _default_headers, + _get, +) + + +class TestDefaultHeadersHostScoping: + """_default_headers only sends X-Api-Key to the authorized Water Data host.""" + + FAKE_TOKEN = "test-fake-token-abc123" + + @pytest.fixture(autouse=True) + def _api_token(self, monkeypatch: pytest.MonkeyPatch) -> None: + """Install one harmless token for every host-scoping behavior test.""" + monkeypatch.setenv("API_USGS_PAT", self.FAKE_TOKEN) + + def test_key_included_for_waterdata_host(self): + """Key IS added when target URL matches api.waterdata.usgs.gov.""" + url = "https://api.waterdata.usgs.gov/ogcapi/v0/collections/daily/items" + headers = _default_headers(url) + assert headers.get("X-Api-Key") == self.FAKE_TOKEN + + def test_key_excluded_for_external_host(self): + """Key is NOT added for an external (non-USGS) host.""" + url = "https://nwis.waterservices.usgs.gov/nwis/iv/" + headers = _default_headers(url) + assert "X-Api-Key" not in headers + + def test_key_excluded_for_wateruse_host(self): + """Key is NOT added for the NWDC water-use host (api.water.usgs.gov).""" + url = "https://api.water.usgs.gov/nwaa-data/data" + headers = _default_headers(url) + assert "X-Api-Key" not in headers + + def test_key_excluded_for_rating_asset_host(self): + """Key is NOT added for rating asset downloads (S3/external).""" + url = "https://labs.waterdata.usgs.gov/sta/v1.1/Datastreams(123)/rating.rdb" + headers = _default_headers(url) + assert "X-Api-Key" not in headers + + def test_key_excluded_for_lookalike_host(self): + """Key is NOT sent to a typosquatting/lookalike domain.""" + url = "https://api.waterdata.usgs.gov.evil.com/ogcapi/v0/daily/items" + headers = _default_headers(url) + assert "X-Api-Key" not in headers + + def test_key_excluded_when_no_url_provided(self): + """Key is NOT added when target_url is None (legacy callers).""" + headers = _default_headers(None) + assert "X-Api-Key" not in headers + + def test_key_excluded_when_no_token_set( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + """No key header at all when API_USGS_PAT is not set.""" + monkeypatch.delenv("API_USGS_PAT") + headers = _default_headers("https://api.waterdata.usgs.gov/ogcapi/v0/daily") + assert "X-Api-Key" not in headers + + def test_non_auth_headers_always_present(self): + """User-Agent, Accept, Accept-Encoding, lang are always present.""" + url = "https://example.com/any" + headers = _default_headers(url) + assert "User-Agent" in headers + assert "Accept" in headers + assert "Accept-Encoding" in headers + assert "lang" in headers + # Key should NOT be sent to example.com + assert "X-Api-Key" not in headers + + def test_generic_ogc_request_excludes_key_for_custom_host(self): + """A caller-supplied OGC base URL never inherits Water Data auth.""" + from dataretrieval.ogc.requests import _construct_api_requests, _ogc_base_url + + with _ogc_base_url("https://features.example.org/ogcapi"): + request = _construct_api_requests("things") + assert "X-Api-Key" not in request.headers + + def test_rating_download_scopes_headers_to_asset_url(self): + """The ratings adapter evaluates auth against each asset href.""" + import pandas as pd + + import dataretrieval.waterdata.ratings as ratings + + asset_url = "https://objects.example.org/ratings/site.rdb" + feature = {"id": "site.rdb", "assets": {"data": {"href": asset_url}}} + response = mock.Mock(text="rating body") + with ( + mock.patch.object(ratings, "_get", return_value=response) as get, + mock.patch.object(ratings, "_raise_for_non_200"), + mock.patch.object(ratings, "read_rdb", return_value=pd.DataFrame()), + mock.patch.object(ratings, "extract_rdb_comment", return_value=""), + ): + ratings._download_and_parse(feature, file_path=None, ssl_check=True) + + assert get.call_args.args[0] == asset_url + assert "X-Api-Key" not in get.call_args.kwargs["headers"] + + def test_sync_redirect_strips_key_before_cross_host_request(self): + """The synchronous transport guard runs again for redirects.""" + seen: list[httpx.Request] = [] + + def handler(request: httpx.Request) -> httpx.Response: + seen.append(request) + if len(seen) == 1: + return httpx.Response( + 302, + headers={"Location": "https://outside.example.org/next"}, + request=request, + ) + return httpx.Response(200, request=request) + + url = "https://api.waterdata.usgs.gov/start" + _get( + url, + headers=_default_headers(url), + follow_redirects=True, + transport=httpx.MockTransport(handler), + ) + + assert seen[0].headers.get("X-Api-Key") == self.FAKE_TOKEN + assert "X-Api-Key" not in seen[1].headers + + def test_async_redirect_strips_key_before_cross_host_request(self): + """The shared asynchronous client policy guards redirects too.""" + seen: list[httpx.Request] = [] + + def handler(request: httpx.Request) -> httpx.Response: + seen.append(request) + if len(seen) == 1: + return httpx.Response( + 302, + headers={"Location": "https://outside.example.org/next"}, + request=request, + ) + return httpx.Response(200, request=request) + + async def run() -> None: + url = "https://api.waterdata.usgs.gov/start" + async with httpx.AsyncClient( + transport=httpx.MockTransport(handler), + **HTTPX_ASYNC_DEFAULTS, + ) as client: + await client.get(url, headers=_default_headers(url)) + + asyncio.run(run()) + + assert seen[0].headers.get("X-Api-Key") == self.FAKE_TOKEN + assert "X-Api-Key" not in seen[1].headers diff --git a/tests/ngwmn_test.py b/tests/ngwmn_test.py index b2695425..8686267f 100644 --- a/tests/ngwmn_test.py +++ b/tests/ngwmn_test.py @@ -106,9 +106,9 @@ def test_state_queryables_still_diverge_upstream(): import httpx from dataretrieval.ngwmn import NGWMN_OGC_API_URL - from dataretrieval.ogc.engine import _default_headers + from dataretrieval.utils import _default_headers - headers = _default_headers() + headers = _default_headers(NGWMN_OGC_API_URL) def queryables(collection): resp = httpx.get( diff --git a/tests/waterdata_chunking_test.py b/tests/waterdata_chunking_test.py index 4a86bf17..ddefafcc 100644 --- a/tests/waterdata_chunking_test.py +++ b/tests/waterdata_chunking_test.py @@ -39,6 +39,7 @@ Unchunkable, ) from dataretrieval.ogc import chunking as _chunking +from dataretrieval.ogc import engine as _engine from dataretrieval.ogc import retry as _retry_mod from dataretrieval.ogc.chunking import ( ChunkedCall, @@ -53,6 +54,7 @@ _combine_chunk_frames, _combine_chunk_responses, ) +from dataretrieval.ogc.dates import _DATE_RANGE_PARAMS from dataretrieval.ogc.interruptions import ( ChunkInterrupted, QuotaExhausted, @@ -67,6 +69,7 @@ _request_bytes, _safe_request_bytes, ) +from dataretrieval.ogc.requests import _construct_api_requests from dataretrieval.ogc.retry import ( _RETRIES_DEFAULT, RetryPolicy, @@ -74,8 +77,6 @@ _retryable, ) from dataretrieval.utils import HTTPX_DEFAULTS -from dataretrieval.waterdata import utils as _utils -from dataretrieval.waterdata.utils import _DATE_RANGE_PARAMS, _construct_api_requests def _aiozero(_d): @@ -1102,7 +1103,7 @@ def test_paginate_terminates_on_empty_string_cursor(): req.content = b"" req.url = "https://example.com/items?limit=1" - df, _ = asyncio.run(_utils._walk_pages(geopd=False, req=req, client=client)) + df, _ = asyncio.run(_engine._walk_pages(geopd=False, req=req, client=client)) # Single send + zero follow-ups: the loop terminated on the empty cursor. assert client.send.called diff --git a/tests/waterdata_filters_test.py b/tests/waterdata_filters_test.py index 6d5f9310..207a1dcf 100644 --- a/tests/waterdata_filters_test.py +++ b/tests/waterdata_filters_test.py @@ -11,8 +11,8 @@ _quote_cql_str, _split_top_level_or, ) +from dataretrieval.ogc.requests import _construct_api_requests from dataretrieval.waterdata import get_continuous -from dataretrieval.waterdata.utils import _construct_api_requests def _query_params(prepared_request): diff --git a/tests/waterdata_progress_test.py b/tests/waterdata_progress_test.py index 36c45d8d..dc752591 100644 --- a/tests/waterdata_progress_test.py +++ b/tests/waterdata_progress_test.py @@ -19,13 +19,13 @@ from dataretrieval.ogc import progress as _progress from dataretrieval.ogc.chunking import ChunkedCall +from dataretrieval.ogc.engine import _paginate, _walk_pages from dataretrieval.ogc.planning import ChunkPlan from dataretrieval.ogc.progress import ( ProgressReporter, current, progress_context, ) -from dataretrieval.waterdata.utils import _paginate, _walk_pages def _run_walk_pages(*, geopd, req, client): diff --git a/tests/waterdata_test.py b/tests/waterdata_test.py index 907232d0..df317244 100644 --- a/tests/waterdata_test.py +++ b/tests/waterdata_test.py @@ -8,6 +8,12 @@ from pandas import DataFrame from dataretrieval.ogc.engine import _dialect +from dataretrieval.ogc.requests import ( + _check_monitoring_location_id, + _construct_api_requests, + _construct_cql_request, + _normalize_str_iterable, +) from dataretrieval.waterdata import ( get_channel, get_combined_metadata, @@ -29,12 +35,8 @@ ) from dataretrieval.waterdata.utils import ( WATERDATA_DIALECT, - _check_monitoring_location_id, _check_profiles, - _construct_api_requests, - _construct_cql_request, _get_args, - _normalize_str_iterable, ) from tests.conftest import flaky_api diff --git a/tests/waterdata_utils_test.py b/tests/waterdata_utils_test.py index 431cf7f4..c39a8b19 100644 --- a/tests/waterdata_utils_test.py +++ b/tests/waterdata_utils_test.py @@ -8,7 +8,6 @@ import pandas as pd import pytest -import dataretrieval.ogc.engine as _engine_module import dataretrieval.ogc.shaping as _shaping_module import dataretrieval.waterdata.stats as _stats_module import dataretrieval.waterdata.utils as _utils_module @@ -19,24 +18,25 @@ ServiceUnavailable, TransientError, ) -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 ( - OGC_API_URL, - _arrange_cols, - _check_ogc_requests, - _error_body, - _finalize_ogc, - _format_api_dates, - _get_args, - _get_resp_data, +from dataretrieval.ogc.dates import _format_api_dates +from dataretrieval.ogc.engine import ( _next_req_url, + _walk_pages, +) +from dataretrieval.ogc.errors import ( + _error_body, _parse_retry_after, _raise_for_non_200, - _row_cap, +) +from dataretrieval.ogc.requests import _check_ogc_requests, _row_cap +from dataretrieval.ogc.shaping import ( + _arrange_cols, + _get_resp_data, _to_snake_case, - _walk_pages, ) +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 OGC_API_URL, _finalize_ogc, _get_args _LOGGER_NAME = _utils_module.__name__ @@ -1080,6 +1080,8 @@ def fake_engine_get_ogc_data(args, service, output_id, **k): captured["args"] = dict(args) return pd.DataFrame(), mock.Mock() - with mock.patch.object(_engine_module, "get_ogc_data", fake_engine_get_ogc_data): + with mock.patch.object( + _utils_module, "_facade_get_ogc_data", fake_engine_get_ogc_data + ): _utils_module.get_ogc_data({"state": "WI"}, "monitoring-locations") assert captured["args"] == {"state": "WI"}