refactor(waterdata): Polish — extract _resolve_max_chunks, tidy iter_sub_args

Three micro-refinements after the previous pass settles. No behavior
change; 130 tests stay green.

- Extract _resolve_max_chunks() so the default + validation rule for
  ``max_chunks`` lives in one place, called from both _plan_list_chunks
  and _plan_joint. The 5-line if-None/if-<1 block was duplicated verbatim.
- _iter_sub_args drops its explicit ``list_keys = list(list_plan)``
  cache; iterating ``list_plan`` directly gives the same insertion-order
  sequence (Python 3.7+ dict guarantee), and ``zip(list_plan, combo)``
  reads as "pair each list-dim name with its chunk for this combo".
- Tighten the wrapper's option resolution to the "default if None else
  passed" form so each line reads in argument order.
- Categorize the _NEVER_CHUNK comment so future additions land in the
  right category instead of a flat narrative.

Author: thodson-usgs

dataretrieval/waterdata/chunking.py

@@ -51,14 +51,16 @@
     _split_top_level_or,
 )
 
-# Params the multi-value chunker must NOT split. ``properties`` defines
-# the response schema; chunking it would shard columns. ``bbox`` is a
-# fixed 4-element coord tuple. Date params are intervals not sets.
-# ``filter`` has its own OR-clause partitioning inside ``_plan_joint``;
-# treating it as a multi-value list would emit malformed CQL.
-# ``filter_lang`` is a scalar string. ``limit`` and ``skip_geometry``
-# are scalars by contract (smuggling a list through type erasure would
-# silently fan into queries with conflicting per-request semantics).
+# Params the multi-value chunker must NOT comma-join across sub-requests.
+# Categorized to make additions intentional:
+#   - response shape:  ``properties`` defines columns; sharding would
+#                       give different columns per chunk.
+#   - structured:      ``bbox`` is a fixed 4-element coord tuple.
+#   - intervals:       date/time ranges are not enumerable sets.
+#   - handled elsewhere: ``filter`` gets OR-clause partitioning in
+#                         ``_plan_joint``; multi-value semantics would
+#                         emit malformed CQL.
+#   - scalar by contract: ``limit``, ``skip_geometry``, ``filter_lang``.
 _NEVER_CHUNK = frozenset(
     {
         "properties",
@@ -213,6 +215,19 @@ def _plan_total(list_plan: dict[str, list[list[Any]]], n_filter_chunks: int) ->
     return list_count * n_filter_chunks
 
 
+def _resolve_max_chunks(max_chunks: int | None) -> int:
+    """Apply ``max_chunks`` default + validation. Both planner entry
+    points share this rule, so pinning it here keeps them in sync."""
+    if max_chunks is None:
+        return _DEFAULT_MAX_CHUNKS
+    if max_chunks < 1:
+        raise ValueError(
+            f"max_chunks must be >= 1; got {max_chunks}. Zero or negative "
+            f"values would silently bypass the cap."
+        )
+    return max_chunks
+
+
 def _worst_case_args(
     base_args: dict[str, Any], list_plan: dict[str, list[list[Any]]]
 ) -> dict[str, Any]:
@@ -263,13 +278,7 @@ def _plan_list_chunks(
     when the smallest reducible plan still doesn't fit or when the
     plan would exceed ``max_chunks`` sub-requests.
     """
-    if max_chunks is None:
-        max_chunks = _DEFAULT_MAX_CHUNKS
-    if max_chunks < 1:
-        raise ValueError(
-            f"max_chunks must be >= 1; got {max_chunks}. Zero or negative "
-            f"values would silently bypass the cap."
-        )
+    max_chunks = _resolve_max_chunks(max_chunks)
     chunkable = _chunkable_params(args)
     if not chunkable:
         return None
@@ -381,13 +390,7 @@ def _plan_joint(
     Raises ``RequestTooLarge`` when no candidate fits or the best plan
     would exceed ``max_chunks``.
     """
-    if max_chunks is None:
-        max_chunks = _DEFAULT_MAX_CHUNKS
-    if max_chunks < 1:
-        raise ValueError(
-            f"max_chunks must be >= 1; got {max_chunks}. Zero or negative "
-            f"values would silently bypass the cap."
-        )
+    max_chunks = _resolve_max_chunks(max_chunks)
 
     filter_expr = args.get("filter")
     clauses: list[str] = []
@@ -455,14 +458,11 @@ def _iter_sub_args(
     filter_chunks: list[str | None],
 ) -> Iterator[dict[str, Any]]:
     """Yield the substituted ``args`` for each sub-request in the joint
-    plan, in deterministic order: list-dim cartesian product (insertion
-    order, Python 3.7+ guarantee) crossed with filter chunks."""
-    list_keys = list(list_plan)
-    list_combos = (
-        itertools.product(*(list_plan[k] for k in list_keys)) if list_plan else [()]
-    )
+    plan, in deterministic order: list-dim cartesian product (dict
+    insertion order — Python 3.7+ guarantee) crossed with filter chunks."""
+    list_combos = itertools.product(*list_plan.values()) if list_plan else [()]
     for combo in list_combos:
-        base = {**args, **dict(zip(list_keys, combo))}
+        base = {**args, **dict(zip(list_plan, combo))}
         for filter_chunk in filter_chunks:
             yield base if filter_chunk is None else {**base, "filter": filter_chunk}
 
@@ -518,14 +518,12 @@ def wrapper(
             args: dict[str, Any],
         ) -> tuple[pd.DataFrame, requests.Response]:
             limit = (
-                url_limit
-                if url_limit is not None
-                else filters._WATERDATA_URL_BYTE_LIMIT
+                filters._WATERDATA_URL_BYTE_LIMIT if url_limit is None else url_limit
             )
             floor = (
-                quota_safety_floor
-                if quota_safety_floor is not None
-                else _DEFAULT_QUOTA_SAFETY_FLOOR
+                _DEFAULT_QUOTA_SAFETY_FLOOR
+                if quota_safety_floor is None
+                else quota_safety_floor
             )
             plan = _plan_joint(args, build_request, limit, max_chunks)
             if plan is None: