Syncing STAC Catalogs with pystac-client

pip install "pystac-client>=0.7.0" "pystac>=1.9.0" requests tenacity "pydantic>=2.0"

from pystac_client import Client

def initialize_stac_client(api_url: str) -> Client:
    """Open a STAC API client and assert required conformance classes."""
    stac_client = Client.open(api_url)
    required_conformance = {
        "https://api.stacspec.org/v1.0.0/core",
        "https://api.stacspec.org/v1.0.0/item-search",
    }
    supported = set(stac_client.get_conforms_to() or [])
    missing = required_conformance - supported
    if missing:
        raise RuntimeError(
            f"Endpoint {api_url!r} is missing conformance classes: {missing}"
        )
    return stac_client

stac_client = Client.open(
    "https://planetarycomputer.microsoft.com/api/stac/v1",
    headers={"Authorization": f"Bearer {token}"},
)

from datetime import datetime, timezone

def build_stac_query(
    bbox: tuple[float, float, float, float],
    start_dt: datetime,
    end_dt: datetime,
    collections: list[str],
    max_cloud_cover: float = 20.0,
) -> dict:
    """Build a STAC search parameter dict with spatial, temporal, and property filters."""
    return {
        "bbox": list(bbox),
        "datetime": f"{start_dt.isoformat()}/{end_dt.isoformat()}",
        "collections": collections,
        "query": {"eo:cloud_cover": {"lt": max_cloud_cover}},
        "limit": 100,  # Optimal page size for most public STAC providers
    }

import logging
from typing import Iterator

import requests
from pydantic import BaseModel, ValidationError
from tenacity import (
    retry,
    retry_if_exception_type,
    stop_after_attempt,
    wait_exponential,
)
from pystac import Item
from pystac_client import Client

logger = logging.getLogger(__name__)


class STACItemValidator(BaseModel):
    """Pydantic model enforcing minimum STAC Item structure."""
    id: str
    geometry: dict
    properties: dict
    assets: dict
    collection: str


@retry(
    stop=stop_after_attempt(4),
    wait=wait_exponential(multiplier=1, min=2, max=60),
    retry=retry_if_exception_type(
        (requests.exceptions.HTTPError, requests.exceptions.ConnectionError)
    ),
)
def stream_validated_items(
    stac_client: Client, query_params: dict
) -> Iterator[Item]:
    """Stream STAC Items page-by-page, validating structure before yielding."""
    search = stac_client.search(**query_params)
    for raw_item in search.items_as_dicts():
        try:
            STACItemValidator(**raw_item)
            yield Item.from_dict(raw_item)
        except ValidationError as exc:
            logger.warning(
                "Skipping malformed item %s: %s",
                raw_item.get("id", "<unknown>"),
                exc,
            )

import json
from pathlib import Path

import requests


def sync_item_assets(item: Item, output_root: Path) -> None:
    """Download all assets for a STAC Item, skipping already-present files."""
    item_dir = output_root / item.collection / item.id
    item_dir.mkdir(parents=True, exist_ok=True)

    for asset_key, asset in item.assets.items():
        # Derive a safe filename; respect original extension when present
        suffix = Path(asset.href.split("?")[0]).suffix or ".bin"
        dest = item_dir / f"{asset_key}{suffix}"

        if dest.exists():
            logger.debug("Asset already present, skipping: %s", dest)
            continue

        logger.info("Downloading %s / %s -> %s", item.id, asset_key, dest)
        response = requests.get(asset.href, stream=True, timeout=60)
        response.raise_for_status()

        tmp_dest = dest.with_suffix(".part")
        try:
            with tmp_dest.open("wb") as fh:
                for chunk in response.iter_content(chunk_size=65_536):
                    fh.write(chunk)
            tmp_dest.rename(dest)
        except Exception:
            tmp_dest.unlink(missing_ok=True)
            raise

    # Write JSON metadata sidecar alongside assets
    meta_path = item_dir / "metadata.json"
    with meta_path.open("w", encoding="utf-8") as fh:
        json.dump(item.to_dict(), fh, indent=2)
    logger.info("Metadata written: %s", meta_path)

def run_stac_sync(
    api_url: str,
    query_params: dict,
    output_dir: str,
) -> dict:
    """Run a full STAC catalog sync and return summary statistics."""
    stac_client = initialize_stac_client(api_url)
    output_root = Path(output_dir)
    stats = {"synced": 0, "skipped": 0, "errors": 0}

    for item in stream_validated_items(stac_client, query_params):
        try:
            sync_item_assets(item, output_root)
            stats["synced"] += 1
        except Exception as exc:  # noqa: BLE001
            logger.error("Failed to sync item %s: %s", item.id, exc)
            stats["errors"] += 1

    logger.info("Sync complete: %s", stats)
    return stats

import planetary_computer

signed_item = planetary_computer.sign(item)
# Now signed_item.assets[...].href contains a time-limited signed URL
sync_item_assets(signed_item, output_root)

WANTED_ROLES = {"data", "overview"}
WANTED_MEDIA = {"image/tiff; application=geotiff; profile=cloud-optimized"}

def filter_assets(item: Item) -> dict:
    """Return only assets matching desired roles or media types."""
    return {
        key: asset
        for key, asset in item.assets.items()
        if (
            set(getattr(asset, "roles", []) or []) & WANTED_ROLES
            or asset.media_type in WANTED_MEDIA
        )
    }

from shapely.geometry import box, mapping

def coerce_geometry(raw_item: dict) -> dict:
    """Replace null geometry with a polygon derived from bbox, if available."""
    if raw_item.get("geometry") is None and raw_item.get("bbox"):
        min_lon, min_lat, max_lon, max_lat = raw_item["bbox"]
        raw_item["geometry"] = mapping(box(min_lon, min_lat, max_lon, max_lat))
        logger.info("Coerced null geometry from bbox for item %s", raw_item.get("id"))
    return raw_item

import concurrent.futures
from functools import partial

def parallel_sync(
    api_url: str,
    query_params: dict,
    output_dir: str,
    max_workers: int = 4,
) -> dict:
    """Sync STAC assets across a thread pool while iterating items serially."""
    stac_client = initialize_stac_client(api_url)
    output_root = Path(output_dir)
    stats = {"synced": 0, "errors": 0}

    _sync = partial(sync_item_assets, output_root=output_root)

    with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as pool:
        futures = {
            pool.submit(_sync, item): item.id
            for item in stream_validated_items(stac_client, query_params)
        }
        for future in concurrent.futures.as_completed(futures):
            item_id = futures[future]
            try:
                future.result()
                stats["synced"] += 1
            except Exception as exc:  # noqa: BLE001
                logger.error("Parallel sync failed for %s: %s", item_id, exc)
                stats["errors"] += 1

    return stats

def sync_with_schema_gate(
    api_url: str,
    query_params: dict,
    output_dir: str,
    dead_letter_dir: str,
) -> None:
    """Run sync and route schema-invalid items to a dead-letter directory."""
    stac_client = initialize_stac_client(api_url)
    output_root = Path(output_dir)
    dead_letter = Path(dead_letter_dir)
    dead_letter.mkdir(parents=True, exist_ok=True)

    for item in stream_validated_items(stac_client, query_params):
        try:
            sync_item_assets(item, output_root)
        except Exception as exc:  # noqa: BLE001
            dl_path = dead_letter / f"{item.id}_error.json"
            dl_path.write_text(
                json.dumps({"id": item.id, "error": str(exc), "item": item.to_dict()}, indent=2)
            )
            logger.error("Item %s routed to dead-letter: %s", item.id, exc)