Scheduling Sentinel-2 Downloads with Prefect

To download new Sentinel-2 scenes automatically, wrap a STAC search and a download step in a @flow, filter each search against a manifest of already-fetched scene IDs, and attach a CronSchedule through a deployment so it runs on an interval. The manifest is what keeps overlapping scheduled windows from re-downloading the same imagery:

from prefect import flow

@flow(name="s2-downloader")
def download_new_scenes(bbox, out_dir="./data", days=3):
    known = load_manifest(out_dir)                 # set of scene IDs seen before
    items = search_recent(bbox, days)              # STAC results, last N days
    new = [it for it in items if it["id"] not in known]
    paths = [fetch_asset.submit(it, out_dir).result() for it in new]
    append_manifest(out_dir, [it["id"] for it in new])
    return paths

This page expands that skeleton into a production-ready scheduled downloader. It sits under Orchestrating Pipelines with Prefect, which covers the broader flow/task model; here we focus narrowly on the scheduling-and-deduplication problem.

Why This Arises in Remote Sensing Workflows

Sentinel-2 acquisitions arrive on a fixed cadence — a 5-day revisit at the equator, tighter toward the poles — and providers such as Earth Search publish L2A Cloud-Optimized GeoTIFFs within about a day of sensing. A monitoring pipeline needs to notice those new scenes and pull them without a human running a script. The naive approach, re-downloading everything the search returns on every run, wastes egress and clobbers local files.

Two facts make deduplication necessary. First, to be sure you never miss a scene during a transient outage, each run must look back further than its own interval — a daily run typically searches two or three days. That overlap means consecutive runs see the same scenes. Second, STAC returns items in full each time regardless of what you already hold. Without a memory of what has been fetched, the overlap turns into repeated downloads. A manifest — a persisted set of scene IDs — gives the flow that memory.

For the full orchestration context, see the parent section on Cloud Execution & Orchestration; for the search mechanics reused below, the date-and-cloud filtering patterns come from Using pystac-client to Filter Sentinel-2 Imagery by Date.

How the scheduled run deduplicates

The diagram shows one scheduled run. The cron trigger fires; the flow loads the manifest, searches STAC, subtracts what it already has, downloads only the remainder, and writes the new IDs back to the manifest.

Scheduled Sentinel-2 downloader with manifest deduplication A cron schedule triggers the flow. The flow loads the manifest of known scene IDs, searches STAC for the last N days, computes the set difference of results minus known IDs, downloads only the new scenes, and appends their IDs back to the manifest for the next run. Cron 0 4 * * * Load manifest known IDs STAC search last N days results − known new scenes only Download retries=3 Append IDs to manifest next run reads the updated manifest

Environment & Setup

Package Minimum version Why required
prefect 2.14 @flow, @task, .submit, CronSchedule, deployments
pystac-client 0.7 Searches the STAC API for recent Sentinel-2 items
rasterio 1.3 Streams the remote COG asset to local disk 
pip install "prefect>=2.14,<3" "pystac-client>=0.7" "rasterio>=1.3"

Start a local Prefect server (prefect server start) or set PREFECT_API_URL for Prefect Cloud. No API key is needed for the public Earth Search STAC endpoint or the Sentinel-2 Open Data bucket.

Complete Working Example

The script below is a full, deployable downloader. The manifest is a JSON file of scene IDs; the search reuses a lookback window; downloads run in parallel with retries; and only genuinely new scenes touch the network.

import json
from datetime import date, timedelta
from pathlib import Path

import rasterio
from prefect import flow, task, get_run_logger
from prefect.tasks import task_input_hash
from pystac_client import Client

STAC_API = "https://earth-search.aws.element84.com/v1"
MANIFEST = "manifest.json"


@task(name="load-manifest")
def load_manifest(out_dir: str) -> set[str]:
    """Return the set of scene IDs already downloaded (empty on first run)."""
    path = Path(out_dir) / MANIFEST
    if not path.exists():
        return set()
    return set(json.loads(path.read_text()))


@task(name="search-recent", retries=2, retry_delay_seconds=15)
def search_recent(bbox: tuple[float, float, float, float], days: int) -> list[dict]:
    """Search STAC for Sentinel-2 L2A scenes over the last `days` days."""
    logger = get_run_logger()
    end = date.today()
    start = end - timedelta(days=days)
    client = Client.open(STAC_API)
    search = client.search(
        collections=["sentinel-2-l2a"],
        bbox=bbox,
        datetime=f"{start.isoformat()}/{end.isoformat()}",
        query={"eo:cloud_cover": {"lt": 30}},
    )
    # Keep only serializable fields so the result caches and passes cleanly
    items = [
        {"id": it.id, "href": it.assets["red"].href}
        for it in search.items()
    ]
    logger.info("Search window %s..%s returned %d items", start, end, len(items))
    return items


