fix(waterdata): show requests as remaining/limit, drop unavailable reset; review fixes

The live USGS API returns X-Ratelimit-Limit + X-Ratelimit-Remaining but no
reset header, so the earlier "resets in X" countdown read a header that never
arrives. Replace it with the available "<remaining> / <limit> requests
remaining" (e.g. "995 / 1,000"); the reset parsing/countdown and its tests are
removed.

Also fixes two issues from code review:
- close() used current_chunk as a proxy for "drew something", but start_chunk
  sets it even when it doesn't render, so a query that failed before any page
  printed a stray newline and burned the once-per-process API-key hint. Now
  gated on a _rendered flag set only after a successful write.
- _maybe_hint_api_key set its latch before the write, so a failed write burned
  the hint process-wide; the latch is now set only after a successful write.

Plus doc-comment accuracy: the ContextVar isolates one query's reporter (it does
not give concurrent queries on one stderr separate lines), and references to the
removed ``filters.chunked`` decorator now point at ``chunking`` / ``_paginate``.

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

Author: thodson-usgs

README.md

@@ -111,11 +111,12 @@ for more information and examples on available services and input parameters.
 
 **Tracking progress:** Paginated and chunked `waterdata` queries report their
 progress on a single, self-updating line on `stderr` — showing the chunk and
-page counts, rows retrieved so far, and the API requests remaining (with the
-time until the hourly limit resets, when the server reports it):
+page counts, rows retrieved so far, and the hourly API quota remaining (the
+``remaining / limit`` from the server's rate-limit headers, shown when you
+have an API key set):
 
 ```text
-Progress: chunk 2/5 · 14 pages · 8,421 rows · 4,870 requests remaining, resets in 47m
+Progress: chunk 2/5 · 14 pages · 8,421 rows · 4,870 / 5,000 requests remaining
 ```
 
 The line appears automatically when `stderr` is an interactive terminal.

dataretrieval/waterdata/_progress.py

@@ -1,19 +1,19 @@
 """A single self-updating status line for paginated / chunked Water Data queries.
 
-Water Data getters fan out two ways the caller can't see: long CQL filters are
-split into URL-length-safe *chunks* (``filters.chunked``), and each request
-follows ``next`` links across an unknown number of *pages* (``utils._walk_pages``
-and ``utils.get_stats_data``). This module surfaces that work as one line on
-stderr, rewritten in place as data arrives::
+Water Data getters fan out two ways the caller can't see: large multi-value
+requests are split into URL-length-safe *chunks* (``chunking`` module), and each
+request follows ``next`` links across an unknown number of *pages*
+(``utils._paginate``). This module surfaces that work as one line on stderr,
+rewritten in place as data arrives::
 
-    Progress: chunk 2/5 · 14 pages · 8,421 rows · 4,870 requests remaining
+    Progress: chunk 2/5 · 14 pages · 8,421 rows · 4,870 / 5,000 requests remaining
 
 It replaces the per-page ``logger.info`` calls that previously narrated the same
 events one line at a time.
 
 The active reporter lives in a :class:`~contextvars.ContextVar` rather than being
 threaded through every signature: progress is a cross-cutting concern that the
-``chunked`` decorator (outer, chunk counts) and the page-walking loops (inner,
+chunk orchestrator (outer, chunk counts) and the page-walking loop (inner,
 page/row/rate-limit counts) both update without knowing about each other. Call
 :func:`progress_context` to activate one and :func:`current` to reach it.
 
@@ -27,27 +27,25 @@
 import contextvars
 import os
 import sys
-import time
 from collections.abc import Iterator
 from contextlib import contextmanager
 from typing import TextIO
 
 
-def _format_duration(seconds: float) -> str:
-    """Compact human duration: ``45s``, ``12m``, ``1h03m`` (clamped at 0)."""
-    secs = int(max(0, seconds))
-    if secs < 60:
-        return f"{secs}s"
-    if secs < 3600:
-        return f"{secs // 60}m"
-    hours, rem = divmod(secs, 3600)
-    minutes = rem // 60
-    return f"{hours}h{minutes:02d}m" if minutes else f"{hours}h"
+def _group_int(value: str) -> str:
+    """Comma-group a plain ASCII integer string; pass anything else through.
+
+    (``str.isdigit`` alone is True for non-decimal unicode digits that ``int``
+    rejects, hence the ``isascii`` guard.)
+    """
+    return f"{int(value):,}" if value.isascii() and value.isdigit() else value
 
 
 # The reporter active for the current query. A ContextVar (not a module global)
-# so concurrent queries — threads or async tasks sharing a client — each track
-# their own progress line.
+# so the chunk orchestrator and the page loop resolve to the same reporter
+# within one query, and an unrelated query in another context can't clobber its
+# state. (It does not give concurrent queries sharing one stderr separate
+# lines — they would still interleave.)
 _active: contextvars.ContextVar[ProgressReporter | None] = contextvars.ContextVar(
     "waterdata_progress", default=None
 )
@@ -88,10 +86,14 @@ def __init__(
         self.pages = 0
         self.rows = 0
         self.rate_remaining: str | None = None
-        # Absolute epoch second when the rate-limit window resets, derived from
-        # the server's reset header so the rendered countdown stays live.
-        self._reset_at: float | None = None
+        # The hourly request quota (``x-ratelimit-limit``), shown as the
+        # denominator when the server reports it.
+        self.rate_limit: 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
+        # proxy: ``start_chunk`` sets it even when it doesn't render.)
+        self._rendered = False
         self._closed = False
 
     def set_chunks(self, total: int) -> None:
@@ -116,25 +118,19 @@ def add_page(self, rows: int = 0) -> None:
         self._render()
 
     def set_rate_remaining(
-        self, value: str | int | None, reset: str | int | None = None
+        self, value: str | int | None, limit: str | int | None = None
     ) -> None:
         """Update the rate-limit display from the response headers.
 
