refactor(ogc): clarify chunking module boundaries (DOI-USGS#346)

Extract response and frame recombination into combining.py and colocate the private fetch/finalize contracts with ChunkedCall.

Centralize transient classification while preserving retry, resumability, exception snapshot, deduplication, nested GeoJSON, and response metadata contracts. Add regression coverage for those behaviors.

Align Ruff 0.16.1 across CI, pre-commit, and test metadata, and apply its Markdown code-fence formatting.

Author: thodson-usgs

.github/workflows/python-package.yml

@@ -20,7 +20,8 @@ jobs:
           python-version: "3.14"
           cache: "pip"
       - name: Install ruff
-        run: pip install ruff
+        # Keep this version aligned with the ruff-pre-commit revision.
+        run: pip install ruff==0.16.1
       - name: Lint with ruff
         run: |
           ruff check . --output-format=github

.pre-commit-config.yaml

@@ -21,7 +21,7 @@ repos:
       - id: debug-statements
 
   - repo: https://github.com/astral-sh/ruff-pre-commit
-    rev: v0.15.15
+    rev: v0.16.1
     hooks:
       - id: ruff-check
         args: [--fix]

CONTRIBUTING.md

@@ -149,19 +149,20 @@ via any automated processes or pipelines.
   * Example:
 
     ``` python
+    LIGHT_MESSAGES = {
+        "English": "There are %(number_of_lights)s lights.",
+        "Pirate": "Arr! Thar be %(number_of_lights)s lights.",
+    }
 
-        LIGHT_MESSAGES = {
-            'English': "There are %(number_of_lights)s lights.",
-            'Pirate':  "Arr! Thar be %(number_of_lights)s lights."
-        }
 
-        def lights_message(language, number_of_lights):
-            """Return a language-appropriate string reporting the light count."""
-            return LIGHT_MESSAGES[language] % locals()
+    def lights_message(language, number_of_lights):
+        """Return a language-appropriate string reporting the light count."""
+        return LIGHT_MESSAGES[language] % locals()
 
-        def is_pirate(message):
-            """Return True if the given message sounds piratical."""
-            return re.search(r"(?i)(arr|avast|yohoho)!", message) is not None
+
+    def is_pirate(message):
+        """Return True if the given message sounds piratical."""
+        return re.search(r"(?i)(arr|avast|yohoho)!", message) is not None
     ```
 
 ---

README.md

@@ -47,6 +47,7 @@ and set it as an environment variable:
 
 ```python
 import os
+
 os.environ["API_USGS_PAT"] = "your_api_key_here"
 ```
 
@@ -59,9 +60,9 @@ from dataretrieval import waterdata
 
 # Get daily streamflow data (returns DataFrame and metadata)
 df, metadata = waterdata.get_daily(
-    monitoring_location_id='USGS-01646500',
-    parameter_code='00060',  # Discharge
-    time='2024-10-01/2025-09-30'
+    monitoring_location_id="USGS-01646500",
+    parameter_code="00060",  # Discharge
+    time="2024-10-01/2025-09-30",
 )
 
 print(f"Retrieved {len(df)} records")
@@ -72,9 +73,9 @@ Retrieve streamflow at multiple locations from October 1, 2024 to the present:
 
 ```python
 df, metadata = waterdata.get_daily(
-    monitoring_location_id=["USGS-13018750","USGS-13013650"],
-    parameter_code='00060',
-    time='2024-10-01/..'
+    monitoring_location_id=["USGS-13018750", "USGS-13013650"],
+    parameter_code="00060",
+    time="2024-10-01/..",
 )
 
 print(f"Retrieved {len(df)} records")
@@ -85,8 +86,8 @@ stream sites in Maryland:
 ```python
 # Get monitoring location information
 df, metadata = waterdata.get_monitoring_locations(
-    state='Maryland',  # full name, postal code ('MD'), or FIPS ('24')
-    site_type_code='ST'  # Stream sites
+    state="Maryland",  # full name, postal code ('MD'), or FIPS ('24')
+    site_type_code="ST",  # Stream sites
 )
 
 print(f"Found {len(df)} stream monitoring locations in Maryland")
@@ -98,9 +99,9 @@ windows to avoid timeouts and other issues:
 ```python
 # Get continuous data for a single monitoring location and water year
 df, metadata = waterdata.get_continuous(
-    monitoring_location_id='USGS-01646500',
-    parameter_code='00065',  # Gage height
-    time='2024-10-01/2025-09-30'
+    monitoring_location_id="USGS-01646500",
+    parameter_code="00065",  # Gage height
+    time="2024-10-01/2025-09-30",
 )
 print(f"Retrieved {len(df)} continuous gage height measurements")
 ```
@@ -125,10 +126,10 @@ from dataretrieval import waterdata
 # enough to span many pages, so it profits from a finer split.
 sites, _ = waterdata.get_monitoring_locations(state="Ohio", site_type_code="ST")
 
-with waterdata.parallel_chunks(32):              # fan out into 32 sub-requests
+with waterdata.parallel_chunks(32):  # fan out into 32 sub-requests
     df, md = waterdata.get_daily(
         monitoring_location_id=sites["monitoring_location_id"],
-        parameter_code="00060",                  # discharge
+        parameter_code="00060",  # discharge
         time="2004-01-01/2023-12-31",
     )
 ```
@@ -167,6 +168,7 @@ API — enable debug-level
 
 ```python
 import logging
+
 logging.basicConfig(level=logging.DEBUG)
 ```
 
@@ -181,14 +183,14 @@ from dataretrieval import ngwmn
 
 # Find the groundwater monitoring sites in a state
 # (state accepts a full name, a postal code like 'WI', or a FIPS code like '55')
-sites, metadata = ngwmn.get_sites(state='Wisconsin')
+sites, metadata = ngwmn.get_sites(state="Wisconsin")
 
 print(f"Found {len(sites)} NGWMN sites in Wisconsin")
 
 # Pull water levels from the first twenty sites over a time window.
 water_levels, metadata = ngwmn.get_water_level(
-    monitoring_location_id=sites['monitoring_location_id'][:20],
-    datetime=['2022-01-01', '2024-01-01']
+    monitoring_location_id=sites["monitoring_location_id"][:20],
+    datetime=["2022-01-01", "2024-01-01"],
 )
 
 print(f"Retrieved {len(water_levels)} water-level observations")
@@ -203,16 +205,15 @@ from dataretrieval import wqp
 
 # Find water quality monitoring sites (returns a DataFrame and metadata)
 sites, metadata = wqp.what_sites(
-    statecode='US:55',  # Wisconsin
-    siteType='Stream'
+    statecode="US:55",  # Wisconsin
+    siteType="Stream",
 )
 
 print(f"Found {len(sites)} stream monitoring sites in Wisconsin")
 
 # Get water quality results
 results, metadata = wqp.get_results(
-    siteid='USGS-05427718',
-    characteristicName='Temperature, water'
+    siteid="USGS-05427718", characteristicName="Temperature, water"
 )
 
 print(f"Retrieved {len(results)} temperature measurements")
@@ -227,18 +228,18 @@ from dataretrieval import nldi
 
 # Get watershed basin for a stream reach
 basin = nldi.get_basin(
-    feature_source='comid',
-    feature_id='13293474'  # NHD reach identifier
+    feature_source="comid",
+    feature_id="13293474",  # NHD reach identifier
 )
 
 print(f"Basin contains {len(basin)} feature(s)")
 
 # Find upstream flowlines
 flowlines = nldi.get_flowlines(
-    feature_source='comid',
-    feature_id='13293474',
-    navigation_mode='UT',  # Upstream tributaries
-    distance=50  # km
+    feature_source="comid",
+    feature_id="13293474",
+    navigation_mode="UT",  # Upstream tributaries
+    distance=50,  # km
 )
 
 print(f"Found {len(flowlines)} upstream tributaries within 50km")
@@ -255,17 +256,17 @@ from dataretrieval import wateruse
 # Monthly public-supply withdrawals for Rhode Island, split into
 # groundwater and surface-water sources (returns a DataFrame and metadata).
 df, metadata = wateruse.get_wateruse(
-    model='wu-public-supply-wd',
-    variable=['pswdtot', 'pswdgw', 'pswdsw'],
-    state='RI',  # name/postal/FIPS; pass a list to fan out over several areas
-    start_date='2020-01',
-    time_resolution='monthly',
+    model="wu-public-supply-wd",
+    variable=["pswdtot", "pswdgw", "pswdsw"],
+    state="RI",  # name/postal/FIPS; pass a list to fan out over several areas
+    start_date="2020-01",
+    time_resolution="monthly",
 )
 
 print(f"Retrieved {len(df)} records across {df['huc12_id'].nunique()} watersheds")
 
 # Aggregate the HUC12 grid to a statewide monthly total (million gallons/day)
-statewide = df.groupby('year_month')['pswdtot_mgd'].sum()
+statewide = df.groupby("year_month")["pswdtot_mgd"].sum()
 print(statewide.head())
 ```
 

dataretrieval/ogc/chunking.py

@@ -74,7 +74,7 @@
 import asyncio
 import functools
 import os
-from collections.abc import Callable, Iterator
+from collections.abc import Awaitable, Callable, Iterator
 from contextlib import contextmanager
 from contextvars import copy_context
 from typing import Any, cast
@@ -86,17 +86,14 @@
 from dataretrieval.utils import HTTPX_DEFAULTS, Ambient, _require_positive_int
 
 from . import progress as _progress
-from .interruptions import (
-    ChunkInterrupted,
-    _Fetch,
-    _Finalize,
-    _passthrough_result,
-)
-from .planning import (
-    ChunkPlan,
+from .combining import (
     _combine_chunk_frames,
     _combine_chunk_responses,
 )
+from .interruptions import (
+    ChunkInterrupted,
+)
+from .planning import ChunkPlan
 from .retry import (
     _NO_RETRY,
     RetryPolicy,
@@ -291,6 +288,30 @@ def parallel_chunks(n: int) -> Iterator[None]:
         yield
 
 
+# ---------------------------------------------------------------------------
+# Type aliases for the ChunkedCall contract.
+# ---------------------------------------------------------------------------
+
+# The per-sub-request fetcher the decorator wraps and ``ChunkedCall`` drives:
+# an ``async def fetch(args) -> (df, response)``.
+_Fetch = Callable[[dict[str, Any]], Awaitable[tuple[pd.DataFrame, httpx.Response]]]
+
+# Caller-supplied transform applied to the combined chunk result, so a
+# resumed call returns the same shape as an un-interrupted one rather than
+# the chunker's raw ``(frame, httpx.Response)``.  This keeps the chunker
+# generic: the OGC getters inject their post-processing (type coercion,
+# column arrangement, ``BaseMetadata``) through ``_finalize_ogc``.
+# The default is identity, so direct ``ChunkedCall`` use is unaffected.
+_Finalize = Callable[[pd.DataFrame, httpx.Response], tuple[pd.DataFrame, Any]]
+
+
+def _passthrough_result(
+    frame: pd.DataFrame, response: httpx.Response
+) -> tuple[pd.DataFrame, Any]:
+    """Default :data:`_Finalize`: return the raw combined pair unchanged."""
+    return frame, response
+
+
 class ChunkedCall:
     """
     Stateful handle for a chunked call.
@@ -417,12 +438,11 @@ def _combine_raw(self) -> tuple[pd.DataFrame, httpx.Response]:
 
         Frames concatenate in sub-args *index* order (``sorted`` keys —
         deterministic, independent of parallel completion order). The
-        aggregated response takes its headers from the most-recently-
-        *completed* sub-request: the ``track`` closure in :meth:`_run`
-        is the only writer of ``self._chunks`` and ``dict`` preserves
-        insertion order, so the chunks' natural order is completion
-        order and the last one carries the freshest
-        ``x-ratelimit-remaining``.
+        aggregated response takes its headers from the response with the
+        lowest reported ``x-ratelimit-remaining`` value. If no response
+        reports that header, it falls back to the last completed response;
+        ``self._chunks`` preserves completion order because the ``track``
+        closure in :meth:`_run` is its only writer.
 
         Returns
         -------
@@ -521,8 +541,9 @@ def resume(self) -> tuple[pd.DataFrame, Any]:
             Combined data from every successful sub-request.
         response
             The finalized aggregate — a raw :class:`httpx.Response`
-            (canonical URL, most-recently-completed sub-request's headers,
-            cumulative elapsed time) by default, or whatever
+            (canonical URL, headers from the response with the lowest reported
+            remaining quota, and summed response elapsed durations) by default,
+            or whatever
             :attr:`finalize` produces (e.g. ``BaseMetadata`` for the OGC
             getters).
 
@@ -603,8 +624,9 @@ async def _run(self, max_concurrent: int | None) -> tuple[pd.DataFrame, Any]:
             Combined data from every sub-request.
         response
             The finalized aggregate — a raw :class:`httpx.Response`
-            (canonical URL, most-recently-completed sub-request's headers,
-            cumulative elapsed time) by default, or whatever
+            (canonical URL, headers from the response with the lowest reported
+            remaining quota, and summed response elapsed durations) by default,
+            or whatever
             :attr:`finalize` produces (e.g. ``BaseMetadata`` for OGC getters).
 
         Raises