feat(storage-tier): window on registration date via --date-field

lhoupert · claude · lhoupert · commit bc65a8aa9523 · 2026-07-08T13:08:05.000+01:00
The batch storage-tier submitter only windowed on sensing `datetime`, so it
could not target the S2 backlog by *registration* time (`properties.created`).
Bulk-converted items were registered long after they were sensed, so a
sensing-date window misses them.

Add `--date-field {datetime,created,updated}` (default `datetime`, backward
compatible). Non-default fields swap the `datetime=` range for a CQL2 `between`
filter on the chosen property — the same pattern `scripts/query_stac.py` uses
to harvest by `updated`.

Also:
- flip `--storage-class` default `STANDARD_IA` -&gt; `STANDARD` (STANDARD_IA is
  unused in the deployment; documented commands stay explicit regardless);
- add `timeout=30` to the webhook POST so a hung endpoint can't stall the
  per-window submission loop (ruff S113, on this script's own code);
- document the registered-window backlog example in operator-tools/README.md.

Plan: claude-docs/plans/s2_tier_standard_backlog.md (Task 2)

Co-Authored-By: Claude Opus 4.8 &lt;noreply@anthropic.com&gt;
diff --git a/operator-tools/README.md b/operator-tools/README.md
@@ -299,6 +299,18 @@ python operator-tools/submit_storage_tier_workflows.py \
     --collection sentinel-2-l2a-staging \
     --storage-class STANDARD_IA \
     --webhook-url http://localhost:12000/samples
+
+# Registered-window backlog — window on when items were *registered*
+# (properties.created) rather than sensed, to drain a bulk-conversion backlog
+# to STANDARD. Uses a CQL2 filter instead of the datetime= range.
+python operator-tools/submit_storage_tier_workflows.py \
+    --date-field created \
+    --start-date 2025-11-01 \
+    --end-date 2026-04-08 \
+    --collection sentinel-2-l2a \
+    --storage-class STANDARD \
+    --process-all-assets \
+    --dry-run
 ```
 
 **All options:**
@@ -308,7 +320,8 @@ python operator-tools/submit_storage_tier_workflows.py \
 | `--start-date` | required | Start of date range (`YYYY-MM-DD`) |
 | `--end-date` | required | End of date range (`YYYY-MM-DD`) |
 | `--collection` | required | STAC collection ID |
-| `--storage-class` | `STANDARD_IA` | Target S3 storage class |
+| `--date-field` | `datetime` | Item date to window on: sensing `datetime`, registration `created`, or last-modified `updated`. Non-default fields use a CQL2 `between` filter |
+| `--storage-class` | `STANDARD` | Target S3 storage class |
 | `--stac-api-url` | prod API URL | STAC API endpoint to query |
 | `--s3-endpoint` | OVH Cloud | S3 endpoint passed to Argo workflows |
 | `--pipeline-image-version` | `v1.6.1` | Docker image tag for Argo jobs |
diff --git a/operator-tools/submit_storage_tier_workflows.py b/operator-tools/submit_storage_tier_workflows.py
@@ -46,15 +46,36 @@ def generate_time_windows(
 
 
 def query_stac_items(
-    stac_api_url: str, collection: str, window_start: str, window_end: str
+    stac_api_url: str,
+    collection: str,
+    window_start: str,
+    window_end: str,
+    date_field: str = "datetime",
 ) -> list[str]:
-    """Query STAC for items with datetime in the given window. Returns list of item IDs."""
+    """Query STAC for items whose ``date_field`` falls in the window. Returns item IDs.
+
+    ``date_field="datetime"`` uses pystac-client's native ``datetime=`` range
+    (the item's sensing time). Other fields (``created``, ``updated``) filter on
+    the registration/update timestamp via a CQL2 ``between`` filter, since those
+    properties are not reachable through the ``datetime=`` kwarg.
+    """
     catalog = Client.open(stac_api_url)
-    search = catalog.search(
-        collections=[collection],
-        datetime=f"{window_start}/{window_end}",
-        limit=100,
-    )
+    if date_field == "datetime":
+        search = catalog.search(
+            collections=[collection],
+            datetime=f"{window_start}/{window_end}",
+            limit=100,
+        )
+    else:
+        search = catalog.search(
+            collections=[collection],
+            filter={
+                "op": "between",
+                "args": [{"property": date_field}, window_start, window_end],
+            },
+            filter_lang="cql2-json",
+            limit=100,
+        )
     return [item.id for item in search.items()]
 
 
@@ -68,6 +89,7 @@ def submit_batch(webhook_url: str, payload: dict[str, object], dry_run: bool) ->
             webhook_url,
             json=payload,
             headers={"Content-Type": "application/json"},
+            timeout=30,
         )
         if response.status_code != 200:
             logger.warning(
@@ -87,7 +109,14 @@ def main() -> None:
     parser.add_argument("--start-date", required=True, help="Start date (YYYY-MM-DD)")
     parser.add_argument("--end-date", required=True, help="End date (YYYY-MM-DD)")
     parser.add_argument("--collection", required=True, help="STAC collection ID")
-    parser.add_argument("--storage-class", default="STANDARD_IA", help="S3 storage class")
+    parser.add_argument(
+        "--date-field",
+        choices=["datetime", "created", "updated"],
+        default="datetime",
+        help="Item date to window on: sensing 'datetime' (default), registration "
+        "'created', or last-modified 'updated'. Non-default fields use a CQL2 filter.",
+    )
+    parser.add_argument("--storage-class", default="STANDARD", help="S3 storage class")
     parser.add_argument("--stac-api-url", default="https://api.explorer.eopf.copernicus.eu/stac")
     parser.add_argument("--s3-endpoint", default="https://s3.de.io.cloud.ovh.net")
     parser.add_argument("--pipeline-image-version", default="v1.6.1")
@@ -118,7 +147,9 @@ def main() -> None:
 
     for i, (window_start, window_end) in enumerate(windows, 1):
         logger.info(f"[{i}/{len(windows)}] Querying window {window_start} to {window_end}")
-        item_ids = query_stac_items(args.stac_api_url, args.collection, window_start, window_end)
+        item_ids = query_stac_items(
+            args.stac_api_url, args.collection, window_start, window_end, args.date_field
+        )
         logger.info(f"  Found {len(item_ids)} items")
 
         if not item_ids:
diff --git a/tests/unit/test_submit_storage_tier_workflows.py b/tests/unit/test_submit_storage_tier_workflows.py
@@ -100,6 +100,64 @@ def test_returns_empty_list_when_no_items(self) -> None:
 
         assert result == []
 
+    def test_date_field_created_uses_cql2_between(self) -> None:
+        """A non-default date_field switches to a CQL2 `between` filter on that property."""
+        mock_item = MagicMock()
+        mock_item.id = "item-a"
+        mock_search = MagicMock()
+        mock_search.items.return_value = [mock_item]
+
+        mock_catalog = MagicMock()
+        mock_catalog.search.return_value = mock_search
+
+        with patch("submit_storage_tier_workflows.Client") as mock_client:
+            mock_client.open.return_value = mock_catalog
+            result = query_stac_items(
+                "https://stac.example.com",
+                "sentinel-2",
+                "2024-01-01T00:00:00Z",
+                "2024-01-02T00:00:00Z",
+                date_field="created",
+            )
+
+        assert result == ["item-a"]
+        mock_catalog.search.assert_called_once_with(
+            collections=["sentinel-2"],
+            filter={
+                "op": "between",
+                "args": [
+                    {"property": "created"},
+                    "2024-01-01T00:00:00Z",
+                    "2024-01-02T00:00:00Z",
+                ],
+            },
+            filter_lang="cql2-json",
+            limit=100,
+        )
+
+    def test_default_date_field_uses_datetime_range(self) -> None:
+        """The default date_field keeps the native pystac-client datetime= range query."""
+        mock_search = MagicMock()
+        mock_search.items.return_value = []
+        mock_catalog = MagicMock()
+        mock_catalog.search.return_value = mock_search
+
+        with patch("submit_storage_tier_workflows.Client") as mock_client:
+            mock_client.open.return_value = mock_catalog
+            query_stac_items(
+                "https://stac.example.com",
+                "sentinel-2",
+                "2024-01-01T00:00:00Z",
+                "2024-01-02T00:00:00Z",
+                date_field="datetime",
+            )
+
+        mock_catalog.search.assert_called_once_with(
+            collections=["sentinel-2"],
+            datetime="2024-01-01T00:00:00Z/2024-01-02T00:00:00Z",
+            limit=100,
+        )
+
 
 class TestSubmitBatch:
     def test_dry_run_does_not_send_request(self) -> None:
@@ -236,6 +294,83 @@ def capture(url: str, payload: dict[str, object], dry_run: bool) -> bool:
         assert "parallelism" not in submitted_payloads[0]
 
 
+class TestMainDateFieldForwarding:
+    def test_date_field_forwarded_to_query(self) -> None:
+        """`--date-field created` reaches query_stac_items."""
+        captured: list[str] = []
+
+        def capture_query(
+            stac_api_url: str,
+            collection: str,
+            window_start: str,
+            window_end: str,
+            date_field: str,
+        ) -> list[str]:
+            captured.append(date_field)
+            return []
+
+        with (
+            patch(
+                "sys.argv",
+                [
+                    "submit_storage_tier_workflows.py",
+                    "--start-date",
+                    "2024-01-01",
+                    "--end-date",
+                    "2024-01-02",
+                    "--collection",
+                    "sentinel-2-l2a",
+                    "--date-field",
+                    "created",
+                    "--dry-run",
+                ],
+            ),
+            patch(
+                "submit_storage_tier_workflows.query_stac_items",
+                side_effect=capture_query,
+            ),
+        ):
+            from submit_storage_tier_workflows import main
+
+            main()
+
+        assert captured == ["created"]
+
+    def test_default_storage_class_is_standard(self) -> None:
+        """The submitted payload defaults storage_class to STANDARD."""
+        submitted_payloads: list[dict[str, object]] = []
+
+        def capture(url: str, payload: dict[str, object], dry_run: bool) -> bool:
+            submitted_payloads.append(payload)
+            return True
+
+        with (
+            patch(
+                "sys.argv",
+                [
+                    "submit_storage_tier_workflows.py",
+                    "--start-date",
+                    "2024-01-01",
+                    "--end-date",
+                    "2024-01-02",
+                    "--collection",
+                    "sentinel-2-l2a",
+                    "--dry-run",
+                ],
+            ),
+            patch(
+                "submit_storage_tier_workflows.query_stac_items",
+                return_value=["item-1"],
+            ),
+            patch("submit_storage_tier_workflows.submit_batch", side_effect=capture),
+        ):
+            from submit_storage_tier_workflows import main
+
+            main()
+
+        assert submitted_payloads[0]["storage_class"] == "STANDARD"
+
+
 class TestMainDateValidation:
     def test_end_before_start_exits(self) -> None:
         with patch(
