refactor: one home for option validation; retire dead engine wrappers

.importlinter

@@ -33,6 +33,10 @@ layers =
     _ambient | _response_metadata | codes | combining | interruptions | rdb
     credentials
     exceptions
+; Argument validation sits at the floor: it raises the stdlib's ``ValueError``
+; and imports nothing first-party, so every layer above can reject a bad
+; option without reaching sideways for a helper.
+    _validation
 ; Every top-level module must be placed in the stack deliberately. A new
 ; top-level module fails this contract until someone decides where it sits.
 exhaustive = True

dataretrieval/_validation.py

@@ -0,0 +1,71 @@
+"""Argument checks shared by every adapter.
+
+Rejecting a value that is not in a closed vocabulary is the one validation
+every adapter does, and it was written eleven times: eight message phrasings
+for one concept, so each new check was a coin flip on wording. That is how
+:func:`~dataretrieval.waterdata.get_reference_table` came to tell callers who
+passed a bad ``collection`` that their *code service* was invalid -- the check
+was copied from :mod:`~dataretrieval.waterdata.samples`, message and local
+variable name included, and the noun was never changed.
+
+This module owns the wording so a new check cannot invent its own. It is a
+leaf with no first-party imports: the vocabularies it validates against live
+with the adapters that define them, and only the rejection is shared.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Collection
+
+
+def _render(options: Collection[object]) -> str:
+    """Format *options* for a message: ``'a', 'b', 'c'``.
+
+    Renders the values rather than their container so ``dict_keys([...])`` and
+    a bare tuple read the same to a caller, who never sees the container.
+    """
+    return ", ".join(repr(option) for option in options)
+
+
+def require_one_of(
+    value: object,
+    options: Collection[object],
+    *,
+    name: str,
+    context: str = "",
+) -> None:
+    """Raise ``ValueError`` unless *value* is one of *options*.
+
+    Parameters
+    ----------
+    value
+        The argument the caller supplied.
+    options
+        The closed vocabulary it must belong to -- typically
+        ``get_args(SomeLiteral)``, a module constant, or a mapping's keys.
+        Rendered in iteration order, so pass a sorted view when the source is
+        unordered and the order would otherwise be arbitrary.
+    name
+        What the value *is*, as the caller's parameter names it (``"service"``,
+        ``"collection"``). It becomes the message's subject, so it must match
+        the parameter the caller actually passed.
+    context
+        Optional qualifier for a vocabulary that depends on another argument,
+        e.g. ``context="service 'wqp'"`` when the valid profiles differ per
+        service.
+
+    Raises
+    ------
+    ValueError
+        If *value* is not in *options*.
+    """
+    if isinstance(options, str):
+        # ``str`` is a Collection, so this type-checks -- and then ``in``
+        # silently means "substring", accepting any fragment of a valid option.
+        raise TypeError(f"options must be a collection of values, not {options!r}")
+    if value in options:
+        return
+    qualifier = f" for {context}" if context else ""
+    raise ValueError(
+        f"Invalid {name}: {value!r}{qualifier}. Valid options are: {_render(options)}."
+    )

dataretrieval/nldi.py

@@ -14,6 +14,7 @@
 from typing import Any, Literal, cast
 
 from dataretrieval._querying import _query_with_retry
+from dataretrieval._validation import require_one_of
 
 __all__ = [
     "get_flowlines",
@@ -473,11 +474,7 @@ def search(
         raise ValueError("Both lat and long are required")
 
     find = cast("Literal['basin', 'flowlines', 'features']", find.lower())
-    if find not in ("basin", "flowlines", "features"):
-        raise ValueError(
-            f"Invalid value for find: {find} - allowed values are:"
-            f" 'basin', 'flowlines', or 'features'"
-        )
+    require_one_of(find, ("basin", "flowlines", "features"), name="find")
     if lat is not None and find != "features":
         raise ValueError(
             f"Invalid value for find: {find} - lat/long is to get features not {find}"
@@ -559,11 +556,7 @@ def _validate_navigation_mode(navigation_mode: str | None) -> str:
             f"navigation_mode is required; allowed values are {_VALID_NAVIGATION_MODES}"
         )
     normalized = navigation_mode.upper()
-    if normalized not in _VALID_NAVIGATION_MODES:
-        raise ValueError(
-            f"Invalid navigation mode '{navigation_mode}';"
-            f" allowed values are {_VALID_NAVIGATION_MODES}"
-        )
+    require_one_of(normalized, _VALID_NAVIGATION_MODES, name="navigation_mode")
     return normalized
 
 

dataretrieval/ogc/__init__.py

@@ -10,8 +10,7 @@
 Collection adapters (NGWMN, Water Data's generic wrapper) import from this
 facade rather than reaching into engine internals — every name here is usable
 through the facade alone. Generic execution policy lives in
-:mod:`dataretrieval.transport`; the engine retains compatibility wrappers at
-previous private paths.
+:mod:`dataretrieval.transport`, which the engine now calls directly.
 """
 
 from dataretrieval.ogc.engine import get_ogc_data