-        ``value`` is ``x-ratelimit-remaining``; ``reset`` is the optional
-        ``x-ratelimit-reset`` companion. Empty/missing values are ignored so a
-        page that omits a header doesn't blank out the last known value. The
-        reset value is interpreted as an absolute epoch second when large
-        (the conventional form) and as seconds-until-reset otherwise; either
-        way it's stored as an absolute deadline so the countdown stays live.
+        ``value`` is ``x-ratelimit-remaining``; ``limit`` is the optional
+        ``x-ratelimit-limit`` quota, shown as the denominator. Empty/missing
+        values are ignored so a page that omits a header doesn't blank out the
+        last known value.
         """
         if value not in (None, ""):
             self.rate_remaining = str(value)
-        if reset not in (None, ""):
-            try:
-                secs = float(reset)
-            except (TypeError, ValueError):
-                return
-            self._reset_at = secs if secs > 1_000_000 else time.time() + secs
+        if limit not in (None, ""):
+            self.rate_limit = str(limit)
 
     def _format(self) -> str:
         parts: list[str] = []
@@ -144,15 +140,12 @@ def _format(self) -> str:
         if self.rows:
             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 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.isascii() and rate.isdigit() else rate
-            segment = f"{rate} requests remaining"
-            if self._reset_at is not None:
-                eta = _format_duration(self._reset_at - time.time())
-                segment += f", resets in {eta}"
+            remaining = _group_int(self.rate_remaining)
+            if self.rate_limit is not None:
+                limit = _group_int(self.rate_limit)
+                segment = f"{remaining} / {limit} requests remaining"
+            else:
+                segment = f"{remaining} requests remaining"
             parts.append(segment)
         return "Progress: " + " · ".join(parts)
 
@@ -165,6 +158,7 @@ def _render(self) -> None:
             self._stream.write("\r" + line + " " * pad)
             self._stream.flush()
             self._last_len = len(line)
+            self._rendered = True
         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
@@ -182,7 +176,7 @@ def close(self) -> None:
         if self._closed:
             return
         self._closed = True
-        if not (self.enabled and (self.pages or self.current_chunk)):
+        if not (self.enabled and self._rendered):
             return
         try:
             self._stream.write("\n")
@@ -195,10 +189,13 @@ def _maybe_hint_api_key(self) -> None:
         global _api_key_hint_shown
         if _api_key_hint_shown or os.getenv("API_USGS_PAT"):
             return
-        _api_key_hint_shown = True
+        # Set the once-per-process latch only after a successful write, so a
+        # failed write (broken pipe) doesn't silently burn the hint for every
+        # later query in the process.
         self._stream.write(
             f"No API key detected — register for higher rate limits at {SIGNUP_URL}\n"
         )
+        _api_key_hint_shown = True
 
 
 @contextmanager

dataretrieval/waterdata/utils.py

@@ -938,7 +938,7 @@ def _paginate(
         if reporter is not None:
             reporter.set_rate_remaining(
                 resp.headers.get(_QUOTA_HEADER),
-                reset=resp.headers.get("x-ratelimit-reset"),
+                limit=resp.headers.get("x-ratelimit-limit"),
             )
             reporter.add_page(rows=len(df))
         while cursor is not None:
@@ -951,7 +951,7 @@ def _paginate(
                 if reporter is not None:
                     reporter.set_rate_remaining(
                         resp.headers.get(_QUOTA_HEADER),
-                        reset=resp.headers.get("x-ratelimit-reset"),
+                        limit=resp.headers.get("x-ratelimit-limit"),
                     )
                     reporter.add_page(rows=len(df))
             except Exception as e:  # noqa: BLE001

tests/waterdata_progress_test.py

@@ -87,43 +87,22 @@ def test_missing_rate_limit_does_not_blank_last_known_value():
     assert "500 requests remaining" in stream.getvalue()
 
 
-def test_format_duration():
-    assert _progress._format_duration(0) == "0s"
-    assert _progress._format_duration(45) == "45s"
-    assert _progress._format_duration(600) == "10m"
-    assert _progress._format_duration(3600) == "1h"
-    assert _progress._format_duration(3700) == "1h01m"
-    assert _progress._format_duration(-5) == "0s"  # clamped
-
-
-def test_reset_countdown_rendered(monkeypatch):
-    # Pin the clock so the countdown is deterministic.
-    monkeypatch.setattr(_progress.time, "time", lambda: 1_000_000_000.0)
+def test_renders_remaining_over_limit():
     stream = io.StringIO()
     reporter = ProgressReporter(stream=stream, enabled=True)
-    reporter.set_rate_remaining("4870", reset="600")  # delta seconds -> 10m
+    reporter.set_rate_remaining("952", limit="1000")
     reporter.add_page(rows=1)
-    out = stream.getvalue()
-    assert "4,870 requests remaining, resets in 10m" in out
-
-
-def test_reset_accepts_absolute_epoch(monkeypatch):
-    monkeypatch.setattr(_progress.time, "time", lambda: 1_000_000_000.0)
-    stream = io.StringIO()
-    reporter = ProgressReporter(stream=stream, enabled=True)
-    reporter.set_rate_remaining("10", reset="1000000900")  # epoch: 900s out -> 15m
-    reporter.add_page()
-    assert "resets in 15m" in stream.getvalue()
+    assert "952 / 1,000 requests remaining" in stream.getvalue()
 
 
-def test_no_reset_segment_without_reset_header():
+def test_no_slash_when_limit_absent():
     stream = io.StringIO()
     reporter = ProgressReporter(stream=stream, enabled=True)
-    reporter.set_rate_remaining("4870")  # remaining only, no reset
+    reporter.set_rate_remaining("4870")  # remaining only, no limit header
     reporter.add_page()
     out = stream.getvalue()
     assert "4,870 requests remaining" in out
-    assert "resets in" not in out
+    assert "/" not in out
 
 
 def test_close_terminates_active_line_with_newline():