feat(waterdata): Migrate to httpx and add async parallel chunker

Replaces ``requests`` with ``httpx`` package-wide and adds an async
parallel branch to the multi-value chunker, governed by the new
``API_USGS_CONCURRENT`` environment variable. Parallel mode is the
default; set ``API_USGS_CONCURRENT=1`` to force the legacy sequential
path. Benchmarked on a 52,753-site / 10-state ``get_daily`` query:
12.69s parallel vs 67.33s sequential (5.3x speedup), identical row
counts, comparable quota burn.

Why httpx: the parallel fan-out runs on a single shared
``httpx.AsyncClient`` so all sub-requests amortize one TCP+TLS
handshake — impossible under the requests stack without a thread pool.
Both modes share one ``_walk_pages_steps`` generator (pagination state
machine) driven by thin sync/async loops; future retry/backoff lands
in one place via the ``("wait", seconds)`` yield pattern the drivers
can translate to ``time.sleep`` / ``await asyncio.sleep``.

Production
- New ``httpx`` dependency, dropped ``requests``; ``pytest-httpx``
  replaces ``requests-mock`` in test extras.
- Three httpx behavior diffs handled defensively: ``InvalidURL``
  (URLs > 64 KB rejected client-side) via ``_safe_request_bytes`` /
  ``_safe_canonical_url``; ``Response.elapsed`` only populated on close
  via ``_safe_elapsed``; ``Response.url`` is a read-only property
  wrapped via ``_set_response_url``.
- ``BaseMetadata.url`` coerced to ``str`` to preserve the string-typed
  contract. ``BaseMetadata.header`` is now ``httpx.Headers`` (see
  backwards-compat note below).
- Chunker decorator gains ``walk_pages_async=`` and honors
  ``API_USGS_CONCURRENT``. Silently falls back to sequential (with INFO
  log) when ``walk_pages_async`` isn't wired or when args trip the
  inner filter chunker — preserves workloads that worked before
  parallel-by-default.
- One shared ``httpx.Client`` / ``AsyncClient`` per chunked call via
  ``ContextVar``. Eliminates per-sub-request TCP+TLS handshakes in
  both modes.
- Bug fix surfaced by the migration: ``filters.py`` was calling
  ``len(probe.url)`` which fails on ``httpx.URL`` — fixed.
- Unified pagination state machine: ``_walk_pages``,
  ``_walk_pages_async``, and ``get_stats_data`` all route through
  ``_walk_pages_steps`` parameterized on ``page_data_fn`` and
  ``next_req_fn``. ~80 lines of duplicated pagination logic collapsed
  into one place; same source of truth across two pagination styles
  (OGC link-header and stats next-token).

Tests
- All 8 test files migrated from ``requests-mock`` to ``pytest-httpx``.
- ``tests/conftest.py`` (new) centralizes ``mock_request``,
  ``assert_url_equivalent``, ``assert_mock_header`` (previously
  triplicated across three test files).
- New tests for parallel-mode contract, env-var parsing
  (explicit-low / malformed / unset), client contextvar publishing,
  and the ``_safe_*`` defensive helpers.
- 194 mocked tests pass; live-API tests unaffected by the migration
  (one pre-existing USGS column-drift failure unrelated to this PR).

Backwards-compat
- ``BaseMetadata.header`` type: ``requests.structures.CaseInsensitiveDict``
  → ``httpx.Headers``. Both behave like case-insensitive dicts for
  reads, but ``httpx.Headers`` carries auto-added entries (``host``,
  ``content-length``) so ``md.header == {"key": "val"}`` literal
  equality breaks. Use ``md.header.get(...)``.
- ``API_USGS_CONCURRENT`` now controls a feature that defaults on.
  Workloads that combined long multi-value lists with chunkable CQL
  filters (rare) automatically fall back to sequential — no caller
  changes needed.

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

Author: thodson-usgs

dataretrieval/nadp.py

@@ -34,7 +34,9 @@
 import warnings
 import zipfile
 
