Scheduling Recurring CSW Harvests with Python

TL;DR: Persist a watermark — the highest dc:modified timestamp you have successfully harvested — to a small state store. On each run, ask the Catalogue Service for the Web (CSW) only for records changed since that watermark using a PropertyIsGreaterThan filter, upsert them by identifier so repeats are harmless, and advance the watermark only when the whole run commits. Wrap the job in an APScheduler cron trigger with an exclusive lock and retry-with-backoff, and you have an incremental harvester that is safe to run every ten minutes forever.

The Core Challenge

A naive harvester re-downloads the entire catalogue on every schedule tick. Against a CSW endpoint holding tens of thousands of ISO 19139 records that is wasteful, slow, and hammers a shared server. The goal is incremental harvesting: fetch only what changed since last time. That reduces the problem to three tightly-coupled guarantees.

First, you need a durable memory of how far you got — a watermark that survives process restarts. Second, you need the CSW query to select strictly the records newer than that watermark, which means translating “since last harvest” into an OGC Filter Encoding constraint on the dc:modified property. Third — and this is where most home-grown harvesters break — the run must be crash-safe. A harvest can die halfway through paging results because the network dropped or the catalogue returned a 500. If the watermark has already moved past records you never stored, they are lost until they change again. The fix is a strict ordering: store records first, advance the watermark last, and make every write idempotent so a re-run of the same window changes nothing.

This page assumes you can already issue a basic query; if not, start with querying a CSW catalog with owslib in Python and the broader CSW Catalog Service Integration guide, then return here to make it recurring.

Production-Ready Code

The script below uses owslib for the CSW protocol and a single SQLite file for both the watermark and the harvested records — SQLite gives you atomic transactions and an ON CONFLICT upsert with zero infrastructure. Scheduling is handled by APScheduler; a fcntl file lock prevents overlapping runs on one host.

"""
csw_harvester.py — incremental, scheduled CSW harvesting.

pip install: owslib>=0.31, apscheduler>=3.10
Standard library: sqlite3, fcntl, time, logging, contextlib, datetime.
"""
from __future__ import annotations

import fcntl
import logging
import sqlite3
import time
from contextlib import contextmanager
from datetime import datetime, timezone
from typing import Iterator

from apscheduler.schedulers.background import BackgroundScheduler
from owslib.csw import CatalogueServiceWeb
from owslib.fes import PropertyIsGreaterThan, SortBy, SortProperty

CSW_URL = "https://example.org/geonetwork/srv/eng/csw"
DB_PATH = "harvest_state.db"
LOCK_PATH = "harvest.lock"
PAGE_SIZE = 50
EPOCH_FLOOR = "1970-01-01T00:00:00Z"   # first-run watermark
MODIFIED_PROP = "dc:modified"          # Dublin Core last-changed property

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


def init_db(db_path: str = DB_PATH) -> None:
    with sqlite3.connect(db_path) as conn:
        conn.executescript(
            """
            CREATE TABLE IF NOT EXISTS harvest_state (
                source     TEXT PRIMARY KEY,
                watermark  TEXT NOT NULL
            );
            CREATE TABLE IF NOT EXISTS records (
                identifier   TEXT PRIMARY KEY,
                title        TEXT,
                modified     TEXT,
                xml          TEXT,
                harvested_at TEXT NOT NULL
            );
            """
        )


def read_watermark(source: str, db_path: str = DB_PATH) -> str:
    with sqlite3.connect(db_path) as conn:
        row = conn.execute(
            "SELECT watermark FROM harvest_state WHERE source = ?", (source,)
        ).fetchone()
    return row[0] if row else EPOCH_FLOOR


def write_watermark(source: str, watermark: str, db_path: str = DB_PATH) -> None:
    with sqlite3.connect(db_path) as conn:
        conn.execute(
            "INSERT INTO harvest_state (source, watermark) VALUES (?, ?) "
            "ON CONFLICT(source) DO UPDATE SET watermark = excluded.watermark",
            (source, watermark),
        )


