Merge remote-tracking branch 'origin/master' into issue1565-extract-extent-from-constraints-with-caching

Author: soxofaan

openeogeotrellis/_backend/post_dry_run.py

@@ -346,20 +346,50 @@ def _extract_spatial_extent_from_load_stac_item_collection(
         for item, band_assets in item_collection.iter_items_with_band_assets()
         for asset in band_assets.values()
     ]
-    _log.info(f"Collected {len(item_collection.items)} items, {len(projection_metadatas)} projection metadata entries")
+
+    # Deduplicate: assets with identical projection properties produce identical
+    # bbox/coverage/resolution results. E.g. Sentinel-2 items have ~43 assets per item
+    # that often share the same proj:bbox.
+    seen_keys = set()
+    unique_projection_metadatas = []
+    for p in projection_metadatas:
+        key = (p._code, p._bbox, p._shape, p._transform)
+        if key not in seen_keys:
+            seen_keys.add(key)
+            unique_projection_metadatas.append(p)
+    _log.info(
+        f"Collected {len(projection_metadatas)} projection metadata entries"
+        f" from {len(item_collection.items)} items,"
+        f" deduplicated to {len(unique_projection_metadatas)} unique entries"
+    )
 
     # Determine most common grid (CRS and resolution) among assets
-    target_grid = _determine_best_grid_from_proj_metadata(projection_metadatas)
+    target_grid = _determine_best_grid_from_proj_metadata(unique_projection_metadatas)
     target_crs = target_grid.crs_raw if target_grid else None
     _log.info(f"Determined {target_grid=}")
 
-    # Merge asset bounding boxes (full native extent, and "aligned" part of covered extent)
+    # Merge asset bounding boxes (full native extent, and "aligned" part of covered extent).
+    # Batch by CRS: compute min/max within each CRS group to avoid
+    # per-element reprojection overhead in the merger.
     assets_full_bbox_merger = BoundingBoxMerger(crs=target_crs)
-    aligned_extent_coverage_merger = BoundingBoxMerger(crs=target_crs)
-    for proj_metadata in projection_metadatas:
+    crs_bbox_groups: Dict[Union[str, None], List[BoundingBox]] = collections.defaultdict(list)
+    for proj_metadata in unique_projection_metadatas:
         if asset_bbox := proj_metadata.to_bounding_box():
-            assets_full_bbox_merger.add(asset_bbox)
-            if extent_orig and (extent_coverage := proj_metadata.coverage_for(extent_orig)):
+            crs_bbox_groups[asset_bbox.crs].append(asset_bbox)
+    for crs, bboxes in crs_bbox_groups.items():
+        merged_group = BoundingBox(
+            west=min(b.west for b in bboxes),
+            south=min(b.south for b in bboxes),
+            east=max(b.east for b in bboxes),
+            north=max(b.north for b in bboxes),
+            crs=crs,
+        )
+        assets_full_bbox_merger.add(merged_group)
+
+    aligned_extent_coverage_merger = BoundingBoxMerger(crs=target_crs)
+    if extent_orig:
+        for proj_metadata in unique_projection_metadatas:
+            if extent_coverage := proj_metadata.coverage_for(extent_orig):
                 aligned_extent_coverage_merger.add(extent_coverage)
     assets_full_bbox = assets_full_bbox_merger.get()
     assets_covered_bbox = aligned_extent_coverage_merger.get()

openeogeotrellis/integrations/stac.py

