feat(waterdata): add get_waterdata for generalized CQL2 queries

Python analogue of R ``dataRetrieval::read_waterdata``. The typed
``get_*`` wrappers (``get_daily``, ``get_continuous``, …) only support
exact-equality predicates on whitelisted parameters. Some users need
more — top-level ``or``, ``like`` with ``%`` wildcards, comparison
operators, nested boolean trees — and today have no surface for it.
``get_waterdata(service, cql, ...)`` accepts a raw CQL2 query
(``dict`` or pre-serialized JSON string) and POSTs it against any
recognized collection, then walks pages and post-processes the
result with the same pipeline the typed wrappers use.

Reuses existing infrastructure: ``_walk_pages``, ``_deal_with_empty``,
``_arrange_cols``, ``_type_cols``, ``_sort_rows``, and
``_switch_properties_id``. The new pieces are:

  - ``_OUTPUT_ID_BY_SERVICE`` (utils.py) — a single mapping from
    service name to the renamed-``id`` column the rest of the package
    exposes, hoisted from the typed wrappers so the generalized entry
    point can pick the right one.
  - ``_construct_cql_request`` (utils.py) — focused POST/CQL2 request
    builder; distinct from ``_construct_api_requests`` because the
    body comes in verbatim rather than being derived from typed
    kwargs.
  - ``get_waterdata`` (api.py) — public entry point.

CQL2 grammar reference:
https://api.waterdata.usgs.gov/docs/ogcapi/complex-queries/

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: thodson-usgs

dataretrieval/waterdata/__init__.py

@@ -28,6 +28,7 @@
     get_stats_date_range,
     get_stats_por,
     get_time_series_metadata,
+    get_waterdata,
 )
 from .filters import FILTER_LANG
 from .nearest import get_nearest_continuous
@@ -64,4 +65,5 @@
     "get_stats_date_range",
     "get_stats_por",
     "get_time_series_metadata",
+    "get_waterdata",
 ]

dataretrieval/waterdata/api.py

@@ -26,10 +26,20 @@
     SERVICES,
 )
 from dataretrieval.waterdata.utils import (
+    _OUTPUT_ID_BY_SERVICE,
+    GEOPANDAS,
     SAMPLES_URL,
+    _arrange_cols,
     _check_profiles,
+    _construct_cql_request,
+    _deal_with_empty,
     _default_headers,
     _get_args,
+    _normalize_str_iterable,
+    _sort_rows,
+    _switch_properties_id,
+    _type_cols,
+    _walk_pages,
     get_ogc_data,
     get_stats_data,
 )
