fix(waterdata): never let the progress line crash or truncate a query

Code review surfaced two real defects:

- ProgressReporter rendering wrote to stderr unguarded. Because the per-page
  `reporter.add_page()` call sits inside `_walk_pages`' broad `except
  Exception`, a render failure — BrokenPipeError when output is piped to
  `head`, or UnicodeEncodeError for the `·` separator on a non-UTF-8 stderr —
  was misread as a failed request and silently truncated the result; on the
  first (pre-loop) page it crashed the query outright. `_render` and `close`
  now swallow stream errors and disable further rendering, so progress is
  strictly best-effort.

- The "Geopandas not installed" advisory lived in the per-page
  `_handle_stats_nesting`, so a multi-page stats query (now at WARNING) emitted
  one warning per page. Moved it to `get_stats_data` so it fires once, matching
  `_walk_pages`.

Also harden rate-limit formatting against non-decimal unicode digits
(`isascii() and isdigit()`), and add regression tests: a broken progress
stream must not truncate pagination, and the geopandas advisory fires once.

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

Author: thodson-usgs

dataretrieval/waterdata/_progress.py

@@ -110,20 +110,28 @@ def _format(self) -> str:
             parts.append(f"{self.rows:,} rows")
         if self.rate_remaining is not None:
             # The header is a string; group it like the row count when it's a
-            # plain integer, otherwise show it verbatim.
+            # plain ASCII integer, otherwise show it verbatim. (``str.isdigit``
+            # alone is True for non-decimal unicode digits that ``int`` rejects.)
             rate = self.rate_remaining
-            rate = f"{int(rate):,}" if rate.isdigit() else rate
+            rate = f"{int(rate):,}" if rate.isascii() and rate.isdigit() else rate
             parts.append(f"{rate} requests left")
         return "Progress: " + " · ".join(parts)
 
     def _render(self) -> None:
         if not self.enabled or self._closed:
             return
-        line = self._format()
-        pad = max(self._last_len - len(line), 0)
-        self._stream.write("\r" + line + " " * pad)
-        self._stream.flush()
-        self._last_len = len(line)
+        try:
+            line = self._format()
+            pad = max(self._last_len - len(line), 0)
+            self._stream.write("\r" + line + " " * pad)
+            self._stream.flush()
+            self._last_len = len(line)
+        except Exception:  # noqa: BLE001
+            # Progress output is best-effort cosmetics; a broken pipe (output
+            # piped to ``head``), a closed stream, or an encoding error must
+            # never disturb — let alone truncate — the query. Disable so we
+            # don't retry on every subsequent page.
+            self.enabled = False
 
     def close(self) -> None:
         """Finalize the line with a trailing newline so it persists on screen.
@@ -137,9 +145,12 @@ def close(self) -> None:
         self._closed = True
         if not (self.enabled and (self.pages or self.current_chunk)):
             return
-        self._stream.write("\n")
-        self._maybe_hint_api_key()
-        self._stream.flush()
+        try:
+            self._stream.write("\n")
+            self._maybe_hint_api_key()
+            self._stream.flush()
+        except Exception:  # noqa: BLE001
+            self.enabled = False
 
     def _maybe_hint_api_key(self) -> None:
         global _api_key_hint_shown

dataretrieval/waterdata/utils.py

@@ -963,12 +963,6 @@ def _handle_stats_nesting(
     if body is None:
         return pd.DataFrame()
 
-    if not geopd:
-        logger.warning(
-            "Geopandas not installed. Geometries will be flattened "
-            "into pandas DataFrames."
-        )
-
     # If geopandas not installed, return a pandas dataframe
     # otherwise return a geodataframe
     if not geopd:
@@ -1119,6 +1113,12 @@ def get_stats_data(
     req = request.prepare()
     logger.debug("Request: %s", req.url)
 
+    if not GEOPANDAS:
+        logger.warning(
+            "Geopandas not installed. Geometries will be flattened "
+            "into pandas DataFrames."
+        )
+
     # create temp client if not provided
     # and close it after the request is done
     close_client = client is None

tests/waterdata_progress_test.py

@@ -102,6 +102,24 @@ def test_close_without_activity_writes_nothing():
     assert stream.getvalue() == ""
 
 
+class _RaisingStream:
+    """A stream whose writes always fail, e.g. a broken pipe (output | head)."""
+
+    def write(self, *_):
+        raise BrokenPipeError("broken pipe")
+
+    def flush(self):
+        pass
+
+
+def test_reporter_swallows_stream_errors_and_disables(monkeypatch):
+    monkeypatch.delenv("API_USGS_PAT", raising=False)
+    reporter = ProgressReporter(stream=_RaisingStream(), enabled=True)
+    reporter.add_page(rows=1)  # render write raises -> must be swallowed
+    reporter.close()  # newline + hint writes raise -> must be swallowed
+    assert reporter.enabled is False
+
+
 # -- API-key pointer -----------------------------------------------------------
 
 
@@ -262,3 +280,25 @@ def test_walk_pages_without_context_does_not_error():
     df, _ = _walk_pages(geopd=False, req=req, client=client)
     assert len(df) == 1
     assert current() is None
+
+
+def test_broken_progress_stream_does_not_truncate_pagination():
+    # A render failure (broken pipe) lands inside _walk_pages' per-page try;
+    # it must NOT be mistaken for a failed request and silently drop pages.
+    resp1 = _resp(
+        [{"id": "1", "properties": {"v": "a"}}], next_url="https://example.com/p2"
+    )
+    resp2 = _resp([{"id": "2", "properties": {"v": "b"}}])
+    client = mock.MagicMock(spec=requests.Session)
+    client.send.return_value = resp1
+    client.request.return_value = resp2
+
+    req = mock.MagicMock(spec=requests.PreparedRequest)
+    req.method = "GET"
+    req.headers = {}
+    req.url = "https://example.com/p1"
+
+    with progress_context(stream=_RaisingStream(), enabled=True):
+        df, _ = _walk_pages(geopd=False, req=req, client=client)
+
+    assert len(df) == 2  # both pages returned despite the broken progress stream

tests/waterdata_utils_test.py

@@ -307,31 +307,37 @@ def test_walk_pages_warns_when_geopandas_missing(caplog):
     assert any("Geopandas not installed" in m for m in _warning_messages(caplog))
 
 
-def test_handle_stats_nesting_warns_when_geopandas_missing(caplog):
-    # This path previously logged at INFO (silent by default); it must warn.
+def test_get_stats_data_warns_once_when_geopandas_missing(caplog, monkeypatch):
+    # The advisory must fire once per query, not once per paginated page
+    # (it previously lived in the per-page _handle_stats_nesting helper).
+    from dataretrieval.waterdata.utils import get_stats_data
+
+    monkeypatch.setattr(_utils_module, "GEOPANDAS", False)
+    monkeypatch.setattr(
+        _utils_module,
+        "_handle_stats_nesting",
+        mock.MagicMock(return_value=pd.DataFrame()),
+    )
     caplog.set_level(logging.WARNING, logger=_LOGGER_NAME)
-    body = {
-        "next": None,
-        "features": [
-            {
-                "properties": {
-                    "monitoring_location_id": "USGS-12345",
-                    "data": [
-                        {
-                            "parameter_code": "00060",
-                            "unit_of_measure": "ft^3/s",
-                            "parent_time_series_id": "ts-1",
-                            "values": [{"statistic_id": "mean", "value": 10.0}],
-                        }
-                    ],
-                },
-            }
-        ],
-    }
 
-    _handle_stats_nesting(body, geopd=False)
+    page1 = mock.MagicMock(status_code=200, headers={})
+    page1.json.return_value = {"next": "tok", "features": []}
+    page2 = mock.MagicMock(status_code=200, headers={})
+    page2.json.return_value = {"next": None, "features": []}
 
-    assert any("Geopandas not installed" in m for m in _warning_messages(caplog))
+    client = mock.MagicMock(spec=requests.Session)
+    client.send.return_value = page1
+    client.request.return_value = page2
+
+    get_stats_data(
+        args={"monitoring_location_id": "USGS-1"},
+        service="observationNormals",
+        expand_percentiles=False,
+        client=client,
+    )
+
+    geo = [m for m in _warning_messages(caplog) if "Geopandas not installed" in m]
+    assert len(geo) == 1
 
 
 # --- _arrange_cols ----------------------------------------------------------