@@ -57,7 +57,7 @@ def read_text_from_href(self, href: str) -> str:
 
 class LoggingStacApiIO(StacApiIO):
     """
-    StacApiIO subclass that logs every HTTP request at INFO level,
+    StacApiIO subclass that logs every HTTP request,
     including the method, URL, response status code, and elapsed time.
 
     Uses a requests response hook so every request made through the session

openeogeotrellis/load_stac.py

@@ -6,6 +6,7 @@
 import functools
 import logging
 import os
+import random
 import re
 import time
 from dataclasses import dataclass
@@ -64,6 +65,26 @@
 STAC_API_PER_PAGE_LIMIT_DEFAULT = 100
 
 
+class _JitteredRetry(Retry):
+    """Retry with jitter to avoid thundering herd on 429 responses.
+
+    - No Retry-After header: full jitter (random in [0, base_backoff])
+    - Retry-After header present: respects it as a minimum with full jitter
+    """
+
+    def get_backoff_time(self) -> float:
+        base = super().get_backoff_time()
+        return random.uniform(0, base)
+
+    def sleep_for_retry(self, response=None) -> bool:
+        retry_after = self.get_retry_after(response)
+        if retry_after is not None:
+            jitter = random.uniform(0, super().get_backoff_time())
+            time.sleep(retry_after + jitter)
+            return True
+        return False
+
+
 class NoDataAvailableException(OpenEOApiException):
     status_code = 400
     code = "NoDataAvailable"
@@ -1236,8 +1257,8 @@ def from_stac_api(
             # https://stac.openeo.vito.be/ and https://stac.terrascope.be
             fields = None
 
-        retry = requests.adapters.Retry(
-            total=4,
+        retry = _JitteredRetry(
+            total=7,
             backoff_factor=2,
             status_forcelist=frozenset([429, 500, 502, 503, 504]),
             allowed_methods=Retry.DEFAULT_ALLOWED_METHODS.union({"POST"}),

tests/conftest.py

@@ -5,6 +5,7 @@
 import shutil
 import sys
 import typing
+import urllib.parse
 from datetime import datetime
 from glob import glob
 from pathlib import Path
@@ -452,6 +453,8 @@ class UrllibAndRequestMocker:
     def __init__(self, urllib_mock, requests_mock):
         self.urllib_mock = urllib_mock
         self.requests_mock = requests_mock
+        self._flexible_responses: dict[str, dict[tuple, bytes]] = {}
+        self._flexible_ignore: dict[str, frozenset] = {}
 
     def get(self, href, data):
         code = 200
@@ -460,6 +463,71 @@ def get(self, href, data):
             data = data.encode("utf-8")
         self.requests_mock.get(href, content=data)
 
+    @staticmethod
+    def _normalize_params(url: str, ignore_params: frozenset) -> tuple:
+        """Parse URL query string, remove ignored params, return hashable normalized form."""
+        parsed = urllib.parse.urlparse(url)
+        params = urllib.parse.parse_qs(parsed.query, keep_blank_values=True)
+        for p in ignore_params:
+            params.pop(p, None)
+        return tuple(sorted((k, tuple(sorted(vs))) for k, vs in params.items()))
+
+    def get_flexible(self, href: str, data, ignore_params=()):
+        """Register a GET mock that matches URLs ignoring specified query parameters.
+        Example usage:
+            mock.get_flexible(
+                "https://stac.test/search?limit=100&bbox=1,2,3,4&collections=foo",
+                data=my_response,
+                ignore_params=("limit",),
+            )
+
+        All registrations for the same base URL path share a single dispatcher
+        callback, so call get_flexible() for **all** URLs under a given path
+        (don't mix with get() for the same base path).
+        """
+        ignore = frozenset(ignore_params)
+        parsed = urllib.parse.urlparse(href)
+        base_url = urllib.parse.urlunparse((parsed.scheme, parsed.netloc, parsed.path, "", "", ""))
+        norm_key = self._normalize_params(href, ignore)
+
+        if isinstance(data, str):
+            data_bytes = data.encode("utf-8")
+        else:
+            data_bytes = data
+
+        if base_url not in self._flexible_responses:
+            self._flexible_responses[base_url] = {}
+            self._flexible_ignore[base_url] = ignore
+
+        self._flexible_responses[base_url][norm_key] = data_bytes
+        self._install_flexible_handler(base_url)
+
+    def _install_flexible_handler(self, base_url: str):
+        responses = self._flexible_responses[base_url]
+        ignore = self._flexible_ignore[base_url]
+        normalize = self._normalize_params
+
+        def lookup(url: str) -> typing.Optional[bytes]:
+            key = normalize(url, ignore)
+            return responses.get(key)
+
+        def urllib_handler(req):
+            data = lookup(req.full_url)
+            if data is not None:
+                return UrllibMocker.Response(data=data)
+            return UrllibMocker.Response(code=404, msg=f"No flexible match for {req.full_url}")
+
+        self.urllib_mock.register(method="GET", url=base_url, response=urllib_handler)
+
+        def requests_handler(request, context):
+            data = lookup(request.url)
+            if data is not None:
+                return data
+            context.status_code = 404
+            return b"Not Found"
+
+        self.requests_mock.get(base_url, content=requests_handler)
+
 
 @pytest.fixture
 def urllib_and_request_mock(urllib_mock, requests_mock) -> UrllibAndRequestMocker:

tests/test_api_result.py

@@ -3748,22 +3748,30 @@ def item_json(path):
             "https://stac.test/collections/sentinel-2-l2a",
             data=get_test_data_file("stac/issue830_alternate_url/collections_sentinel-2-l2a.json").read_text(),
         )
-        urllib_and_request_mock.get(
-            "https://stac.test/search", data=item_json("stac/issue830_alternate_url/search.json")
+        # Use get_flexible for search URLs to avoid coupling to the page limit default
+        urllib_and_request_mock.get_flexible(
+            "https://stac.test/search", data=item_json("stac/issue830_alternate_url/search.json"),
+            ignore_params=("limit",),
         )
-        urllib_and_request_mock.get(
-            "https://stac.test/search?limit=20&bbox=5.07%2C51.215%2C5.08%2C51.22&datetime=2024-06-23T00%3A00%3A00Z%2F2024-06-23T23%3A59%3A59.999000Z&collections=sentinel-2-l2a&fields=%2Bproperties.proj%3Abbox%2C%2Bproperties.proj%3Aepsg%2C%2Bproperties.proj%3Ashape",
-            data=item_json("stac/issue830_alternate_url/search_queried.json"))
-        urllib_and_request_mock.get(
-            "https://stac.test/search?limit=20&bbox=5.07%2C51.215%2C5.08%2C51.22&datetime=2024-06-16T00%3A00%3A00Z%2F2024-06-23T23%3A59%3A59.999000Z&collections=sentinel-2-l2a&fields=%2Bproperties.proj%3Abbox%2C%2Bproperties.proj%3Ashape%2C%2Bproperties.proj%3Aepsg&token=MTcxOTEzOTU3OTAyNCxTMkJfTVNJTDJBXzIwMjQwNjIzVDEwNDYxOV9OMDUxMF9SMDUxX1QzMVVGU18yMDI0MDYyM1QxMjIxNTYsc2VudGluZWwtMi1sMmE%3D",
-            data=item_json("stac/issue830_alternate_url/search_queried_page2.json"))
-        urllib_and_request_mock.get(
-            "https://stac.test/search?limit=20&bbox=5.07%2C51.215%2C5.08%2C51.22&datetime=2024-06-23T00%3A00%3A00Z%2F2024-06-23T23%3A59%3A59.999000Z&collections=sentinel-2-l2a&fields=%2Btype%2C%2Bgeometry%2C%2Bproperties%2C%2Bid%2C%2Bbbox%2C%2Bstac_version%2C%2Bassets%2C%2Blinks%2C%2Bcollection",
+        urllib_and_request_mock.get_flexible(
+            "https://stac.test/search?limit=100&bbox=5.07%2C51.215%2C5.08%2C51.22&datetime=2024-06-23T00%3A00%3A00Z%2F2024-06-23T23%3A59%3A59.999000Z&collections=sentinel-2-l2a&fields=%2Bproperties.proj%3Abbox%2C%2Bproperties.proj%3Aepsg%2C%2Bproperties.proj%3Ashape",
             data=item_json("stac/issue830_alternate_url/search_queried.json"),
+            ignore_params=("limit",),
         )
-        urllib_and_request_mock.get(
-            "https://stac.test/search?limit=20&bbox=5.07%2C51.215%2C5.08%2C51.22&datetime=2024-06-23T00%3A00%3A00Z%2F2024-06-23T23%3A59%3A59.999000Z&collections=sentinel-2-l2a",
+        urllib_and_request_mock.get_flexible(
+            "https://stac.test/search?limit=100&bbox=5.07%2C51.215%2C5.08%2C51.22&datetime=2024-06-16T00%3A00%3A00Z%2F2024-06-23T23%3A59%3A59.999000Z&collections=sentinel-2-l2a&fields=%2Bproperties.proj%3Abbox%2C%2Bproperties.proj%3Ashape%2C%2Bproperties.proj%3Aepsg&token=MTcxOTEzOTU3OTAyNCxTMkJfTVNJTDJBXzIwMjQwNjIzVDEwNDYxOV9OMDUxMF9SMDUxX1QzMVVGU18yMDI0MDYyM1QxMjIxNTYsc2VudGluZWwtMi1sMmE%3D",
+            data=item_json("stac/issue830_alternate_url/search_queried_page2.json"),
+            ignore_params=("limit",),
+        )
+        urllib_and_request_mock.get_flexible(
+            "https://stac.test/search?limit=100&bbox=5.07%2C51.215%2C5.08%2C51.22&datetime=2024-06-23T00%3A00%3A00Z%2F2024-06-23T23%3A59%3A59.999000Z&collections=sentinel-2-l2a&fields=%2Btype%2C%2Bgeometry%2C%2Bproperties%2C%2Bid%2C%2Bbbox%2C%2Bstac_version%2C%2Bassets%2C%2Blinks%2C%2Bcollection",
+            data=item_json("stac/issue830_alternate_url/search_queried.json"),
+            ignore_params=("limit",),
+        )
+        urllib_and_request_mock.get_flexible(
+            "https://stac.test/search?limit=100&bbox=5.07%2C51.215%2C5.08%2C51.22&datetime=2024-06-23T00%3A00%3A00Z%2F2024-06-23T23%3A59%3A59.999000Z&collections=sentinel-2-l2a",
             data=item_json("stac/issue830_alternate_url/search_queried.json"),
+            ignore_params=("limit",),
         )
 
         process_graph = {
@@ -3842,9 +3850,10 @@ def item_json(path):
                 "stac/issue830_alternate_url_s3/catalogue.dataspace.copernicus.eu/stac/index.json"
             ).read_text(),
         )
-        urllib_and_request_mock.get(
+        urllib_and_request_mock.get_flexible(
             "https://catalogue.dataspace.copernicus.eu/stac/search?limit=20&bbox=5.07%2C51.215%2C5.08%2C51.22&datetime=2023-06-01T00%3A00%3A00Z%2F2023-06-30T23%3A59%3A59.999000Z&collections=GLOBAL-MOSAICS",
             data=item_json("stac/issue830_alternate_url_s3/catalogue.dataspace.copernicus.eu/stac/search_queried.json"),
+            ignore_params=("limit",),
         )
 
         process_graph = {
@@ -4392,7 +4401,11 @@ def feature_collection(request, _) -> dict:
             assert "fields" not in request.qs
             assert request.qs.get("filter-lang") == filter_lang
             assert request.qs.get("filter") == filter
-            assert request.body == body or request.json() == body
+            if body:
+                for k, v in body.items():
+                    if k == "limit":
+                        continue
+                    assert (isinstance(request.body, dict) and request.body.get(k) == v) or request.json().get(k) == v
 
             def item(path) -> dict:
                 return json.loads(
@@ -4748,9 +4761,10 @@ def item_json(path):
                 "stac/issue_copernicus_global_mosaics/stac.dataspace.copernicus.eu/v1/collections/sentinel-2-global-mosaics.json"
             ),
         )
-        urllib_and_request_mock.get(
+        urllib_and_request_mock.get_flexible(
             "https://stac.dataspace.copernicus.eu/v1/search?limit=20&bbox=2.1%2C35.31%2C2.2%2C35.32&datetime=2023-01-01T00%3A00%3A00Z%2F2023-01-01T23%3A59%3A59.999000Z&collections=sentinel-2-global-mosaics",
             data=item_json("stac/issue_copernicus_global_mosaics/stac.dataspace.copernicus.eu/v1/search_queried.json"),
+            ignore_params=("limit",),
         )
         urllib_and_request_mock.get(
             "https://stac.dataspace.copernicus.eu/v1/",