@@ -2832,3 +2842,153 @@ def get_channel(
     args = _get_args(locals())
 
     return get_ogc_data(args, output_id, service)
+
+
+def get_waterdata(
+    service: str,
+    cql: str | dict,
+    *,
+    properties: str | Iterable[str] | None = None,
+    bbox: list[float] | None = None,
+    limit: int | None = None,
+    skip_geometry: bool | None = None,
+    convert_type: bool = True,
+    client: requests.Session | None = None,
+) -> tuple[pd.DataFrame, BaseMetadata]:
+    """Generalized OGC API CQL2 query.
+
+    Python analogue of R ``dataRetrieval::read_waterdata``. Use this
+    when you need a predicate the typed wrappers (``get_daily``,
+    ``get_continuous``, …) can't express — top-level ``or``, ``like``
+    with ``%`` wildcards, comparison operators, nested boolean trees,
+    geometry-based predicates beyond a bbox, and so on. The typed
+    wrappers are nicer when they cover the case; reach for this when
+    they don't.
+
+    The CQL2 grammar is documented at
+    https://api.waterdata.usgs.gov/docs/ogcapi/complex-queries/.
+
+    Parameters
+    ----------
+    service : str
+        OGC collection name (e.g. ``"daily"``, ``"monitoring-locations"``).
+        See :data:`dataretrieval.waterdata.utils._OUTPUT_ID_BY_SERVICE`
+        for the recognized list.
+    cql : str or dict
+        CQL2 query. A ``dict`` is JSON-serialized for transport; a
+        ``str`` is sent through unchanged. The query goes into the
+        HTTP POST body with ``Content-Type:
+        application/query-cql-json``.
+    properties : str or iterable of str, optional
+        Server-side property whitelist (passed as ``properties=`` on
+        the URL). Reduces payload size and bypasses the response-shape
+        post-processing for any column not listed. ``"id"`` resolves
+        to the service's ``output_id`` (e.g. ``daily_id``) the same way
+        it does in the typed wrappers.
+    bbox : list of float, optional
+        Bounding box ``[xmin, ymin, xmax, ymax]`` in CRS 4326.
+    limit : int, optional
+        Page size, clamped server-side to 50,000.
+    skip_geometry : bool, optional
+        If True, the server omits geometry from each feature
+        (``skipGeometry=true``).
+    convert_type : bool, default True
+        Coerce date/datetime/numeric columns to typed dtypes after the
+        DataFrame is built.
+    client : requests.Session, optional
+        Reuse an existing HTTP session (handy for batching or
+        injecting custom retry/auth adapters). A short-lived session
+        is created internally if not provided.
+
+    Returns
+    -------
+    df : pandas.DataFrame or geopandas.GeoDataFrame
+        Result of the query. GeoDataFrame when ``geopandas`` is
+        installed and geometry is present.
+    md : :class:`dataretrieval.utils.BaseMetadata`
+        Request metadata (URL, query time, response headers).
+
+    Examples
+    --------
+    .. code::
+
+        >>> # Daily values for two parameter codes at two sites
+        >>> # (compound AND-of-INs).
+        >>> from dataretrieval import waterdata
+        >>> cql = {
+        ...     "op": "and",
+        ...     "args": [
+        ...         {
+        ...             "op": "in",
+        ...             "args": [
+        ...                 {"property": "parameter_code"},
+        ...                 ["00060", "00065"],
+        ...             ],
+        ...         },
+        ...         {
+        ...             "op": "in",
+        ...             "args": [
+        ...                 {"property": "monitoring_location_id"},
+        ...                 ["USGS-07367300", "USGS-03277200"],
+        ...             ],
+        ...         },
+        ...     ],
+        ... }
+        >>> df, md = waterdata.get_waterdata(
+        ...     service="daily",
+        ...     cql=cql,
+        ...     time=("2023-01-01", "2024-01-01"),
+        ... )
+
+        >>> # Monitoring locations whose HUC starts with "02070010"
+        >>> # (LIKE with the CQL2 ``%`` wildcard).
+        >>> df, md = waterdata.get_waterdata(
+        ...     service="monitoring-locations",
+        ...     cql='{"op": "like", "args": ['
+        ...     '{"property": "hydrologic_unit_code"},'
+        ...     ' "02070010%"]}',
+        ... )
+    """
+    if service not in _OUTPUT_ID_BY_SERVICE:
+        raise ValueError(
+            f"Unknown service {service!r}. Valid services: "
+            f"{sorted(_OUTPUT_ID_BY_SERVICE)}."
+        )
+    output_id = _OUTPUT_ID_BY_SERVICE[service]
+
+    # ``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 config file) don't need to re-parse it.
+    body = json.dumps(cql, separators=(",", ":")) if isinstance(cql, dict) else cql
+
+    if isinstance(properties, str):
+        properties_list = [properties]
+    elif properties is None:
+        properties_list = None
+    else:
+        properties_list = _normalize_str_iterable(properties, "properties")
+
+    # Translate user-facing names (``daily_id``) to the wire-format
+    # ``id`` the OGC API expects, matching the typed wrappers.
+    wire_properties = _switch_properties_id(
+        properties_list, id_name=output_id, service=service
+    )
+
+    req = _construct_cql_request(
+        service=service,
+        cql_body=body,
+        properties=wire_properties or None,
+        bbox=bbox,
+        limit=limit,
+        skip_geometry=skip_geometry,
+    )
+
+    df, response = _walk_pages(geopd=GEOPANDAS, req=req, client=client)
+
+    df = _deal_with_empty(df, properties_list, service)
+    if convert_type:
+        df = _type_cols(df)
+    df = _arrange_cols(df, properties_list, output_id)
+    df = _sort_rows(df)
+
+    return df, BaseMetadata(response)

dataretrieval/waterdata/utils.py

@@ -157,6 +157,26 @@ def _switch_properties_id(properties: list[str] | None, id_name: str, service: s
 # parameters and require POST with CQL2 JSON instead.
 _CQL2_REQUIRED_SERVICES = frozenset({"monitoring-locations"})
 
+# Service name → the column the rest of the package exposes the
+# per-record OGC ``id`` under. Used by the generalized
+# :func:`get_waterdata` entry point, which doesn't have a per-service
+# wrapper to hard-code this in. Values here match what each typed
+# ``get_*`` function in :mod:`dataretrieval.waterdata.api` uses for
+# its own ``output_id``.
+_OUTPUT_ID_BY_SERVICE: dict[str, str] = {
+    "daily": "daily_id",
+    "continuous": "continuous_id",
+    "latest-continuous": "latest_continuous_id",
+    "latest-daily": "latest_daily_id",
+    "field-measurements": "field_measurement_id",
+    "field-measurements-metadata": "field_series_id",
+    "monitoring-locations": "monitoring_location_id",
+    "time-series-metadata": "time_series_id",
+    "combined-metadata": "combined_meta_id",
+    "peaks": "peak_id",
+    "channel-measurements": "channel_measurements_id",
+}
+
 
 def _parse_datetime(value: str) -> datetime | None:
     """Parse a single datetime string against the supported formats.
@@ -528,6 +548,59 @@ def _construct_api_requests(
     return request.prepare()
 
 
+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,
+) -> requests.PreparedRequest:
+    """Build a POST/CQL2 request for the generalized ``get_waterdata`` path.
+
+    Distinct from :func:`_construct_api_requests` because that function
+    derives the CQL2 body from typed kwargs; here the body is passed
+    through verbatim so a caller can express predicates the typed
+    wrappers can't (top-level ``or``, ``like`` with ``%`` wildcards,
+    comparison operators, …).
+
+    Parameters
+    ----------
+    service : str
+        Collection name, e.g. ``"daily"``.
+    cql_body : str
+        Pre-serialized CQL2-JSON. Sent as the request body unchanged.
+    properties : list of str, optional
+        Server-side property whitelist. Joined with commas.
+    bbox : list of float, optional
+        Bounding box ``[xmin, ymin, xmax, ymax]``. Joined with commas.
+    limit : int, optional
+        Page size; clamped to the server max of 50,000.
+    skip_geometry : bool, optional
+        If True, sets ``skipGeometry=true`` so the server omits
+        geometry from each feature.
+    """
+    service_url = f"{OGC_API_URL}/collections/{service}/items"
+    params: dict[str, Any] = {
+        "skipGeometry": bool(skip_geometry),
+        "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)
+    headers = _default_headers()
+    headers["Content-Type"] = "application/query-cql-json"
+    request = requests.Request(
+        method="POST",
+        url=service_url,
+        headers=headers,
+        data=cql_body,
+        params=params,
+    )
+    return request.prepare()
+
+
 def _next_req_url(resp: requests.Response) -> str | None:
     """
     Extracts the URL for the next page of results from an HTTP response from a