Refactor OGC and fan-out internals

dataretrieval/ogc/dates.py

@@ -164,3 +164,14 @@ def _format_api_dates(
             return None
         formatted.append(one)
     return "/".join(formatted)
+
+
+def _format_date_params(
+    params: dict[str, str | Sequence[str | None] | None], *, date_only: bool
+) -> None:
+    """Format every date-shaped OGC parameter in ``params`` in place."""
+    for key in _DATE_RANGE_PARAMS:
+        if key in params:
+            params[key] = _format_api_dates(
+                params[key], date=date_only and key != "last_modified"
+            )

dataretrieval/ogc/filters.py

@@ -90,6 +90,17 @@ def _resume_after_or(expr: str, i: int) -> int | None:
     return _skip_space(expr, after_word)
 
 
+def _advance_structure(
+    ch: str, in_quote: str | None, depth: int
+) -> tuple[str | None, int]:
+    """Advance quote and parenthesis state by one CQL character."""
+    if in_quote is not None:
+        return (None if ch == in_quote else in_quote), depth
+    if ch in ("'", '"'):
+        return ch, depth
+    return None, depth + {"(": 1, ")": -1}.get(ch, 0)
+
+
 def _split_top_level_or(expr: str) -> list[str]:
     """Split ``expr`` at each top-level ``OR``, respecting quotes and parens.
 
@@ -102,27 +113,18 @@ def _split_top_level_or(expr: str) -> list[str]:
     depth = 0
     in_quote: str | None = None
     i = 0
-    n = len(expr)
-    while i < n:
+    while i < len(expr):
         ch = expr[i]
-        if in_quote is not None:
-            if ch == in_quote:
-                in_quote = None
-        elif ch in ("'", '"'):
-            in_quote = ch
-        elif ch == "(":
-            depth += 1
-        elif ch == ")":
-            depth -= 1
-        elif depth == 0 and ch.isspace():
+        if in_quote is None and depth == 0 and ch.isspace():
             resume = _resume_after_or(expr, i + 1)
             if resume is not None:
                 parts.append(expr[last:i].strip())
                 last = i = resume
                 continue
+        in_quote, depth = _advance_structure(ch, in_quote, depth)
         i += 1
     parts.append(expr[last:].strip())
-    return [p for p in parts if p]
+    return [part for part in parts if part]
 
 
 def _check_numeric_filter_pitfall(filter_expr: str) -> None:

dataretrieval/ogc/planning.py

@@ -255,6 +255,26 @@ def _split_at(chunks: list[list[str]], idx: int) -> None:
     chunks[idx : idx + 1] = [chunk[:mid], chunk[mid:]]
 
 
+def _largest_splittable_chunk(
+    axes: list[_Axis],
+    chunks: dict[str, list[list[str]]],
+    *,
+    total: int,
+    max_chunks: int,
+) -> tuple[_Axis, int] | None:
+    """Select the largest chunk whose axis can split within the cap."""
+    candidate: tuple[_Axis, int] | None = None
+    candidate_size = -1
+    for axis in axes:
+        axis_chunks = chunks[axis.arg_key]
+        if total + total // len(axis_chunks) > max_chunks:
+            continue
+        for idx, chunk in enumerate(axis_chunks):
+            if len(chunk) > 1 and len(chunk) > candidate_size:
+                candidate, candidate_size = (axis, idx), len(chunk)
+    return candidate
+
+
 class ChunkPlan:
     """
     Strategy for issuing one user-level request as URL-fitting chunks.
@@ -506,17 +526,12 @@ def _refine(self, max_chunks: int) -> None:
             # work across chunks rather than fitting a byte budget. A
             # chunk of size 1 can't be split further. Stable input order breaks
             # ties by axis order, then lowest index within an axis.
-            candidate: tuple[_Axis, int] | None = None
-            candidate_size = -1
-            for axis in self.axes:
-                axis_chunks = self.chunks[axis.arg_key]
-                if total + total // len(axis_chunks) > max_chunks:
-                    continue  # any split of this axis would overshoot the cap
-                for idx, chunk in enumerate(axis_chunks):
-                    if len(chunk) <= 1:
-                        continue
-                    if len(chunk) > candidate_size:
-                        candidate, candidate_size = (axis, idx), len(chunk)
+            candidate = _largest_splittable_chunk(
+                self.axes,
+                self.chunks,
+                total=total,
+                max_chunks=max_chunks,
+            )
             if candidate is None:
                 # Every axis is saturated at one atom per chunk or would
                 # overshoot the cap; stop below it rather than exceed it.

dataretrieval/ogc/requests.py

@@ -17,7 +17,7 @@
 
 import httpx
 
-from dataretrieval.ogc.dates import _DATE_RANGE_PARAMS, _format_api_dates
+from dataretrieval.ogc.dates import _DATE_RANGE_PARAMS, _format_date_params
 from dataretrieval.ogc.policy import DEFAULT_DIALECT, OgcDialect
 from dataretrieval.transport.http import default_headers as _default_headers
 
@@ -160,14 +160,7 @@ def _construct_api_requests(
     service_url = _items_url(collection, base_url)
     if dialect is None:
         dialect = DEFAULT_DIALECT
-    for key in _DATE_RANGE_PARAMS:
-        if key in kwargs:
-            kwargs[key] = _format_api_dates(
-                kwargs[key],
-                date=(
-                    collection in dialect.date_only_services and key != "last_modified"
-                ),
-            )
+    _format_date_params(kwargs, date_only=collection in dialect.date_only_services)
     params, post_params = _partition_request_params(
         kwargs, use_cql2=collection in dialect.cql2_services
     )
@@ -292,6 +285,23 @@ def _check_monitoring_location_id(
     return value
 
 
+def _normalize_request_arg(name: str, value: Any, no_normalize: frozenset[str]) -> Any:
+    """Apply the OGC normalization rule for one included getter argument."""
+    if name == "monitoring_location_id":
+        return _check_monitoring_location_id(value)
+    if name == "properties":
+        return _as_str_list(value, name)
+    if (
+        name in no_normalize
+        and isinstance(value, Iterable)
+        and not isinstance(value, str)
+    ):
+        return value.tolist() if hasattr(value, "tolist") else list(value)
+    if isinstance(value, str) or not isinstance(value, Iterable):
+        return value
+    return _normalize_str_iterable(value, name)
+
+
 def prepare_request_args(
     local_vars: dict[str, Any],
     exclude: set[str] | None = None,
@@ -318,17 +328,8 @@ def prepare_request_args(
         to_exclude.update(exclude)
 
     args: dict[str, Any] = {}
-    for k, v in local_vars.items():
-        if k in to_exclude or v is None:
+    for name, value in local_vars.items():
+        if name in to_exclude or value is None:
             continue
-        if k == "monitoring_location_id":
-            args[k] = _check_monitoring_location_id(v)
-        elif k == "properties":
-            args[k] = _as_str_list(v, k)
-        elif k in no_normalize and isinstance(v, Iterable) and not isinstance(v, str):
-            args[k] = v.tolist() if hasattr(v, "tolist") else list(v)
-        elif isinstance(v, str) or not isinstance(v, Iterable):
-            args[k] = v
-        else:
-            args[k] = _normalize_str_iterable(v, k)
+        args[name] = _normalize_request_arg(name, value, no_normalize)
     return args

dataretrieval/transport/fanout.py

@@ -666,31 +666,34 @@ async def track(
                     return_exceptions=True,
                 )
                 failures = [r for r in results if isinstance(r, BaseException)]
-                for exc in failures:
-                    if not isinstance(exc, Exception):
-                        raise exc
-                # Classify first, build once. Every failure has to be
-                # examined -- a non-transient sibling must surface raw -- but
-                # only the first transient is ever raised. Asking
-                # ``wrap_failure`` per failure would snapshot the combined
-                # frame N times (a full concat over every completed
-                # chunk) and discard all but one, which a batch of
-                # chunks failing together makes routine.
-                first_transient: BaseException | None = None
-                for exc in failures:
-                    if _classify_chunk_error(exc) is None:
-                        raise self._normalize_failure(exc)
-                    if first_transient is None:
-                        first_transient = exc
-                if first_transient is not None:
-                    interrupted = self.wrap_failure(first_transient)
-                    if interrupted is None:
-                        # Unreachable: classified as transient just above.
-                        raise self._normalize_failure(first_transient)
-                    raise interrupted from first_transient
+                self._raise_failures(failures)
 
         return self.finalize(*self._combine_raw())
 
+    def _raise_failures(self, failures: list[BaseException]) -> None:
+        """Raise failures according to fan-out's precedence rules."""
+        for exc in failures:
+            if not isinstance(exc, Exception):
+                raise exc
+        # Classify first, build once. Every failure has to be examined -- a
+        # non-transient sibling must surface raw -- but only the first transient
+        # is ever raised. Asking ``wrap_failure`` per failure would snapshot the
+        # combined frame N times (a full concat over every completed chunk) and
+        # discard all but one, which a batch of chunks failing together makes
+        # routine.
+        first_transient: BaseException | None = None
+        for exc in failures:
+            if _classify_chunk_error(exc) is None:
+                raise self._normalize_failure(exc)
+            if first_transient is None:
+                first_transient = exc
+        if first_transient is not None:
+            interrupted = self.wrap_failure(first_transient)
+            if interrupted is None:
+                # Unreachable: classified as transient just above.
+                raise self._normalize_failure(first_transient)
+            raise interrupted from first_transient
+
 
 __all__ = [
     "FanOut",