feat: create cli command to change storage tier

operator-tools/README.md

@@ -85,10 +85,18 @@ uv run operator-tools/manage_item.py info sentinel-2-l2a-staging ITEM_ID --s3-st
 # Debug S3 URL extraction
 uv run operator-tools/manage_item.py info sentinel-2-l2a-staging ITEM_ID --s3-stats --debug
 
-# Sync storage tiers for a single item (dry run)
+# Sync storage tier metadata for a single item (dry run)
 uv run operator-tools/manage_item.py sync-storage-tiers sentinel-2-l2a-staging ITEM_ID \
     --s3-endpoint https://s3.de.io.cloud.ovh.net --dry-run
 
+# Change storage tier for a single item (dry run - safe, no writes)
+uv run operator-tools/manage_item.py change-storage-tier sentinel-2-l2a-staging ITEM_ID \
+    --storage-class STANDARD_IA --s3-endpoint https://s3.de.io.cloud.ovh.net --dry-run
+
+# Change storage tier for a single item (requires confirmation)
+uv run operator-tools/manage_item.py change-storage-tier sentinel-2-l2a-staging ITEM_ID \
+    --storage-class STANDARD_IA --s3-endpoint https://s3.de.io.cloud.ovh.net -y
+
 # Delete single item with S3 cleanup (dry run)
 uv run operator-tools/manage_item.py delete sentinel-2-l2a-staging ITEM_ID --clean-s3 --dry-run
 
@@ -100,6 +108,7 @@ uv run operator-tools/manage_item.py delete sentinel-2-l2a-staging ITEM_ID --cle
 - Detailed item inspection with S3 statistics
 - Storage tier statistics from STAC metadata
 - Sync storage tiers with S3 (single item)
+- **Change S3 storage tier** for a single item and update STAC metadata
 - Debug mode for S3 URL extraction troubleshooting
 - Delete with automatic S3 validation
 - Dry-run mode for safe testing
@@ -149,10 +158,19 @@ uv run operator-tools/manage_collections.py info sentinel-2-l2a-staging --s3-sta
 # Debug S3 URL extraction
 uv run operator-tools/manage_collections.py info sentinel-2-l2a-staging --s3-stats --debug
 
-# Sync storage tiers for entire collection (dry run)
+# Sync storage tier metadata for entire collection (dry run)
 uv run operator-tools/manage_collections.py sync-storage-tiers sentinel-2-l2a-staging \
     --s3-endpoint https://s3.de.io.cloud.ovh.net --dry-run
 
+# Change storage tier for items in a date range (dry run - safe, no writes)
+uv run operator-tools/manage_collections.py change-storage-tier sentinel-2-l2a-staging \
+    --storage-class STANDARD_IA --start-date 2024-01-01 --end-date 2024-03-31 \
+    --s3-endpoint https://s3.de.io.cloud.ovh.net --dry-run
+
+# Change storage tier for all items in a collection (requires confirmation)
+uv run operator-tools/manage_collections.py change-storage-tier sentinel-2-l2a-staging \
+    --storage-class STANDARD_IA --s3-endpoint https://s3.de.io.cloud.ovh.net -y
+
 # Clean a collection (dry run first!)
 uv run operator-tools/manage_collections.py clean sentinel-2-l2a-staging --dry-run
 uv run operator-tools/manage_collections.py clean sentinel-2-l2a-staging
@@ -173,6 +191,7 @@ uv run operator-tools/manage_collections.py batch-create stac/
 - **Validated S3 cleanup** - Verifies all S3 objects deleted before removing STAC items
 - **Comprehensive S3 support** - Handles individual files, directories, and Zarr stores
 - **Sync storage tiers** - Keep STAC metadata in sync with S3 storage classes
+- **Change storage tier** - Move items to a different storage class, with date filtering
 - **Debug mode** - Detailed S3 URL extraction and validation info
 - **Safety first** - STAC items preserved if S3 cleanup fails
 
