feat: Migrate to httpx and add async parallel chunker

Replace ``requests`` with ``httpx`` package-wide and add an opt-in
async parallel fan-out for the multi-value chunker, gated on the
``API_USGS_CONCURRENT`` env var.

* ``httpx`` ships sync and async clients on a unified API, so the
  same request shape powers both the synchronous getters callers
  use today and the new ``_fan_out_async`` parallel path; the
  unmaintained ``requests`` had no async story.
* ``API_USGS_CONCURRENT=1`` (default in tests) keeps the serial
  ``ChunkedCall.resume()`` path over one shared ``httpx.Client``.
  ``API_USGS_CONCURRENT=N`` (N > 1; default 16 in production) or
  ``unbounded`` fans the plan out through ``_fan_out_async`` over
  one shared ``httpx.AsyncClient``, bounded by
  ``asyncio.Semaphore(N)``.
* Both paths publish their client on a ``ContextVar``
  (``_chunked_session`` / ``_chunked_async_session``) so paginated
  helpers downstream reuse the connection pool across every
  sub-request of a chunked call.
* The parallel path preserves the same safety contracts as the
  serial path: it probes the first sub-request alone to read
  ``x-ratelimit-remaining`` before fanning out the rest
  (``RequestExceedsQuota``), and uses ``asyncio.gather(
  return_exceptions=True)`` so a transient failure surfaces as a
  ``ChunkInterrupted`` whose ``.call`` is a ``ChunkedCall`` holding
  the sparse-indexed completed sub-requests; ``exc.call.resume()``
  re-issues only the unfinished ones via the sync path.
* The wrapper falls back to the serial path (with a
  ``UserWarning``) when ``asyncio.get_running_loop()`` returns —
  so Jupyter / IPython kernels and async apps don't see a
  confusing ``RuntimeError`` — and when the decorator was set up
  without a ``fetch_async=`` sibling.
* Three defensive helpers smooth over httpx behaviours that
  ``requests`` didn't have: ``_safe_request_bytes`` swallows
  ``httpx.InvalidURL`` so the planner's halving loop keeps
  shrinking past httpx's 64 KB URL cap; ``_safe_elapsed`` falls
  back to ``timedelta(0)`` when ``.elapsed`` is missing (mock
  transports); ``_set_response_url`` rewrites the URL via the
  bound request, since httpx makes ``Response.url`` read-only.

Tests: ``pyproject.toml`` switches ``requests``/``requests-mock``
to ``httpx``/``pytest-httpx``; ``tests/conftest.py`` adds a
``requests_mock``-shaped shim over ``httpx_mock`` and an autouse
fixture pinning ``API_USGS_CONCURRENT=1`` so historical tests
stay on the deterministic serial path. New async-mode tests cover
the parallel fan-out, the probe-first quota check, the resumable
``ChunkInterrupted.call`` after a mid-fan-out failure, the
running-event-loop fallback, and the missing-``fetch_async``
warning.

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,17 @@
 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. ``requests`` defaulted to
+# following redirects on GET; ``httpx`` does not. ``timeout=None``
+# preserves the no-timeout behavior the chunker docs promise.
+HTTPX_DEFAULTS = {"follow_redirects": True, "timeout": None}
+
 
 def to_str(listlike, delimiter=","):
     """Translates list-like objects into strings.
@@ -205,7 +210,7 @@ class BaseMetadata:
         Response url
     query_time: datetme.timedelta
         Response elapsed time
-    header: requests.structures.CaseInsensitiveDict
+    header: httpx.Headers
         Response headers
 
     """
@@ -216,7 +221,7 @@ def __init__(self, response) -> None:
         Parameters
         ----------
         response: Response
-            Response object from requests module
+            Response object from httpx module
 
         Returns
         -------
@@ -225,8 +230,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 +259,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 +291,32 @@ 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.
+        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 +328,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,
@@ -2110,7 +2114,7 @@ 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 +2340,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 +2414,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()