@task(
    name="fetch-asset",
    tags=["s2-download"],
    retries=3,
    retry_delay_seconds=[10, 30, 90],
    cache_key_fn=task_input_hash,
    cache_expiration=timedelta(days=14),
)
def fetch_asset(item: dict, out_dir: str) -> str:
    """Stream one COG to disk. Cached by input so a retry never double-fetches."""
    logger = get_run_logger()
    dest = Path(out_dir) / f"{item['id']}.tif"
    with rasterio.open(item["href"]) as src:     # /vsicurl/ range-reads the header
        profile = src.profile
        data = src.read()                        # pixel fetch happens here
    with rasterio.open(dest, "w", **profile) as dst:
        dst.write(data)
    logger.info("Fetched %s -> %s", item["id"], dest)
    return str(dest)


@task(name="append-manifest")
def append_manifest(out_dir: str, new_ids: list[str]) -> None:
    """Merge new scene IDs into the manifest so the next run skips them."""
    path = Path(out_dir) / MANIFEST
    existing = set(json.loads(path.read_text())) if path.exists() else set()
    existing.update(new_ids)
    path.write_text(json.dumps(sorted(existing)))


@flow(name="s2-scheduled-downloader")
def download_new_scenes(
    bbox: tuple[float, float, float, float],
    out_dir: str = "./data",
    days: int = 3,
) -> list[str]:
    logger = get_run_logger()
    Path(out_dir).mkdir(parents=True, exist_ok=True)

    known = load_manifest(out_dir)
    items = search_recent(bbox, days)
    # Set-difference on the scene ID is what deduplicates overlapping windows
    new = [it for it in items if it["id"] not in known]
    logger.info("%d/%d scenes are new", len(new), len(items))

    if not new:
        logger.info("Nothing new to download this run.")
        return []

    futures = [fetch_asset.submit(it, out_dir) for it in new]
    paths = [f.result() for f in futures]
    append_manifest(out_dir, [it["id"] for it in new])
    logger.info("Downloaded %d new scenes", len(paths))
    return paths


if __name__ == "__main__":
    download_new_scenes(bbox=(11.0, 46.0, 11.5, 46.5))

Run it once by hand to seed the manifest, then run it again immediately — the second run logs 0/N scenes are new and downloads nothing, confirming the deduplication works.

Variant Patterns

1. Attach a cron schedule via a deployment

To let the flow run itself, register a deployment with a CronSchedule. A daily 04:00 UTC run with a 2-day lookback comfortably catches new acquisitions while the manifest absorbs the overlap.

from prefect.deployments import Deployment
from prefect.server.schemas.schedules import CronSchedule
from downloader import download_new_scenes

Deployment.build_from_flow(
    flow=download_new_scenes,
    name="daily-s2",
    schedule=CronSchedule(cron="0 4 * * *", timezone="UTC"),
    parameters={"bbox": [11.0, 46.0, 11.5, 46.5], "days": 2},
    work_queue_name="downloads",
).apply()

Register with python deploy.py, then run a worker: prefect agent start -q downloads. Prefer IntervalSchedule(interval=timedelta(hours=12)) if you want twice-daily runs spaced from first execution rather than fixed clock times.

2. Store the manifest in shared object storage

A local JSON file breaks when different workers pick up runs. Point the load/append helpers at an S3 object so any worker sees the same state; the flow logic is unchanged.

import json
import boto3

_s3 = boto3.client("s3")

def load_manifest_s3(bucket: str, key: str) -> set[str]:
    try:
        obj = _s3.get_object(Bucket=bucket, Key=key)
        return set(json.loads(obj["Body"].read()))
    except _s3.exceptions.NoSuchKey:
        return set()

def append_manifest_s3(bucket: str, key: str, new_ids: list[str]) -> None:
    ids = load_manifest_s3(bucket, key)
    ids.update(new_ids)
    _s3.put_object(Bucket=bucket, Key=key, Body=json.dumps(sorted(ids)).encode())

3. Download multiple bands per scene

Real pipelines rarely want only the red band. Expand each item to a list of assets and map fetch_asset across them, keying the manifest by scene ID once all bands land.

BANDS = ["red", "nir", "green"]

@task(name="expand-assets")
def expand_assets(item, out_dir):
    return [{"id": f"{item['id']}_{b}", "href": item["assets"][b]} for b in BANDS]

Pair this with a tag-based concurrency limit (prefect concurrency-limit create s2-download 4) so multiplying bands by scenes does not overwhelm the archive.

Common Errors

Every run re-downloads the same scenes The manifest is not being persisted or not being read. Confirm append_manifest runs after a successful download and that load_manifest targets the same out_dir. If runs land on different workers, move the manifest to shared storage as in Variant 2.

KeyError: 'red' when building items The asset key differs by STAC provider — Earth Search uses red, others use B04. Print list(item.assets) once to confirm the key name before hard-coding it.

Scheduled deployment exists but nothing runs A schedule needs a worker polling its work_queue_name. Start prefect agent start -q downloads and verify with prefect deployment inspect s2-scheduled-downloader/daily-s2 that the next run times are populated.