This guide is part of Mastering Geospatial Data Ingestion in Python, the reference for building production-grade Python pipelines that ingest, validate, and load spatial data at scale.

Syncing STAC Catalogs with pystac-client

The SpatioTemporal Asset Catalog (STAC) specification has become the de facto standard for organizing, discovering, and accessing geospatial raster and vector assets. For data engineers and GIS analysts building automated ingestion pipelines, manually traversing STAC endpoints is inefficient and fragile: undocumented pagination quirks, unsigned cloud asset URLs, and inconsistent conformance implementations silently break naive traversal scripts within hours of going to production. Syncing STAC catalogs with pystac-client provides a Python-native interface to query, paginate, validate, and extract metadata and assets at scale — one that survives API drift and cloud-storage permission changes without manual intervention.

The diagram below maps the full sync lifecycle, from client initialization through asset persistence:

Prerequisites & Environment Configuration

Before implementing catalog synchronization, ensure your environment meets the following baseline requirements:

• Python 3.10+ — required for structural pattern matching, modern type hints, and compatibility with pystac-client>=0.7.
• Core dependencies — pystac-client>=0.7.0, pystac>=1.9.0, requests>=2.31.0, pydantic>=2.0, and tenacity>=8.2 for resilient retry logic.
• Network access — unrestricted outbound HTTPS to STAC API endpoints such as https://earth-search.aws.element84.com/v1 (AWS Element 84) or https://planetarycomputer.microsoft.com/api/stac/v1 (Microsoft Planetary Computer).
• Storage target — a local filesystem path, S3 bucket, or cloud storage mount with write permissions and the appropriate IAM credentials configured in the execution environment.

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

Understanding the STAC Item, Collection, and Catalog JSON structures is essential — specifically, how STAC separates structured metadata from binary asset HREFs. This separation is what makes STAC sync patterns fundamentally different from plain HTTP directory traversal.

Version / Compatibility Matrix

Always pin both pystac-client and pystac together; they share internal model classes that break silently on version mismatches.

Step-by-Step Implementation

Step 1 — Ingest: Client Initialization & Conformance Validation

STAC APIs are not uniform. Some endpoints implement the full ItemSearch and Collections conformance classes; others expose only legacy catalog browsing via link traversal. Failing to validate conformance before issuing search queries results in cryptic 404 or 405 errors that are hard to distinguish from transient network failures.

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

Pass authentication headers directly to Client.open() when the endpoint requires tokens:

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

Step 2 — Diagnose: Query Construction & Spatial-Temporal Filtering

STAC queries use RFC 3339 timestamps, WGS84 bounding boxes, and CQL2 property filters. Constructing precise queries minimizes payload size and prevents the API from triggering rate limits on open-ended scans. Always set explicit temporal windows and spatial extents — unbounded queries frequently trigger server-side query-cost limits and return stale historical records.

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
    }

bbox follows the GeoJSON convention: [min_lon, min_lat, max_lon, max_lat] in EPSG:4326. Before passing a bbox derived from a projected dataset, apply CRS normalization to a unified reference system — submitting a projected bbox against a WGS84-native API silently returns zero results.

Step 3 — Retrieve: Paginated Item Streaming

The pystac-client .items() generator abstracts cursor-based pagination, but you still need explicit backoff for 429 Too Many Requests responses and should follow STAC catalog pagination best practices to avoid memory accumulation on large collections.

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,
            )

Note that @retry wraps the entire generator initialization, not individual yield calls. For per-item retries on download failures, implement a separate decorated function (see Step 4).

Step 4 — Repair/Transform: Asset Resolution & Idempotent Download

Once items pass schema validation, resolve asset HREFs and download them in streaming chunks. Idempotency is critical: a sync job that re-downloads existing files on every run wastes bandwidth and masks real delta-detection logic. Check for file presence before issuing any HTTP request.

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)

The .part suffix pattern ensures that an interrupted download never leaves a truncated file that passes the dest.exists() check on the next run.

Step 5 — Validate & Log: End-to-End Sync Orchestration

Tie the stages together into a single callable entry point. Logging should emit structured fields (item ID, collection, asset count) rather than free-form strings so orchestration platforms like Airflow or Prefect can parse them into metrics.

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

Advanced Patterns & Edge Cases

Handling Cloud-Native Asset URLs (Signed URLs & Requester-Pays)

Many STAC providers — including AWS Earth Search and Microsoft Planetary Computer — store assets in cloud buckets that require signed URLs or requester-pays headers. A 403 Forbidden on an asset URL almost always means the asset needs signing before download, not that the credentials are wrong.