dataretrieval/ogc/engine.py

@@ -27,10 +27,9 @@
 import functools
 import logging
 from collections.abc import (
-    Awaitable,
     Callable,
 )
-from typing import TYPE_CHECKING, Any, TypeVar
+from typing import TYPE_CHECKING, Any
 
 import httpx
 import pandas as pd
@@ -52,7 +51,7 @@
     _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
 from dataretrieval.transport.links import resolve_next_url
 from dataretrieval.transport.pagination import paginate
 from dataretrieval.transport.retry import RetryPolicy
@@ -63,9 +62,6 @@
 # Set up logger for this module
 logger = logging.getLogger(__name__)
 
-# Compatibility alias: the old name used internally and in tests.
-_DEFAULT_DIALECT = DEFAULT_DIALECT
-
 
 def _next_req_url(
     resp: httpx.Response, *, body: dict[str, Any] | None = None
@@ -120,39 +116,16 @@ def _next_req_url(
     return None
 
 
-_Cursor = TypeVar("_Cursor")
-
-
-async def _paginate(
-    initial_req: httpx.Request,
-    *,
-    parse_response: Callable[[httpx.Response], tuple[pd.DataFrame, _Cursor | None]],
-    follow_up: Callable[[_Cursor, httpx.AsyncClient], Awaitable[httpx.Response]],
-    client: httpx.AsyncClient | None = None,
-    raise_for_status: Callable[[httpx.Response], None] = _raise_for_non_200,
-    row_cap: int | None = None,
-) -> tuple[pd.DataFrame, httpx.Response]:
-    """Compatibility wrapper around collection-neutral cursor pagination."""
-    session = client if client is not None else active_client()
-    return await paginate(
-        initial_req,
-        parse_response=parse_response,
-        follow_up=follow_up,
-        client=session,
-        raise_for_status=raise_for_status,
-        row_cap=row_cap,
-    )
-
-
 def _ogc_parse_response(
     resp: httpx.Response, *, geopd: bool
 ) -> tuple[pd.DataFrame, str | None]:
     """Parse one OGC API page: extract the DataFrame and the next-page URL.
 
     The parse strategy :func:`_walk_pages` hands to
-    :func:`_paginate`. Coerces falsy cursors (empty href, etc.) to
-    ``None`` so the paginate loop's ``while cursor is not None``
-    terminates instead of spinning on a meaningless value.
+    :func:`~dataretrieval.transport.pagination.paginate`. Coerces falsy
+    cursors (empty href, etc.) to ``None`` so the paginate loop's
+    ``while cursor is not None`` terminates instead of spinning on a
+    meaningless value.
     """
     body = resp.json()
     return (
@@ -171,7 +144,8 @@ async def _walk_pages(
     """
     Iterate paginated OGC API responses and aggregate them into one DataFrame.
 
-    Thin wrapper that hands off to :func:`_paginate` with
+    Thin wrapper that hands off to
+    :func:`~dataretrieval.transport.pagination.paginate` with
     OGC-specific strategies: pages are parsed via :func:`_get_resp_data`
     (through :func:`_ogc_parse_response`) and the next-page cursor is the
     URL from the response's ``links`` array (per :func:`_next_req_url`).
@@ -184,7 +158,7 @@ async def _walk_pages(
         The initial HTTP request to send.
     client : httpx.AsyncClient, optional
         Caller-borrowed client; ``None`` defers client management to
-        :func:`_paginate`.
+        :func:`~dataretrieval.transport.pagination.paginate`.
     row_cap : int, optional
         Stop following pages once this many rows have accumulated and
         truncate to exactly this many. ``None`` (default) walks every page.
@@ -203,9 +177,9 @@ async def _walk_pages(
     Raises
     ------
     DataRetrievalError
-        See :func:`_paginate`.
+        See :func:`~dataretrieval.transport.pagination.paginate`.
     httpx.HTTPError
-        See :func:`_paginate`.
+        See :func:`~dataretrieval.transport.pagination.paginate`.
     """
     method = req.method  # ``httpx.Request.method`` is already upper-cased.
     headers = req.headers
@@ -214,11 +188,12 @@ async def _walk_pages(
     async def follow_up(cursor: str, sess: httpx.AsyncClient) -> httpx.Response:
         return await sess.request(method, cursor, headers=headers, content=content)
 
-    return await _paginate(
+    return await paginate(
         req,
         parse_response=functools.partial(_ogc_parse_response, geopd=geopd),
         follow_up=follow_up,
         client=client,
+        raise_for_status=_raise_for_non_200,
         row_cap=row_cap,
     )
 
@@ -301,7 +276,7 @@ def get_ogc_data(
         _require_positive_int(max_rows, "max_rows")
 
     if dialect is None:
-        dialect = _DEFAULT_DIALECT
+        dialect = DEFAULT_DIALECT
     args = args.copy()
     args["collection"] = collection
     args = _switch_arg_id(args, id_name=output_id, collection=collection)

dataretrieval/ogc/requests.py

@@ -332,7 +332,3 @@ def prepare_request_args(
         else:
             args[k] = _normalize_str_iterable(v, k)
     return args
-
-
-# Compatibility alias for existing private imports from ``ogc.engine``.
-_get_args = prepare_request_args

dataretrieval/ogc/schema.py

@@ -13,6 +13,7 @@
 import pandas as pd
 
 from dataretrieval._response_metadata import BaseMetadata
+from dataretrieval._validation import require_one_of
 from dataretrieval.ogc.errors import _raise_for_non_200
 from dataretrieval.transport.http import HTTPX_DEFAULTS
 from dataretrieval.transport.http import default_headers as _default_headers
@@ -27,8 +28,7 @@ def _check_ogc_requests(
     ``base_url`` names the API to ask; it defaults to the one in scope for the
     current call rather than to any particular collection.
     """
-    if req_type not in ("queryables", "schema"):
-        raise ValueError(f"req_type must be 'queryables' or 'schema', got {req_type!r}")
+    require_one_of(req_type, ("queryables", "schema"), name="req_type")
     url = f"{base_url}/collections/{endpoint}/{req_type}"
     response = _get(url, headers=_default_headers(url), **HTTPX_DEFAULTS)
     _raise_for_non_200(response)

dataretrieval/transport/pagination.py

@@ -42,9 +42,15 @@
 async def _client_for(
     client: httpx.AsyncClient | None,
 ) -> AsyncIterator[httpx.AsyncClient]:
-    """Borrow a caller client or open a guarded short-lived client."""
-    if client is not None:
-        yield client
+    """Borrow a client: the caller's, else the running drive's, else a new one.
+
+    Preferring the executor's published client over a fresh one keeps every
+    page of every request on one connection pool. Both callers wanted that and
+    each spelled it itself before it moved here.
+    """
+    borrowed = client if client is not None else active_client()
+    if borrowed is not None:
+        yield borrowed
         return
     async with open_async_client() as new:
         yield new
@@ -177,7 +183,7 @@ async def fetch(request: httpx.Request) -> tuple[pd.DataFrame, httpx.Response]:
             request,
             parse_response=parse_response,
             follow_up=follow_up,
-            client=client if client is not None else active_client(),
+            client=client,
             raise_for_status=raise_for_status,
         )
 

dataretrieval/utils.py

@@ -30,7 +30,6 @@
 USER_AGENT = _transport_http.USER_AGENT
 _default_headers = _transport_http.default_headers
 _get = _transport_http.get
-_network_error = _transport_http.network_error
 # Public functions whose implementation moved to the private query module; this
 # is the path they are documented at.
 query = _querying.query

dataretrieval/waterdata/cql.py

@@ -14,6 +14,7 @@
 
 import pandas as pd
 
+from dataretrieval._validation import require_one_of
 from dataretrieval.waterdata.utils import (
     _OUTPUT_ID_BY_COLLECTION,
     _accept_legacy_kwargs,
@@ -136,11 +137,7 @@ def get_cql(
         ...     ' "02070010%"]}',
         ... )
     """
-    if collection not in _OUTPUT_ID_BY_COLLECTION:
-        raise ValueError(
-            f"Unknown collection {collection!r}. Valid collections: "
-            f"{sorted(_OUTPUT_ID_BY_COLLECTION)}."
-        )
+    require_one_of(collection, sorted(_OUTPUT_ID_BY_COLLECTION), name="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

dataretrieval/waterdata/nearest.py

@@ -10,6 +10,7 @@
 
 import pandas as pd
 
+from dataretrieval._validation import require_one_of
 from dataretrieval.waterdata.time_series import get_continuous
 
 if TYPE_CHECKING:
@@ -205,8 +206,7 @@ def _check_nearest_kwargs(kwargs: dict[str, Any], on_tie: OnTie) -> None:
                 f"get_nearest_continuous constructs its own {forbidden!r}; "
                 "do not pass it directly"
             )
-    if on_tie not in _VALID_ON_TIE:
-        raise ValueError(f"on_tie must be one of {_VALID_ON_TIE}; got {on_tie!r}")
+    require_one_of(on_tie, _VALID_ON_TIE, name="on_tie")
 
 
 def _build_window_or_filter(targets: pd.DatetimeIndex, window_td: pd.Timedelta) -> str:

dataretrieval/waterdata/reference.py

@@ -12,6 +12,7 @@
 
 import pandas as pd
 
+from dataretrieval._validation import require_one_of
 from dataretrieval.ogc.schema import queryables_frame
 from dataretrieval.waterdata.types import (
     METADATA_COLLECTIONS,
@@ -93,12 +94,7 @@ def get_reference_table(
         ...     query={"id": "00001,00002"},
         ... )
     """
-    valid_code_services = get_args(METADATA_COLLECTIONS)
-    if collection not in valid_code_services:
-        raise ValueError(
-            f"Invalid code service: '{collection}'. "
-            f"Valid options are: {valid_code_services}."
-        )
+    require_one_of(collection, get_args(METADATA_COLLECTIONS), name="collection")
 
     # Give the ID column the collection name, singularized and underscored.
     if collection == "counties":