@contextmanager
def single_run_lock(lock_path: str = LOCK_PATH) -> Iterator[bool]:
    """Non-blocking exclusive lock; yields False if another run holds it."""
    fh = open(lock_path, "w")
    try:
        fcntl.flock(fh, fcntl.LOCK_EX | fcntl.LOCK_NB)
        yield True
    except BlockingIOError:
        yield False
    finally:
        fcntl.flock(fh, fcntl.LOCK_UN)
        fh.close()


def fetch_changed_records(csw: CatalogueServiceWeb, since: str) -> Iterator[object]:
    """Page GetRecords for everything modified after `since` (exclusive)."""
    constraint = PropertyIsGreaterThan(propertyname=MODIFIED_PROP, literal=since)
    sort_asc = SortBy([SortProperty(MODIFIED_PROP, "ASC")])
    start = 1
    while True:
        csw.getrecords2(
            constraints=[constraint],
            sortby=sort_asc,
            esn="full",
            startposition=start,
            maxrecords=PAGE_SIZE,
        )
        for identifier in csw.records:
            yield csw.records[identifier]
        nextrecord = csw.results["nextrecord"]
        if not nextrecord or nextrecord <= start:
            break
        start = nextrecord


def upsert_record(conn: sqlite3.Connection, rec: object) -> str | None:
    """Idempotent write keyed by identifier. Returns the record's dc:modified."""
    modified = getattr(rec, "modified", None) or ""
    conn.execute(
        "INSERT INTO records (identifier, title, modified, xml, harvested_at) "
        "VALUES (?, ?, ?, ?, ?) "
        "ON CONFLICT(identifier) DO UPDATE SET "
        "  title = excluded.title, modified = excluded.modified, "
        "  xml = excluded.xml, harvested_at = excluded.harvested_at",
        (
            rec.identifier,
            getattr(rec, "title", None),
            modified,
            getattr(rec, "xml", b"").decode("utf-8", "replace")
            if isinstance(getattr(rec, "xml", None), bytes)
            else getattr(rec, "xml", None),
            datetime.now(timezone.utc).isoformat(),
        ),
    )
    return modified or None


def harvest_once(source: str = CSW_URL) -> None:
    with single_run_lock() as acquired:
        if not acquired:
            log.warning("Another harvest is in progress; skipping this tick.")
            return

        watermark = read_watermark(source)
        log.info("Harvesting records modified after %s", watermark)
        csw = _connect_with_retry(source)

        high_water = watermark
        count = 0
        with sqlite3.connect(DB_PATH) as conn:
            for rec in fetch_changed_records(csw, watermark):
                modified = upsert_record(conn, rec)
                count += 1
                if modified and modified > high_water:
                    high_water = modified
            conn.commit()  # all records durable before the watermark moves

        # Advance the watermark ONLY after a clean, committed run.
        if high_water > watermark:
            write_watermark(source, high_water)
        log.info("Harvest complete: %d records, watermark now %s", count, high_water)


def _connect_with_retry(url: str, attempts: int = 5) -> CatalogueServiceWeb:
    """Exponential backoff on transient connection / server errors."""
    delay = 2.0
    for attempt in range(1, attempts + 1):
        try:
            return CatalogueServiceWeb(url, timeout=30)
        except Exception as exc:  # network, 5xx, malformed capabilities
            if attempt == attempts:
                raise
            log.warning("Connect attempt %d failed (%s); retrying in %.0fs",
                        attempt, exc, delay)
            time.sleep(delay)
            delay = min(delay * 2, 60)
    raise RuntimeError("unreachable")


if __name__ == "__main__":
    init_db()
    scheduler = BackgroundScheduler(timezone="UTC")
    scheduler.add_job(
        harvest_once,
        trigger="cron",
        minute="*/10",       # every 10 minutes
        id="csw_harvest",
        max_instances=1,     # never two at once inside this process
        coalesce=True,       # collapse missed runs into one
        misfire_grace_time=300,
    )
    scheduler.start()
    log.info("Scheduler started; press Ctrl+C to exit.")
    try:
        while True:
            time.sleep(1)
    except (KeyboardInterrupt, SystemExit):
        scheduler.shutdown()