For Planetary Computer, install planetary-computer and sign items before iterating assets:

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)

For AWS Earth Search using requester-pays, configure your environment with AWS_REQUEST_PAYER=requester and use boto3 to stream assets directly to S3 rather than routing through local disk.

Filtering Assets by Role & Media Type

Production pipelines rarely need all assets in a STAC Item — a Sentinel-2 item may contain 13 spectral bands plus multiple preview thumbnails and metadata XMLs. Filter assets before downloading to cut bandwidth costs:

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
        )
    }

For deeper bulk imagery acquisition patterns including COG validation, see Bulk Downloading Satellite Imagery with Python.

Null Geometry & bbox Fallback

Some providers return null in the geometry field for items that have a valid bbox property. The Pydantic validator catches this before items reach disk, but silently dropping them loses real data. Implement a bbox-to-polygon fallback:

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

Call coerce_geometry(raw_item) inside stream_validated_items before the STACItemValidator step. For systematic geometry repair across pipeline stages, review the approach in Geometry Repair with Shapely & GeoPandas.

Performance Optimization

For syncs covering collections with thousands of items, serialize metadata to SQLite or DuckDB rather than individual JSON files, and decouple asset downloads from item iteration using concurrent.futures.ThreadPoolExecutor:

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

Keep max_workers at or below 4 for public endpoints to stay within typical rate-limit envelopes. For enterprise deployments with authenticated access and higher quotas, benchmark with 8–16 workers.

Integration into ETL Pipelines

Schema Enforcement Hooks

Wrap run_stac_sync in a thin adapter that enforces output schema before passing records downstream. If the collection schema is managed in a registry (e.g. Avro schemas in a Confluent Schema Registry), load and validate the output metadata JSON before writing to the dead-letter queue:

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)

Orchestration Hooks (Airflow / Prefect / Dagster)

For Airflow, wrap run_stac_sync in a PythonOperator and pass query_params via XCom to allow upstream tasks to dynamically scope the sync window (e.g., set start_dt from the last successful DAG run timestamp). For Prefect, use a @flow with @task-decorated wrappers around initialize_stac_client and stream_validated_items to get automatic retry and observability. When ingesting alongside community vector data, coordinate the STAC sync cadence with the Overpass API extraction workflow so raster acquisitions and vector baselines share the same temporal window.

For multi-source pipelines that include legacy government data alongside STAC raster imagery, automating government portal downloads shows how to normalize non-STAC metadata into your ingestion queue before it reaches the schema validation stage.

Troubleshooting Reference

Reading a Collection’s Conformance and Queryables First

Two endpoints decide what your sync code is allowed to assume, and both are worth calling once at startup rather than discovering through failures.

/conformance lists the specification classes the server implements. The presence of item-search means POST search is available; filter means CQL2 predicates are supported; sort means you can ask for deterministic ordering rather than hoping for it. A catalogue missing item-search still works, but only through per-collection item listings, which changes the shape of the whole sync.

/queryables describes which item properties can appear in a filter, and with what types. This is where a sync breaks quietly: filtering on eo:cloud_cover against a catalogue that does not expose it returns everything rather than an error, and the pipeline downloads ten times the intended data. Asserting that the properties you filter on appear in the queryables response turns that into a startup failure with a clear message.

Cache both responses per catalogue for the lifetime of the run and log them with the run record. When a provider changes what it supports — which happens more often than the specification’s stability suggests — the log is what tells you the pipeline’s assumptions expired rather than its code broke.

Handling Catalogues That Move

STAC endpoints are more mobile than most APIs: collections get renamed, catalogues move to new hosts, and providers migrate between the cloud platforms that sponsor them. Two habits absorb most of that churn.

Resolve the collection by id through the root catalogue rather than hard-coding an item-search URL, so a host change is a single configuration edit. And when a search returns zero items for a query that previously worked, treat it as an error rather than an empty result — a renamed collection, an expired token and a genuinely empty window all look identical otherwise, and only the last one is acceptable to pass downstream silently.

Related

• STAC Catalog Pagination Best Practices in Python — generator-based pagination, limit/max_items controls, and next-link traversal
• Bulk Downloading Satellite Imagery with Python — STAC-driven imagery acquisition with COG validation and checksum verification
• Fetching OSM Data via Overpass API — complement STAC raster syncs with structured vector baselines from OpenStreetMap
• Automating Government Portal Downloads — bridge non-STAC legacy sources into a shared ingestion queue
• Geometry Repair with Shapely & GeoPandas — fix null or self-intersecting geometries before downstream spatial joins
• Filtering STAC Items by Cloud Cover and Datetime