-import requests
+import httpx
+
+from dataretrieval.utils import HTTPX_DEFAULTS
 
 _DEPRECATION_MESSAGE = (
     "The `nadp` module is deprecated and will be removed from `dataretrieval` "
@@ -213,7 +215,7 @@ def get_zip(url, filename):
     """
     _warn_deprecated()
 
-    req = requests.get(url + filename)
+    req = httpx.get(url + filename, **HTTPX_DEFAULTS)
     req.raise_for_status()
 
     # z = zipfile.ZipFile(io.BytesIO(req.content))

dataretrieval/nwis.py

@@ -11,8 +11,8 @@
 import warnings
 from json import JSONDecodeError
 
+import httpx
 import pandas as pd
-import requests
 
 from dataretrieval.rdb import read_rdb
 from dataretrieval.utils import BaseMetadata
@@ -110,7 +110,7 @@ def wrapper(*args, **kwargs):
     return wrapper
 
 
-def _parse_json_or_raise(response: requests.Response) -> pd.DataFrame:
+def _parse_json_or_raise(response: httpx.Response) -> pd.DataFrame:
     """Parse a JSON NWIS response, raising a helpful error on HTML responses."""
     try:
         return _read_json(response.json())
@@ -364,9 +364,7 @@ def get_stats(
 
 
 @_deprecated
-def query_waterdata(
-    service: str, ssl_check: bool = True, **kwargs
-) -> requests.models.Response:
+def query_waterdata(service: str, ssl_check: bool = True, **kwargs) -> httpx.Response:
     """
     Queries waterdata.
 
@@ -382,7 +380,7 @@ def query_waterdata(
 
     Returns
     -------
-    request: ``requests.models.Response``
+    request: ``httpx.Response``
         The response object from the API request to the web service
     """
     major_params = ["site_no", "state_cd"]
@@ -412,7 +410,7 @@ def query_waterdata(
 @_deprecated
 def query_waterservices(
     service: str, ssl_check: bool = True, **kwargs
-) -> requests.models.Response:
+) -> httpx.Response:
     """
     Queries waterservices.usgs.gov
 
@@ -451,7 +449,7 @@ def query_waterservices(
 
     Returns
     -------
-    request: ``requests.models.Response``
+    request: ``httpx.Response``
         The response object from the API request to the web service
 
     """
@@ -1123,7 +1121,7 @@ class NWIS_Metadata(BaseMetadata):
         Response url
     query_time: datetme.timedelta
         Response elapsed time
-    header: requests.structures.CaseInsensitiveDict
+    header: httpx.Headers
         Response headers
     comments: str | None
         Metadata comments, if any
@@ -1143,7 +1141,7 @@ def __init__(self, response, **parameters) -> None:
         Parameters
         ----------
         response: Response
-            Response object from requests module
+            Response object from httpx module
         parameters: unpacked dictionary
             Unpacked dictionary of the parameters supplied in the request
 

dataretrieval/streamstats.py

@@ -7,7 +7,9 @@
 
 import json
 
-import requests
+import httpx
+
+from dataretrieval.utils import HTTPX_DEFAULTS
 
 
 def download_workspace(workspaceID, format=""):
@@ -32,7 +34,7 @@ def download_workspace(workspaceID, format=""):
     payload = {"workspaceID": workspaceID, "format": format}
     url = "https://streamstats.usgs.gov/streamstatsservices/download"
 
-    r = requests.get(url, params=payload)
+    r = httpx.get(url, params=payload, **HTTPX_DEFAULTS)
 
     r.raise_for_status()
     return r
@@ -125,7 +127,7 @@ def get_watershed(
     }
     url = "https://streamstats.usgs.gov/streamstatsservices/watershed.geojson"
 
-    r = requests.get(url, params=payload)
+    r = httpx.get(url, params=payload, **HTTPX_DEFAULTS)
 
     r.raise_for_status()
 

dataretrieval/utils.py

@@ -5,12 +5,18 @@
 import warnings
 from collections.abc import Iterable
 
+import httpx
 import pandas as pd
-import requests
 
 import dataretrieval
 from dataretrieval.codes import tz
 
+# Shared kwargs for every ``httpx`` call site in the package. ``requests``
+# defaulted to following redirects on GET; ``httpx`` does not. ``timeout=None``
+# preserves the no-timeout behavior the chunker docs promise (caller-driven
+# timeout policy via wrappers).
+HTTPX_DEFAULTS = {"follow_redirects": True, "timeout": None}
+
 
 def to_str(listlike, delimiter=","):
     """Translates list-like objects into strings.
@@ -205,7 +211,7 @@ class BaseMetadata:
         Response url
     query_time: datetme.timedelta
         Response elapsed time
-    header: requests.structures.CaseInsensitiveDict
+    header: httpx.Headers
         Response headers
 
     """
@@ -216,7 +222,7 @@ def __init__(self, response) -> None:
         Parameters
         ----------
         response: Response
-            Response object from requests module
+            Response object from httpx module
 
         Returns
         -------
@@ -225,8 +231,8 @@ def __init__(self, response) -> None:
 
         """
 
-        # These are built from the API response
-        self.url = response.url
+        # Coerce httpx.URL -> str: BaseMetadata.url has always been str.
+        self.url = str(response.url)
         self.query_time = response.elapsed
         self.header = response.headers
         self.comment = None
@@ -254,18 +260,29 @@ def __repr__(self) -> str:
         return f"{type(self).__name__}(url={self.url})"
 
 
+_URL_TOO_LONG_EXAMPLE = """
+                    # n is the number of chunks to divide the query into \n
+                    split_list = np.array_split(site_list, n)
+                    data_list = []  # list to store chunk results in \n
+                    # loop through chunks and make requests \n
+                    for site_list in split_list: \n
+                        data = nwis.get_record(sites=site_list, service='dv', \n
+                                               start=start, end=end) \n
+                        data_list.append(data)  # append results to list"""
+
+
 def query(url, payload, delimiter=",", ssl_check=True):
     """Send a query.
 
-    Wrapper for requests.get that handles errors, converts listed
+    Wrapper for httpx.get that handles errors, converts listed
     query parameters to comma separated strings, and returns response.
 
     Parameters
     ----------
     url: string
         URL to query
     payload: dict
-        query parameters passed to ``requests.get``
+        query parameters passed to ``httpx.get``
     delimiter: string
         delimiter to use with lists
     ssl_check: bool
@@ -275,19 +292,34 @@ def query(url, payload, delimiter=",", ssl_check=True):
     Returns
     -------
     string: query response
-        The response from the API query ``requests.get`` function call.
+        The response from the API query ``httpx.get`` function call.
     """
 
     for key, value in payload.items():
         payload[key] = to_str(value, delimiter)
-    # for index in range(len(payload)):
-    #    key, value = payload[index]
-    #    payload[index] = (key, to_str(value))
+    # httpx serializes None params as ``foo=``; USGS rejects with 400.
+    # 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}
 
     # define the user agent for the query
     user_agent = {"user-agent": f"python-dataretrieval/{dataretrieval.__version__}"}
 
-    response = requests.get(url, params=payload, headers=user_agent, verify=ssl_check)
+    try:
+        response = httpx.get(
+            url,
+            params=payload,
+            headers=user_agent,
+            verify=ssl_check,
+            **HTTPX_DEFAULTS,
+        )
+    except httpx.InvalidURL as exc:
+        # httpx rejects oversize URLs client-side; mirror the 414 branch
+        # so caller handling is unchanged.
+        raise ValueError(
+            "Request URL too long. Modify your query to use fewer sites. "
+            f"httpx rejected the URL client-side: {exc}. Pseudo-code "
+            f"example of how to split your query: \n {_URL_TOO_LONG_EXAMPLE}"
+        ) from exc
 
     if response.status_code == 400:
         raise ValueError(
@@ -299,24 +331,14 @@ def query(url, payload, delimiter=",", ssl_check=True):
             + f"URL: {response.url}"
         )
     elif response.status_code == 414:
-        _reason = response.reason
-        _example = """
-                    # n is the number of chunks to divide the query into \n
-                    split_list = np.array_split(site_list, n)
-                    data_list = []  # list to store chunk results in \n
-                    # loop through chunks and make requests \n
-                    for site_list in split_list: \n
-                        data = nwis.get_record(sites=site_list, service='dv', \n
-                                               start=start, end=end) \n
-                        data_list.append(data)  # append results to list"""
         raise ValueError(
             "Request URL too long. Modify your query to use fewer sites. "
-            + f"API response reason: {_reason}. Pseudo-code example of how to "
-            + f"split your query: \n {_example}"
+            f"API response reason: {response.reason_phrase}. Pseudo-code "
+            f"example of how to split your query: \n {_URL_TOO_LONG_EXAMPLE}"
         )
     elif response.status_code in [500, 502, 503]:
         raise ValueError(
-            f"Service Unavailable: {response.status_code} {response.reason}. "
+            f"Service Unavailable: {response.status_code} {response.reason_phrase}. "
             + f"The service at {response.url} may be down or experiencing issues."
         )
 

dataretrieval/waterdata/api.py

@@ -13,11 +13,15 @@
 from typing import get_args
 from urllib.parse import quote
 
+import httpx
 import pandas as pd
-import requests
-from requests.models import PreparedRequest
 
-from dataretrieval.utils import BaseMetadata, _attach_datetime_columns, to_str
+from dataretrieval.utils import (
+    HTTPX_DEFAULTS,
+    BaseMetadata,
+    _attach_datetime_columns,
+    to_str,
+)
 from dataretrieval.waterdata.filters import FILTER_LANG
 from dataretrieval.waterdata.types import (
     CODE_SERVICES,
@@ -188,7 +192,11 @@ def get_daily(
         :mod:`dataretrieval.waterdata.filters` for syntax, auto-chunking,
         and the lexicographic-comparison pitfall.
     convert_type : boolean, optional
-        If True, converts columns to appropriate types.
+        If True, converts columns to appropriate types. Long multi-value
+        inputs are transparently chunked across sub-requests that fan
+        out concurrently by default — see
+        :mod:`dataretrieval.waterdata.chunking` for the
+        ``API_USGS_CONCURRENT`` tuning knob.
 
     Returns
     -------
@@ -2110,7 +2118,11 @@ def get_codes(code_service: CODE_SERVICES) -> pd.DataFrame:
 
     url = f"{SAMPLES_URL}/codeservice/{code_service}?mimeType=application%2Fjson"
 
-    response = requests.get(url, headers=_default_headers())
+    response = httpx.get(
+        url,
+        headers=_default_headers(),
+        **HTTPX_DEFAULTS,
+    )
 
     response.raise_for_status()
 
@@ -2336,12 +2348,14 @@ def get_samples(
 
     url = f"{SAMPLES_URL}/{service}/{profile}"
 
-    req = PreparedRequest()
-    req.prepare_url(url, params=params)
-    logger.info("Request: %s", req.url)
+    logger.info("Request: %s", httpx.URL(url).copy_merge_params(params))
 
-    response = requests.get(
-        url, params=params, verify=ssl_check, headers=_default_headers()
+    response = httpx.get(
+        url,
+        params=params,
+        verify=ssl_check,
+        headers=_default_headers(),
+        **HTTPX_DEFAULTS,
     )
 
     response.raise_for_status()
@@ -2408,12 +2422,14 @@ def get_samples_summary(
     url = f"{SAMPLES_URL}/summary/{quote(monitoringLocationIdentifier, safe='')}"
     params = {"mimeType": "text/csv"}
 
-    req = PreparedRequest()
-    req.prepare_url(url, params=params)
-    logger.info("Request: %s", req.url)
+    logger.info("Request: %s", httpx.URL(url).copy_merge_params(params))
 
-    response = requests.get(
-        url, params=params, verify=ssl_check, headers=_default_headers()
+    response = httpx.get(
+        url,
+        params=params,
+        verify=ssl_check,
+        headers=_default_headers(),
+        **HTTPX_DEFAULTS,
     )
 
     response.raise_for_status()