feat(waterdata): async parallel chunker over httpx.AsyncClient

Multi-value Water Data queries (many monitoring locations, many parameter
codes, large CQL2 filters) can exceed the server's ~8 KB URL/body limit
and need to be split into multiple sub-requests. This adds an
``async``-only chunker that:

* Plans the fan-out: every multi-value list parameter and the cql-text
  ``filter`` (along its top-level ``OR`` clauses) is modeled as a chunkable
  axis; ``ChunkPlan`` greedy-halves the biggest axis until every
  sub-request URL fits the byte budget, then iterates the cartesian product.
* Dispatches concurrently: ``ChunkedCall._run`` gathers every pending
  sub-request through one shared ``httpx.AsyncClient`` with the connection
  pool sized from ``API_USGS_CONCURRENT`` (default 16; ``unbounded``
  removes the cap). A single ``anyio`` blocking portal lets the sync
  facade work from inside event loops (Jupyter, async apps).
* Survives transients: typed ``RateLimited`` (429) and
  ``ServiceUnavailable`` (5xx) trigger bounded retry-with-backoff
  (``API_USGS_RETRIES`` default 4; full jitter; honors ``Retry-After``
  up to a 60 s cap). Anything still failing escalates to a resumable
  ``ChunkInterrupted`` subclass carrying ``.call`` — call ``.call.resume()``
  once the underlying condition clears; only the still-pending
  sub-requests are re-issued.
* Combines and finalizes: the OGC getters inject ``utils._finalize_ogc``
  (type coercion, column arrangement, ``max_rows`` truncation,
  ``BaseMetadata``) through the chunker's ``finalize`` hook so a
  successful first call and a resumed call yield the same shape.

Surfaces and integration:

* ``dataretrieval/waterdata/chunking.py`` (new module): ``RetryPolicy``,
  ``ChunkPlan``, ``ChunkedCall``, ``ChunkInterrupted`` /
  ``QuotaExhausted`` / ``ServiceInterrupted``, ``multi_value_chunked``
  decorator.
* ``utils._fetch_once`` is the decorated async fetcher; pagination
  helpers ``_paginate`` / ``_walk_pages`` are async and share the
  chunker's client via a ``ContextVar``.
* ``api.get_reference_table(... max_rows=...)``: new preview cap.
* ``_progress``: per-call status line (chunk count, pages, rows,
  rate-limit remaining); ``API_USGS_PROGRESS`` opt-in/off.

Deps: add ``geopandas>=0.10`` + ``mapclassify`` to ``[doc]`` extras so
``WaterData_demo.ipynb``'s ``.set_crs().explore()`` cell executes (the
plain-pandas frame lacks ``.set_crs``).

Tests: full async chunker suite (planning, retry taxonomy, resume,
client-sharing, progress reporter, finalize injection) + live-API
regression tests covering every public getter. 298 offline + 63 live
tests pass.

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

Author: thodson-usgs

dataretrieval/waterdata/_progress.py

@@ -121,6 +121,9 @@ def __init__(
         # The hourly request quota (``x-ratelimit-limit``), shown as the
         # denominator when the server reports it.
         self.rate_limit: str | None = None
+        # Transient note shown while a sub-request backs off before a
+        # retry; cleared by the next page/chunk so it doesn't linger.
+        self.retry_note: str | None = None
         self._last_len = 0
         # Whether anything was actually written to the stream — drives whether
         # close() needs a terminating newline. (``current_chunk`` is a poor
@@ -140,13 +143,33 @@ def start_chunk(self, index: int) -> None:
         avoids a premature "0 pages" frame before the first page arrives.
         """
         self.current_chunk = index
+        self.retry_note = None
         if self.total_chunks > 1:
             self._render()
 
     def add_page(self, rows: int = 0) -> None:
         """Record one fetched page carrying ``rows`` rows and redraw."""
         self.pages += 1
         self.rows += int(rows)
+        self.retry_note = None
+        self._render()
+
+    def note_retry(self, *, attempt: int, wait: float) -> None:
+        """Show that a sub-request is backing off before retry ``attempt``.
+
+        Cleared by the next :meth:`add_page` / :meth:`start_chunk` (or by
+        :meth:`close`) so the line returns to normal once the retry resolves.
+        """
+        # Keep sub-second waits explicit (avoid misleading ``0s``) while
+        # rendering whole-second waits without unnecessary ``.0`` noise.
+        # ``float()`` to support Python 3.9-3.11: ``round(int, 1)`` returns an
+        # int and ``int.is_integer()`` (used below) only exists on 3.12+.
+        wait_1dp = round(float(wait), 1)
+        if wait_1dp < 1 or not wait_1dp.is_integer():
+            secs = f"{wait_1dp:.1f}s"
+        else:
+            secs = f"{wait_1dp:.0f}s"
+        self.retry_note = f"retrying (attempt {attempt}, waiting {secs})"
         self._render()
 
     def set_rate_remaining(
@@ -179,6 +202,8 @@ def _format(self) -> str:
             else:
                 segment = f"{remaining} requests remaining"
             parts.append(segment)
+        if self.retry_note is not None:
+            parts.append(self.retry_note)
         if self.service:
             return f"Retrieving: {self.service} · " + " · ".join(parts)
         return "Progress: " + " · ".join(parts)
@@ -209,6 +234,13 @@ def close(self) -> None:
         """
         if self._closed:
             return
+        # A retry note set during the final backoff would otherwise freeze as
+        # the persisted last line of a call that has since completed or given
+        # up; clear it and redraw (while still un-closed, so ``_render`` runs)
+        # so the final state isn't a stale "retrying".
+        if self.enabled and self._rendered and self.retry_note is not None:
+            self.retry_note = None
+            self._render()
         self._closed = True
         if not (self.enabled and self._rendered):
             return

dataretrieval/waterdata/api.py

@@ -2022,6 +2022,7 @@ def get_reference_table(
     collection: str,
     limit: int | None = None,
     query: dict | None = None,
+    max_rows: int | None = None,
 ) -> tuple[pd.DataFrame, BaseMetadata]:
     """Get metadata reference tables for the USGS Water Data API.
 
@@ -2046,6 +2047,12 @@ def get_reference_table(
     query: dictionary, optional
         The optional args parameter can be used to pass a dictionary of
         query parameters to the collection API call.
+    max_rows : int, optional
+        Cap the total number of rows returned, stopping pagination early
+        instead of downloading the whole table. Useful for cheaply
+        previewing large tables (e.g. ``hydrologic-unit-codes`` has ~125k
+        rows). Unlike ``limit`` (the per-page size), this bounds the total
+        result. The default (None) downloads every page.
 
     Returns
     -------
@@ -2092,7 +2099,9 @@ def get_reference_table(
     query_args = dict(query) if query else {}
     if limit is not None:
         query_args["limit"] = limit
-    return get_ogc_data(args=query_args, output_id=output_id, service=collection)
+    return get_ogc_data(
+        args=query_args, output_id=output_id, service=collection, max_rows=max_rows
+    )
 
 
 def get_codes(code_service: CODE_SERVICES) -> pd.DataFrame: