cron vs Airflow vs Prefect for municipal feed schedules

# ingest.py — no framework imports. This is the part that must not care.
from dataclasses import dataclass


@dataclass(frozen=True)
class Result:
    source_id: str
    verdict: str          # "run" | "skip" | "restated"
    records: int


def ingest_source(source_id: str, store, fetch, logical_date) -> Result:
    """Idempotent for a given (source_id, logical_date). Takes its clock and its
    IO as arguments, so it can be exercised without a scheduler, a network, or a
    real database."""
    wm = store.load(source_id)
    probe = fetch.probe(source_id, as_of=logical_date)

    if wm and not probe.changed_since(wm):
        return Result(source_id, "skip", 0)

    payload = fetch.get(source_id, as_of=logical_date)
    if wm and payload.published_at and wm.published_at \
            and payload.published_at < wm.published_at:
        return Result(source_id, "restated", 0)

    n = store.upsert(source_id, payload, logical_date)   # idempotent upsert
    store.advance_watermark(source_id, payload)          # only after the write
    return Result(source_id, "run", n)

#!/usr/bin/env python3
"""run_ingest.py — invoked by: 15 2 * * *  /opt/venv/bin/python run_ingest.py"""
import logging
import sys
from datetime import date

from ingest import ingest_source
from infra import Store, Fetcher, lease

SOURCES = ["larimer", "boulder", "weld", "adams"]

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("ingest")


def main(logical_date: date) -> int:
    store, fetch = Store(), Fetcher()
    failures = []
    for source_id in SOURCES:
        # A lease PER SOURCE, not a global lock: a global one would recreate the
        # cross-jurisdiction coupling the whole design exists to avoid.
        with lease(f"ingest:{source_id}", ttl_seconds=3600) as held:
            if not held:
                log.warning("%s: already leased by another run, skipping", source_id)
                continue
            try:
                r = ingest_source(source_id, store, fetch, logical_date)
                log.info("%s: %s (%d records)", source_id, r.verdict, r.records)
            except Exception:
                log.exception("%s: failed", source_id)
                failures.append(source_id)
    if failures:
        log.error("run finished with %d failure(s): %s", len(failures), failures)
        return 1
    return 0


if __name__ == "__main__":
    given = sys.argv[1] if len(sys.argv) > 1 else None
    main(date.fromisoformat(given) if given else date.today())

from datetime import timedelta

import pendulum
from airflow.decorators import dag, task

from ingest import ingest_source
from infra import Store, Fetcher

SOURCES = ["larimer", "boulder", "weld", "adams"]


@dag(
    schedule="15 2 * * *",
    start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
    catchup=False,                       # municipal feeds are not replayable by date
    max_active_runs=1,                   # the lease, expressed declaratively
    default_args={
        "retries": 0,                    # retries live IN the task, not here
        "execution_timeout": timedelta(minutes=45),
    },
    tags=["municipal", "ingest"],
)
def municipal_ingest():
    @task(map_index_template="")
    def ingest(source_id: str, **context):
        # The logical date is the reason to use Airflow at all: this is what makes
        # `airflow tasks run ... 2026-08-04` a real backfill rather than a re-run.
        logical_date = context["logical_date"].date()
        r = ingest_source(source_id, Store(), Fetcher(), logical_date)
        return {"source": r.source_id, "verdict": r.verdict, "records": r.records}

    @task
    def publication_gate(results: list[dict]):
        held = [r["source"] for r in results if r["verdict"] == "restated"]
        if held:
            raise ValueError(f"held for review: {held}")
        return sum(r["records"] for r in results)

    publication_gate(ingest.expand(source_id=SOURCES))


municipal_ingest()

from datetime import date

from prefect import flow, task, get_run_logger
from prefect.futures import wait

from ingest import ingest_source
from infra import Store, Fetcher

SOURCES = ["larimer", "boulder", "weld", "adams"]


@task(retries=0, tags=["municipal-portal"], task_run_name="ingest-{source_id}")
def ingest(source_id: str, logical_date: date):
    log = get_run_logger()
    r = ingest_source(source_id, Store(), Fetcher(), logical_date)
    log.info("%s: %s (%d records)", r.source_id, r.verdict, r.records)
    return r


@flow(name="municipal-ingest", log_prints=True)
def municipal_ingest(logical_date: date | None = None):
    logical_date = logical_date or date.today()
    futures = [ingest.submit(s, logical_date) for s in SOURCES]
    wait(futures)

    results, failed = [], []
    for f in futures:
        try:
            results.append(f.result())
        except Exception as exc:                 # one source failing is not the run failing
            failed.append(str(exc))

    held = [r.source_id for r in results if r.verdict == "restated"]
    if held or failed:
        raise ValueError(f"held={held} failed={failed}")
    return sum(r.records for r in results)

def test_second_run_is_a_skip(fake_store, fake_fetch):
    first = ingest_source("larimer", fake_store, fake_fetch, date(2026, 8, 4))
    second = ingest_source("larimer", fake_store, fake_fetch, date(2026, 8, 4))
    assert first.verdict == "run" and first.records > 0
    assert second.verdict == "skip" and second.records == 0


def test_watermark_is_not_advanced_when_the_write_fails(fake_store, fake_fetch):
    fake_store.fail_next_upsert()
    before = fake_store.load("larimer")
    with pytest.raises(RuntimeError):
        ingest_source("larimer", fake_store, fake_fetch, date(2026, 8, 4))
    assert fake_store.load("larimer") == before      # the important assertion