diff --git a/dataretrieval/ogc/dates.py b/dataretrieval/ogc/dates.py index 248e667c..b3cc973d 100644 --- a/dataretrieval/ogc/dates.py +++ b/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" + ) diff --git a/dataretrieval/ogc/filters.py b/dataretrieval/ogc/filters.py index 50875409..a918fc5a 100644 --- a/dataretrieval/ogc/filters.py +++ b/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: diff --git a/dataretrieval/ogc/planning.py b/dataretrieval/ogc/planning.py index e5af5411..d3796c7d 100644 --- a/dataretrieval/ogc/planning.py +++ b/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. diff --git a/dataretrieval/ogc/requests.py b/dataretrieval/ogc/requests.py index d3b76ae4..c7c6ee14 100644 --- a/dataretrieval/ogc/requests.py +++ b/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 diff --git a/dataretrieval/transport/fanout.py b/dataretrieval/transport/fanout.py index 53b46c8a..d22de05f 100644 --- a/dataretrieval/transport/fanout.py +++ b/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",