@@ -342,6 +361,61 @@ Sampling 5 of 43 items...
     Size: ~100.5 GB
 ```
 
+### Change S3 Storage Tier
+
+Move items to a different storage class (STANDARD, STANDARD_IA, EXPRESS_ONEZONE), then automatically update the STAC metadata to reflect the change.
+
+**Debug workflow** — always test a single item first:
+```bash
+ITEM_ID="S2A_MSIL2A_20250831T103701_N0511_R008_T31TFL_20250831T145420"
+
+# 1. Single item dry run (no writes)
+uv run operator-tools/manage_item.py change-storage-tier sentinel-2-l2a-staging $ITEM_ID \
+    --storage-class STANDARD_IA --s3-endpoint https://s3.de.io.cloud.ovh.net --dry-run
+
+# 2. Single item live run (prompts for confirmation)
+uv run operator-tools/manage_item.py change-storage-tier sentinel-2-l2a-staging $ITEM_ID \
+    --storage-class STANDARD_IA --s3-endpoint https://s3.de.io.cloud.ovh.net -y
+
+# 3. Verify: storage tier info should show the new class
+uv run operator-tools/manage_item.py info sentinel-2-l2a-staging $ITEM_ID --s3-stac-info
+```
+
+**Collection-level** — with optional date filtering:
+```bash
+# Dry run for a date range
+uv run operator-tools/manage_collections.py change-storage-tier sentinel-2-l2a-staging \
+    --storage-class STANDARD_IA --start-date 2024-01-01 --end-date 2024-03-31 \
+    --s3-endpoint https://s3.de.io.cloud.ovh.net --dry-run
+
+# Live run (prompts for confirmation showing item count)
+uv run operator-tools/manage_collections.py change-storage-tier sentinel-2-l2a-staging \
+    --storage-class STANDARD_IA --start-date 2024-01-01 --end-date 2024-03-31 \
+    --s3-endpoint https://s3.de.io.cloud.ovh.net -y
+
+# All items (no date filter)
+uv run operator-tools/manage_collections.py change-storage-tier sentinel-2-l2a-staging \
+    --storage-class STANDARD --s3-endpoint https://s3.de.io.cloud.ovh.net -y
+```
+
+**Options:**
+
+| Option | Description |
+|--------|-------------|
+| `--storage-class` | Target class: `STANDARD`, `STANDARD_IA`, or `EXPRESS_ONEZONE` (required) |
+| `--start-date` | Filter items on or after this date (`YYYY-MM-DD`) |
+| `--end-date` | Filter items on or before this date (`YYYY-MM-DD`) |
+| `--s3-endpoint` | S3 endpoint URL (falls back to `AWS_ENDPOINT_URL`) |
+| `--include-pattern` | fnmatch pattern for objects to include (repeatable) |
+| `--exclude-pattern` | fnmatch pattern for objects to exclude (repeatable) |
+| `--dry-run` | Show what would change without writing anything |
+| `-y` / `--yes` | Skip confirmation prompt |
+
+**Safety guarantees:**
+- S3 storage class is changed first; STAC metadata is only updated if S3 succeeds
+- Items with S3 failures are tracked and reported in the final summary — STAC is not touched
+- `--dry-run` propagates to S3 operations; nothing is written in either system
+
 ### Clean S3 Data (with Validation)
 
 Remove both STAC items and their associated S3 data with automatic validation:

operator-tools/manage_collections.py

@@ -1213,5 +1213,217 @@ def sync_storage_tiers(
         raise click.Abort() from e
 
 
+@cli.command()
+@click.argument("collection_id")
+@click.option(
+    "--storage-class",
+    required=True,
+    type=click.Choice(["STANDARD", "STANDARD_IA", "EXPRESS_ONEZONE"]),
+    help="Target S3 storage class",
+)
+@click.option(
+    "--start-date",
+    help="Start date filter, inclusive (YYYY-MM-DD)",
+)
+@click.option(
+    "--end-date",
+    help="End date filter, inclusive (YYYY-MM-DD)",
+)
+@click.option(
+    "--s3-endpoint",
+    help="S3 endpoint URL (required, uses AWS_ENDPOINT_URL env var if not specified)",
+)
+@click.option(
+    "--include-pattern",
+    "include_patterns",
+    multiple=True,
+    help="fnmatch pattern for objects to include (repeatable)",
+)
+@click.option(
+    "--exclude-pattern",
+    "exclude_patterns",
+    multiple=True,
+    help="fnmatch pattern for objects to exclude (repeatable)",
+)
+@click.option(
+    "--dry-run",
+    is_flag=True,
+    help="Show what would be changed without actually changing",
+)
+@click.option(
+    "--yes",
+    "-y",
+    is_flag=True,
+    help="Skip confirmation prompt",
+)
+@click.pass_context
+def change_storage_tier(
+    ctx: click.Context,
+    collection_id: str,
+    storage_class: str,
+    start_date: str | None,
+    end_date: str | None,
+    s3_endpoint: str | None,
+    include_patterns: tuple[str, ...],
+    exclude_patterns: tuple[str, ...],
+    dry_run: bool,
+    yes: bool,
+) -> None:
+    """
+    Change S3 storage tier for items in a collection, optionally filtered by date.
+
+    This command changes the S3 storage class for all objects in each item's
+    Zarr store, then updates STAC item metadata to reflect the new storage class.
+
+    Example:
+        manage_collections.py change-storage-tier sentinel-2-l2a-staging --storage-class STANDARD_IA --s3-endpoint https://s3.de.io.cloud.ovh.net --dry-run
+        manage_collections.py change-storage-tier sentinel-2-l2a-staging --storage-class STANDARD_IA --start-date 2024-01-01 --end-date 2024-03-31 --s3-endpoint https://s3.de.io.cloud.ovh.net -y
+    """
+    manager: STACCollectionManager = ctx.obj["manager"]
+    api_url: str = ctx.obj["api_url"]
+
+    # Get S3 endpoint
+    if not s3_endpoint:
+        s3_endpoint = os.getenv("AWS_ENDPOINT_URL")
+        if not s3_endpoint:
+            click.echo(
+                "❌ S3 endpoint required. Use --s3-endpoint or set AWS_ENDPOINT_URL", err=True
+            )
+            raise click.Abort()
+
+    # Validate date format early
+    from datetime import datetime
+
+    def _parse_date(date_str: str, param_name: str) -> None:
+        try:
+            datetime.strptime(date_str, "%Y-%m-%d")
+        except ValueError:
+            click.echo(f"❌ Invalid {param_name} format: '{date_str}'. Use YYYY-MM-DD.", err=True)
+            raise click.Abort() from None
+
+    if start_date:
+        _parse_date(start_date, "--start-date")
+    if end_date:
+        _parse_date(end_date, "--end-date")
+
+    try:
+        # Search items with optional date filter
+        catalog = Client.open(api_url)
+        if start_date or end_date:
+            start_str = f"{start_date}T00:00:00Z" if start_date else "1900-01-01T00:00:00Z"
+            end_str = f"{end_date}T23:59:59Z" if end_date else "2100-12-31T23:59:59Z"
+            search = catalog.search(
+                collections=[collection_id],
+                filter={"op": "between", "args": [{"property": "datetime"}, start_str, end_str]},
+                filter_lang="cql2-json",
+                limit=100,
+                max_items=None,
+            )
+        else:
+            search = catalog.search(collections=[collection_id], max_items=None)
+
+        items = list(search.items())
+
+        if not items:
+            click.echo("✅ No items matched the specified criteria.")
+            return
+
+        # Build date range message for confirmation
+        date_range_msg = ""
+        if start_date or end_date:
+            date_range_msg = f" ({start_date or 'any'} → {end_date or 'any'})"
+
+        # Confirmation prompt
+        if not dry_run and not yes:
+            click.confirm(
+                f"⚠️  This will change storage class for {len(items)} item(s) in '{collection_id}'"
+                f"{date_range_msg} to {storage_class}.\n\nContinue?",
+                abort=True,
+            )
+
+        click.echo(
+            f"\n{'DRY RUN: ' if dry_run else ''}Changing storage tier for {len(items)} item(s)"
+        )
+        click.echo(f"Collection: {collection_id}")
+        if date_range_msg:
+            click.echo(f"Date range: {date_range_msg.strip()}")
+        click.echo(f"Target storage class: {storage_class}")
+
+        from change_storage_tier import process_stac_item  # noqa: E402
+        from update_stac_storage_tier import update_item_storage_tiers  # noqa: E402
+
+        items_changed = 0
+        items_failed = 0
+        failed_item_ids: list[str] = []
+
+        with click.progressbar(
+            items, label="Processing items", show_pos=True, show_percent=True
+        ) as bar:
+            for item in bar:
+                item_id = item.id
+                stac_item_url = f"{manager.api_url}/collections/{collection_id}/items/{item_id}"
+
+                stats = process_stac_item(
+                    stac_item_url,
+                    storage_class,
+                    dry_run,
+                    s3_endpoint,
+                    list(include_patterns) or None,
+                    list(exclude_patterns) or None,
+                )
+
+                if stats["failed"] > 0:
+                    items_failed += 1
+                    failed_item_ids.append(item_id)
+                elif stats["processed"] > 0:
+                    if not dry_run:
+                        # Re-fetch and update STAC metadata
+                        item_dict = manager.item_manager.get_item(collection_id, item_id)
+                        if item_dict:
+                            pystac_item = Item.from_dict(item_dict)
+                            update_item_storage_tiers(pystac_item, s3_endpoint)
+
+                            # Use DELETE then POST (pgstac doesn't support PUT)
+                            delete_url = (
+                                f"{manager.api_url}" f"/collections/{collection_id}/items/{item_id}"
+                            )
+                            manager.session.delete(delete_url, timeout=30)
+                            create_url = f"{manager.api_url}" f"/collections/{collection_id}/items"
+                            manager.session.post(
+                                create_url,
+                                json=pystac_item.to_dict(),
+                                headers={"Content-Type": "application/json"},
+                                timeout=30,
+                            )
+                        else:
+                            items_failed += 1
+                            failed_item_ids.append(item_id)
+                            continue
+                    items_changed += 1
+
+        # Summary
+        click.echo("\n" + "=" * 60)
+        click.echo("CHANGE SUMMARY")
+        click.echo("=" * 60)
+        click.echo(f"Items processed: {len(items)}")
+        click.echo(f"✅ Items changed: {items_changed}")
+        if items_failed > 0:
+            click.echo(f"❌ Items failed: {items_failed}")
+            click.echo("\nFailed items:")
+            for fid in failed_item_ids:
+                click.echo(f"  - {fid}")
+
+        if dry_run:
+            click.echo(f"\n{'─'*60}")
+            click.echo("DRY RUN - No changes were made")
+            click.echo(f"{'─'*60}")
+
+        click.echo("=" * 60)
+
+    except Exception as e:
+        click.echo(f"❌ Operation failed: {e}", err=True)
+        raise click.Abort() from e
+
+
 if __name__ == "__main__":
     cli()