Step-by-Step Walkthrough

The watermark round-trip. read_watermark returns the stored dc:modified string for this source, or EPOCH_FLOOR on the very first run so the initial harvest selects everything. Storing the watermark in the same SQLite file as the records is deliberate: it keeps the two facts — what you have and how far you got — colocated and easy to reason about. write_watermark uses an ON CONFLICT ... DO UPDATE upsert so it works identically for the first and subsequent runs.

Building the incremental constraint. fetch_changed_records wraps a single PropertyIsGreaterThan(propertyname="dc:modified", literal=since) filter. This is OGC Filter Encoding: it tells the CSW server to return only records whose dc:modified property is strictly greater than the watermark. Using greater-than (not greater-than-or-equal) excludes the boundary record you already hold, so you do not re-fetch it every cycle. Sorting ascending on the same property with SortBy means you page through changes oldest-first, which keeps high_water monotonically increasing as you go.

Paging safely. getrecords2 populates csw.records (a dict keyed by identifier) and csw.results. The loop reads csw.results["nextrecord"]; a CSW server returns 0 when there are no more records, so the guard if not nextrecord or nextrecord <= start terminates cleanly and also defends against a misbehaving server that fails to advance the cursor — without it, a stuck nextrecord would spin forever.

Idempotent upsert. upsert_record writes each record keyed on its identifier primary key with ON CONFLICT(identifier) DO UPDATE. Harvesting the same record twice simply overwrites the row with identical data — a no-op in effect. This is what makes re-processing safe: a crashed run that is retried, or an overlap window from a coarse timestamp, cannot create duplicates.

Ordering is the safety property. Inside harvest_once, every record is upserted and the transaction is committed before write_watermark runs. If the process dies mid-harvest, the watermark still points at the last fully-successful run, and the next tick re-selects the same window. Combined with idempotent upserts, that gives at-least-once delivery with no data loss and no duplicates — the sweet spot for a metadata harvester. This same discipline underpins the broader patterns in Automated Metadata Harvesting Workflows.

Overlap prevention and backoff. single_run_lock takes a non-blocking fcntl.flock; if a previous harvest is still running when the next cron tick fires, the new run yields False and exits immediately rather than piling on. _connect_with_retry retries transient connection and server errors with exponential backoff capped at 60 seconds, so a brief catalogue outage does not fail the whole schedule.

Scheduling. BackgroundScheduler runs the job in a daemon thread on a cron trigger (minute="*/10"). max_instances=1 stops APScheduler itself from launching a second overlapping job, coalesce=True collapses a backlog of missed triggers (for example after the machine was asleep) into a single run, and misfire_grace_time tolerates a late start.

Verification

Run the harvester and confirm the watermark advances and records land:

python csw_harvester.py &
sleep 15
sqlite3 harvest_state.db \
  "SELECT source, watermark FROM harvest_state;
   SELECT count(*) AS records FROM records;"

Expected output after the first tick against a live catalogue:

https://example.org/geonetwork/srv/eng/csw|2026-07-09T22:14:07Z
records
1284

Run it a second time within the same window and confirm idempotency: the record count stays flat unless the catalogue actually changed, and the log shows Harvest complete: 0 records when nothing was modified after the watermark.

Gotchas & Edge Cases

*/10 * * * * cd /opt/harvester && /opt/harvester/.venv/bin/python -c "import csw_harvester as h; h.init_db(); h.harvest_once()" >> /var/log/csw_harvest.log 2>&1

Back to Automated Metadata Harvesting Workflows

Related:

• CSW Catalog Service Integration — the GetRecords / GetRecordById operations and Filter Encoding foundations this harvester builds on
• Querying a CSW Catalog with owslib in Python — constructing constraints, paging, and reading csw.records before making it recurring
• Automated Metadata Harvesting Workflows — orchestration, deduplication, and idempotency patterns across harvesting sources