Scheduling e5

Python
1.5 hours

Prerequisites: Quality and provenance (e4)

How does a pipeline run every week without you, and how do you find out when a run broke? A job is one command and an exit code; the scheduler starts it in an environment that isn't your shell's, overlapping runs and failed runs are normal, and a failure nobody is told about is a silent one. A simulation of a weekly refresh, then the real job, written to a file and run eight times as a scheduler would run it, then cron, systemd timers and Windows Task Scheduler. Everything runs in the Python environment, on the lab tables (the setup page).

Concept

A pipeline that runs without you, and tells you when it breaks.

A job is one command with an exit code: 0 for success, anything else for failure. The scheduler sees only that code, so a job catches nothing silently: it writes the reason, puts the traceback in its error output and exits 1.
The scheduler's environment isn't yours. cron, a systemd timer and Task Scheduler start the job without your shell's profile, so the PG* settings you typed in a terminal aren't there. Give them to the job where the scheduler reads them (the crontab, the unit's EnvironmentFile=, the account's variables), keep the password in the password file, and have the job check them by name before it connects. Its log names a missing setting and never prints a value: logs are read by more people than passwords should be.
Logs: one line per step, with a timestamp, to a file or the journal.
Alerts: a failure nobody hears about is a silent one. systemd's OnFailure= starts another unit when a job fails; that unit alerts a person only if its command sends mail or a push. cron mails a job's output to MAILTO only when the crontab line doesn't redirect it to a file, so a line ending >> refresh.log 2>&1 mails nothing. Task Scheduler records the last result, and can restart a failed task.
A refresh you can restart (the snapshot rule of e3, run by a scheduler):
- the job key is one checksum over the input's checksum, TRANSFORM_VERSION, the config and SCHEMA_VERSION. Bump TRANSFORM_VERSION when the code's output changes, and SCHEMA_VERSION when the target's DDL does;
- a persistent target and a state table in their own schema, created once: the state row keeps the key the target was loaded with;
- one transaction replaces the target's rows and updates the state row, so a failed run leaves both as they were, and a run killed half-way commits nothing;
- a lock covers the comparison and the replacement, so two runs that overlap, a slow one and the next one or a manual re-run, can't both load. pg_try_advisory_xact_lock doesn't wait: a run that finds it held does nothing and exits 0, since the run holding it is doing the work. The lock is the transaction's, so it is freed when the job ends, however it ends.
Incremental loads move only what changed. A file with a published checksum tells you whether anything changed at all. Records that changed or were deleted upstream are the hard part: an append-only increment never sees them. For a sample this size, the honest increment is a whole reload, in one transaction, when the job key changes. A weekly GBIF refresh is a new download with a new DOI, so it is a new run of e1 to e4.

One refresh, one transaction: a held lock or an unchanged key skips, any error rolls back, and new rows commit with their key

Data: the lab data's manifest.json and datasets.csv; the lab's datasets table, for the shape of the job's copy.

Worked example

A simulation of a weekly refresh, in one process on TEMP tables, so it leaves the lab as it was and can be checked in one go. It compares the published checksum with the one it loaded last, reloads in one transaction when they differ, checks the row count, and exits 1 with the traceback on any failure.

Checked 2026-11-02 on PostgreSQL 16.15, PostGIS 3.4.2, GEOS 3.12.1, GDAL 3.8.4; Python 3.12.3 with requirements.txt sha256 d5bbe28024cc.

e5-job · Python

Runs after: py-connect

# A simulation: both "weeks" run in this one process, on TEMP tables that vanish when it ends. A scheduler runs a job as
# a new process each time, against tables that persist; e5-refresh writes that job, and e5-runs runs it.
import hashlib
import json
import logging
import os
import sys
import urllib.request

BASE = os.environ.get("LEARN_BASE", "https://critwire.com")
FILE = "datasets.csv"
# A scheduled job writes a timestamp on every line (add %(asctime)s); this lab leaves it out so its output can be checked.
logging.basicConfig(stream=sys.stdout, level=logging.INFO, format="%(levelname)s %(name)s: %(message)s")
log = logging.getLogger("refresh")


def read(url):
    with urllib.request.urlopen(url, timeout=60) as r:
        return r.read()


def refresh(conn):
    """One scheduled run: nothing to do if the published file is unchanged; else reload it in one transaction, checked."""
    entry = json.loads(read(f"{BASE}/learn/data/manifest.json"))["files"][FILE]
    last = conn.execute("SELECT sha256 FROM refresh_state WHERE file = %s", (FILE,)).fetchone()
    if last is not None and last[0] == entry["sha256"]:
        log.info("%s unchanged (sha256 %s): nothing to do", FILE, entry["sha256"][:12])
        return
    data = read(f"{BASE}/learn/data/{FILE}")
    if hashlib.sha256(data).hexdigest() != entry["sha256"]:
        raise RuntimeError(f"{FILE} doesn't match its manifest")
    with conn.transaction():  # a failed check below leaves the previous copy and state as they were
        conn.execute("TRUNCATE my_datasets")
        with conn.cursor().copy("COPY my_datasets FROM STDIN (FORMAT csv, HEADER true)") as copy:
            copy.write(data)
        rows = conn.execute("SELECT count(*) FROM my_datasets").fetchone()[0]
        if rows != entry["rows"]:
            raise RuntimeError(f"loaded {rows} rows of {FILE}; the manifest says {entry['rows']}")
        conn.execute("INSERT INTO refresh_state VALUES (%s, %s) ON CONFLICT (file) DO UPDATE SET sha256 = EXCLUDED.sha256",
                     (FILE, entry["sha256"]))
    log.info("%s loaded: %d rows, sha256 %s", FILE, rows, entry["sha256"][:12])


def main():
    """Exit 0 on success, 1 on any failure, with the traceback in the output. The scheduler sees only the exit code:
    systemd marks the unit failed; cron mails the output only if the crontab line doesn't redirect it to a file."""
    try:
        with connect() as conn:
            # TEMP tables: the simulation leaves learn_curriculum as it was; a real job's tables live in its own schema.
            conn.execute("CREATE TEMP TABLE my_datasets (LIKE datasets)")
            conn.execute("CREATE TEMP TABLE refresh_state (file text PRIMARY KEY, sha256 text NOT NULL)")
            for run in (1, 2):  # this week's run, then next week's with nothing new
                log.info("run %d", run)
                refresh(conn)
    except Exception:
        log.exception("refresh failed")
        return 1
    return 0


# A scheduled script ends with `raise SystemExit(main())`; here the lab goes on, so only a failure stops it.
exit_code = main()
if exit_code:
    raise SystemExit(exit_code)

-- result: e5-job · python · sha256 886a4128154e
INFO refresh: run 1
INFO refresh: datasets.csv loaded: 400 rows, sha256 3bd8c1726239
INFO refresh: run 2
INFO refresh: datasets.csv unchanged (sha256 3bd8c1726239): nothing to do

In e5.py and e5.ipynb, with the blocks it runs after.

The first run loads datasets.csv, 400 rows, and records its sha256 (3bd8c1726239…); the second finds the same checksum and does nothing. Both log one line per step, and the job exits 0. Point LEARN_BASE at a server that isn't running and it logs the traceback and exits 1. What the simulation can't show is what a scheduler does: start a new process each week, against tables that outlived the last one, maybe while the last one is still running.

Lab

The real job. This block writes refresh.py to the current folder and prints nothing; the job is the string. It reads the PG* settings, LEARN_BASE, REFRESH_SCHEMA (the schema holding its target and state tables, etl by default) and REFRESH_LICENSES (the licenses it keeps, CC0 1.0,CC BY 4.0 by default: its config). Each run is one transaction: the lock, the key compared with the state row's, the file fetched and checked against the manifest, staged with COPY and counted, then the target's rows replaced and the state row updated, together. A held lock or an equal key exits 0 having changed nothing; any error rolls the whole run back and exits 1.

e5-refresh · Python

from pathlib import Path

JOB = r'''"""refresh.py: makes <REFRESH_SCHEMA>.datasets hold the published datasets.csv, under this code and config.
Exit 0: loaded, or nothing to do. Exit 1: failed, and nothing changed. The reason goes to stdout, the traceback to stderr."""
import hashlib
import json
import logging
import os
import sys
import traceback
import urllib.request
from pathlib import Path

import psycopg
from psycopg import sql
from psycopg.types.json import Jsonb

FILE = "datasets.csv"
TRANSFORM_VERSION = 1  # bump when this code's output changes
SCHEMA_VERSION = 1  # bump when the target's DDL changes
SETTINGS = ("PGHOST", "PGPORT", "PGUSER", "PGDATABASE")

# A scheduled job writes a timestamp on every line (add %(asctime)s); this lab leaves it out so its output can be checked.
logging.basicConfig(stream=sys.stdout, level=logging.INFO, format="%(levelname)s %(name)s: %(message)s")
log = logging.getLogger("refresh")


def missing_settings():
    """The connection settings this environment lacks, by name: no value, the password least of all, reaches the log."""
    missing = [name for name in SETTINGS if not os.environ.get(name)]
    default = (Path(os.environ.get("APPDATA", "")) / "postgresql" / "pgpass.conf" if os.name == "nt"
               else Path.home() / ".pgpass")
    if not os.environ.get("PGPASSWORD") and not Path(os.environ.get("PGPASSFILE", default)).is_file():
        missing.append("PGPASSWORD or a password file")
    return missing


def read(url):
    with urllib.request.urlopen(url, timeout=60) as r:
        return r.read()


def job_key(parts):
    """One sha256 over the four parts, as canonical JSON (sorted keys, no spaces), so equal parts give one key (e3)."""
    return hashlib.sha256(json.dumps(parts, sort_keys=True, separators=(",", ":")).encode()).hexdigest()


def refresh(conn, base, schema, licenses):
    """One run in one transaction: the lock, the comparison, the replacement and the new state commit together, or
    nothing does. Returns the log line."""
    entry = json.loads(read(f"{base}/learn/data/manifest.json"))["files"][FILE]
    parts = {"input": entry["sha256"], "transform": TRANSFORM_VERSION,
             "config": {"file": FILE, "licenses": licenses}, "schema": SCHEMA_VERSION}
    key = job_key(parts)
    target, state = sql.Identifier(schema, "datasets"), sql.Identifier(schema, "refresh_state")
    with conn.transaction():
        # The transaction's own lock: COMMIT or ROLLBACK frees it, however the job ends. Held means another run is at work.
        if not conn.execute("SELECT pg_try_advisory_xact_lock(hashtext(%s))", (f"refresh:{schema}",)).fetchone()[0]:
            return "another refresh holds the lock: skipped"
        last = conn.execute(sql.SQL("SELECT job_key, parts FROM {} WHERE job = %s").format(state), (FILE,)).fetchone()
        if last is not None and last[0] == key:
            return "nothing changed: skipped"
        data = read(f"{base}/learn/data/{FILE}")
        if hashlib.sha256(data).hexdigest() != entry["sha256"]:
            raise RuntimeError(f"{FILE} doesn't match its manifest's sha256")
        conn.execute(sql.SQL("CREATE TEMP TABLE stage (LIKE {}) ON COMMIT DROP").format(target))
        with conn.cursor().copy("COPY pg_temp.stage FROM STDIN (FORMAT csv, HEADER true)") as copy:
            copy.write(data)
        staged = conn.execute("SELECT count(*) FROM pg_temp.stage").fetchone()[0]
        if staged != entry["rows"]:
            raise RuntimeError(f"staged {staged} rows; the manifest says {entry['rows']}")
        conn.execute(sql.SQL("TRUNCATE {}").format(target))
        loaded = conn.execute(sql.SQL("INSERT INTO {} SELECT * FROM pg_temp.stage WHERE license = ANY(%s)").format(target),
                              (licenses,)).rowcount
        conn.execute(sql.SQL("INSERT INTO {} VALUES (%s, %s, %s) ON CONFLICT (job) DO UPDATE "
                             "SET job_key = EXCLUDED.job_key, parts = EXCLUDED.parts").format(state),
                     (FILE, key, Jsonb(parts)))
    if last is None:
        return f"first load: loaded {loaded} rows"
    changed = " and ".join(p for p in sorted(parts) if parts[p] != last[1].get(p))
    return f"{changed} changed: loaded {loaded} rows"


def main():
    missing = missing_settings()
    if missing:
        log.error("failed: not set in this environment: %s", ", ".join(missing))
        return 1
    base = os.environ.get("LEARN_BASE", "https://critwire.com")
    schema = os.environ.get("REFRESH_SCHEMA", "etl")
    licenses = sorted(name.strip() for name in os.environ.get("REFRESH_LICENSES", "CC0 1.0,CC BY 4.0").split(","))
    try:
        with psycopg.connect(autocommit=True) as conn:  # autocommit: conn.transaction() is then the one transaction
            log.info(refresh(conn, base, schema, licenses))
    except Exception as e:
        log.error("failed: %s", e)
        traceback.print_exc()
        return 1
    return 0


if __name__ == "__main__":
    raise SystemExit(main())
'''

Path("refresh.py").write_text(JOB, encoding="utf-8")

-- result: e5-refresh · python · sha256 e3b0c44298fc

In e5.py and e5.ipynb, with the blocks it runs after.

Run it as a scheduler does: a new process each time, here eight times. The block creates a schema learn_e5 with the target and state tables, as you would create etl's once, and drops it at the end. A is the published datasets.csv; B is a local copy with one title corrected; C is B with one made-up dataset added, under a manifest that still says 400 rows, as if the publisher's count went wrong. Each run gets this block's environment; run 8 without PGDATABASE. Run 6 holds the job's lock from this block's own session, as a run still going would; run 7 starts two runs at once.

e5-runs · Python

Runs after: py-connect, e5-refresh

import csv
import hashlib
import io
import json
import os
import subprocess
import sys
import tempfile
import urllib.request
from pathlib import Path

BASE = os.environ.get("LEARN_BASE", "https://critwire.com")
SCHEMA = "learn_e5"
LICENSES = {"CC0 1.0", "CC BY 4.0"}  # refresh.py's default config


def rows_of(data):
    return [tuple(r) for r in csv.reader(io.StringIO(data.decode("utf-8"), newline=""))][1:]


def csv_file(header, rows):
    out = io.StringIO()
    csv.writer(out, lineterminator="\n").writerows([header, *rows])
    return out.getvalue().encode("utf-8")


def publish(folder, data, rows):
    """A copy of /learn/data/ on this machine holding `data`, its manifest saying `rows` rows; returns its LEARN_BASE."""
    d = Path(folder) / "learn" / "data"
    d.mkdir(parents=True)
    (d / "datasets.csv").write_bytes(data)
    entry = {"sha256": hashlib.sha256(data).hexdigest(), "rows": rows}
    (d / "manifest.json").write_text(json.dumps({"files": {"datasets.csv": entry}}), encoding="utf-8")
    return Path(folder).as_uri()


def start(base=None, licenses=None, without=None):
    """refresh.py as a new process, with this block's environment and these changes."""
    env = {k: v for k, v in os.environ.items() if k != without}
    env["REFRESH_SCHEMA"] = SCHEMA
    if base:
        env["LEARN_BASE"] = base
    if licenses:
        env["REFRESH_LICENSES"] = licenses
    return subprocess.Popen([sys.executable, "-I", "refresh.py"], env=env, text=True,
                            stdout=subprocess.PIPE, stderr=subprocess.PIPE)


def finish(proc):
    out, err = proc.communicate(timeout=120)
    return proc.returncode, out.strip(), " · traceback on stderr" if "Traceback" in err else ""


def show(n, label, proc):
    code, out, err = finish(proc)
    print(f"run {n} · {label} · exit {code} · {out}{err}")


def holds(conn, inputs):
    """Which input the target's rows and the state's key come from: read, not assumed."""
    rows = set(conn.execute(f"SELECT CAST(dataset_key AS text), title, publisher, coalesce(doi, ''), license "
                            f"FROM {SCHEMA}.datasets").fetchall())
    target = next((n for n, data in inputs.items() if rows == {r for r in rows_of(data) if r[4] in LICENSES}), "no input")
    sha = conn.execute(f"SELECT parts->>'input' FROM {SCHEMA}.refresh_state WHERE job = 'datasets.csv'").fetchone()[0]
    state = next((n for n, data in inputs.items() if hashlib.sha256(data).hexdigest() == sha), "no input")
    print(f"  the target holds {target}; the state's input is {state}")


with connect() as conn, tempfile.TemporaryDirectory() as tmp:
    conn.autocommit = True  # each statement commits, so the job's own sessions see the tables
    conn.execute(f"CREATE SCHEMA {SCHEMA}")
    try:
        conn.execute(f"CREATE TABLE {SCHEMA}.datasets (dataset_key uuid PRIMARY KEY, title text NOT NULL, "
                     "publisher text NOT NULL, doi text, license text NOT NULL)")
        conn.execute(f"CREATE TABLE {SCHEMA}.refresh_state (job text PRIMARY KEY, job_key text NOT NULL, parts jsonb NOT NULL)")
        with urllib.request.urlopen(f"{BASE}/learn/data/datasets.csv", timeout=60) as r:
            a = r.read()
        header, *rows = csv.reader(io.StringIO(a.decode("utf-8"), newline=""))
        b_rows = [[*rows[0][:1], rows[0][1] + " (corrected)", *rows[0][2:]], *rows[1:]]
        made_up = ["00000000-0000-4000-8000-000000000001", "A made-up dataset", "Nobody", "", "CC0 1.0"]
        inputs = {"A": a, "B": csv_file(header, b_rows), "C": csv_file(header, [*b_rows, made_up])}
        b_base = publish(Path(tmp) / "b", inputs["B"], len(b_rows))
        c_base = publish(Path(tmp) / "c", inputs["C"], len(b_rows))  # one row more than its manifest says

        show(1, "A", start())
        show(2, "A again", start())
        show(3, "A, REFRESH_LICENSES=CC0 1.0", start(licenses="CC0 1.0"))
        show(4, "B, one title corrected", start(base=b_base))
        show(5, "C, one row added, its manifest still says 400 rows", start(base=c_base))
        holds(conn, inputs)
        lock = f"refresh:{SCHEMA}"
        conn.execute("SELECT pg_advisory_lock(hashtext(%s))", (lock,))  # as a run that is still going holds it
        try:
            show(6, "A, the lock held by another session", start())
        finally:
            conn.execute("SELECT pg_advisory_unlock(hashtext(%s))", (lock,))
        holds(conn, inputs)
        both = [finish(p) for p in [start(), start()]]
        # Which one wins, and whether the other meets the lock or the new state, varies from run to run.
        outcomes = sorted("loaded" if ": loaded " in out else "not loaded" for _, out, _ in both)
        print(f"run 7 · A, two runs started together · exit {' and '.join(str(c) for c, _, _ in both)} · {', '.join(outcomes)}")
        holds(conn, inputs)
        show(8, "A, without PGDATABASE", start(without="PGDATABASE"))
    finally:
        conn.execute(f"DROP SCHEMA {SCHEMA} CASCADE")

-- result: e5-runs · python · sha256 bda9eaea67a6
run 1 · A · exit 0 · INFO refresh: first load: loaded 400 rows
run 2 · A again · exit 0 · INFO refresh: nothing changed: skipped
run 3 · A, REFRESH_LICENSES=CC0 1.0 · exit 0 · INFO refresh: config changed: loaded 124 rows
run 4 · B, one title corrected · exit 0 · INFO refresh: config and input changed: loaded 400 rows
run 5 · C, one row added, its manifest still says 400 rows · exit 1 · ERROR refresh: failed: staged 401 rows; the manifest says 400 · traceback on stderr
  the target holds B; the state's input is B
run 6 · A, the lock held by another session · exit 0 · INFO refresh: another refresh holds the lock: skipped
  the target holds B; the state's input is B
run 7 · A, two runs started together · exit 0 and 0 · loaded, not loaded
  the target holds A; the state's input is A
run 8 · A, without PGDATABASE · exit 1 · ERROR refresh: failed: not set in this environment: PGDATABASE

In e5.py and e5.ipynb, with the blocks it runs after.

Each line is one process: its exit code and the one line it logged. Run 2 skips because the state row outlived run 1's process. Run 3 changes only the config, so it reloads, keeping the 124 CC0 1.0 datasets; run 4 changes the input and puts the default config back. Run 5 stages 401 rows against a manifest of 400 and fails: it exits 1, and the target and the state still hold B, because the replacement and the state row would have committed together. Run 6 finds the lock held and changes nothing. In run 7 one process loads A; the other either finds the lock held or, starting a moment later, finds the new state, and both are right to exit 0. After A, B and A again the target holds A. Run 8 names the missing setting and prints no value.

Scheduling it

Put refresh.py in a folder of its own, ~/etl here, with a .venv built from the lab's requirements.txt (the setup page), and create the schema etl with its two tables once, as e5-runs creates learn_e5's.

On Linux with cron, every Monday at 04:17. cron reads VAR=value lines at the top of the crontab; the password stays in ~/.pgpass:

crontab -e
# then add these lines:
PGHOST=localhost
PGPORT=5432
PGUSER=postgres
PGDATABASE=learn
17 4 * * 1 cd "$HOME/etl" && .venv/bin/python -I refresh.py >> refresh.log 2>&1

With >> refresh.log 2>&1, everything the job prints goes to refresh.log, so cron has nothing to mail and a failure is only a line in that file. Leave the redirect off and set MAILTO=you@example.org to have cron mail each run's output, if the machine can send mail.

Or a systemd user timer, which catches up on a run missed while the machine was off (Persistent=true) and logs to the journal:

mkdir -p ~/.config/systemd/user
printf 'PGHOST=localhost\nPGPORT=5432\nPGUSER=postgres\nPGDATABASE=learn\n' > ~/etl/refresh.env
cat > ~/.config/systemd/user/refresh.service <<'EOF'
[Unit]
Description=Weekly lab data refresh
OnFailure=refresh-failed.service

[Service]
Type=oneshot
WorkingDirectory=%h/etl
EnvironmentFile=%h/etl/refresh.env
ExecStart=%h/etl/.venv/bin/python -I refresh.py
EOF
cat > ~/.config/systemd/user/refresh-failed.service <<'EOF'
[Unit]
Description=Record that the refresh failed

[Service]
Type=oneshot
ExecStart=/usr/bin/logger -p user.err "refresh failed: journalctl --user -u refresh.service"
EOF
cat > ~/.config/systemd/user/refresh.timer <<'EOF'
[Unit]
Description=Run the refresh every Monday

[Timer]
OnCalendar=Mon *-*-* 04:17:00
Persistent=true

[Install]
WantedBy=timers.target
EOF
systemctl --user daemon-reload
systemctl --user enable --now refresh.timer
loginctl enable-linger "$USER"     # run it even when you aren't logged in
journalctl --user -u refresh.service   # the log

refresh-failed.service only writes a line to the system log, which alerts nobody: someone has to read it. To be told, replace its ExecStart with a command that sends mail or a push to your phone.

On Windows, with Task Scheduler, from PowerShell. The task runs as you, so it sees your user environment variables (setx PGHOST localhost, and so on, once) and your password file, %APPDATA%\postgresql\pgpass.conf:

$action = New-ScheduledTaskAction -Execute "$HOME\etl\.venv\Scripts\python.exe" -Argument "-I refresh.py" -WorkingDirectory "$HOME\etl"
$trigger = New-ScheduledTaskTrigger -Weekly -DaysOfWeek Monday -At 4:17am
$settings = New-ScheduledTaskSettingsSet -StartWhenAvailable -RestartCount 2 -RestartInterval (New-TimeSpan -Minutes 30)
Register-ScheduledTask -TaskName "Lab data refresh" -Action $action -Trigger $trigger -Settings $settings
Get-ScheduledTaskInfo -TaskName "Lab data refresh"   # LastTaskResult 0 means success

Task Scheduler keeps the last result, and the task's history, and tells nobody: check LastTaskResult, or add a second task triggered by the failure event that sends a message.

Result

A refresh job that a scheduler can run as often as it likes. Each run is one process and one transaction: an unchanged job key skips, a changed input or config reloads, a failed check leaves the target and state as they were, and of two runs at once exactly one loads. After A, B and A again, the target holds A. The job checks its connection settings by name and logs no value, and the scheduling recipes say plainly where a failure ends up: a log file, the journal, or a task's last result, until you add the mail or push that tells a person.

Exercise

A job key covers only what you put in it. Change refresh.py's transform so each DOI is stored as a link that resolves, https://doi.org/ followed by the DOI. Write the change as a copy of refresh.py, and a second copy that also sets TRANSFORM_VERSION = 2. Into a schema learn_e5ex of your own, created as e5-runs creates learn_e5, run the old job once, then the new code without the bump, then with it, twice, each as a new process. After each run, count the target's DOIs that are links.

Show answer

e5-ex · Python

Runs after: py-connect, e5-refresh

import os
import subprocess
import sys
import tempfile
from pathlib import Path

SCHEMA = "learn_e5ex"
JOB = Path("refresh.py").read_text(encoding="utf-8")


def edit(text, old, new):
    """`text` with its one `old` replaced by `new`: an edit that finds nothing to change fails, it doesn't pass silently."""
    if text.count(old) != 1:
        raise ValueError(f"expected one {old!r} in refresh.py")
    return text.replace(old, new)


# The new transform: each DOI stored as a link, https://doi.org/<doi>. Same input, same config, a different output.
LINKS = edit(JOB, "SELECT * FROM pg_temp.stage WHERE",
             "SELECT dataset_key, title, publisher, 'https://doi.org/' || doi, license FROM pg_temp.stage WHERE")


def run(conn, n, label, script):
    """`script` as a new process, with this block's environment, then the target's DOIs that are links."""
    done = subprocess.run([sys.executable, "-I", str(script)], env={**os.environ, "REFRESH_SCHEMA": SCHEMA},
                          text=True, capture_output=True, timeout=120)
    links = conn.execute(f"SELECT count(*) FROM {SCHEMA}.datasets WHERE starts_with(doi, 'https://doi.org/')").fetchone()[0]
    print(f"run {n} · {label} · exit {done.returncode} · {done.stdout.strip()} · {links} DOIs as links")


with connect() as conn, tempfile.TemporaryDirectory() as tmp:
    conn.autocommit = True  # each statement commits, so the job's own sessions see the tables
    unbumped, bumped = Path(tmp) / "refresh_links.py", Path(tmp) / "refresh_v2.py"
    unbumped.write_text(LINKS, encoding="utf-8")
    bumped.write_text(edit(LINKS, "TRANSFORM_VERSION = 1", "TRANSFORM_VERSION = 2"), encoding="utf-8")
    conn.execute(f"CREATE SCHEMA {SCHEMA}")
    try:
        conn.execute(f"CREATE TABLE {SCHEMA}.datasets (dataset_key uuid PRIMARY KEY, title text NOT NULL, "
                     "publisher text NOT NULL, doi text, license text NOT NULL)")
        conn.execute(f"CREATE TABLE {SCHEMA}.refresh_state (job text PRIMARY KEY, job_key text NOT NULL, parts jsonb NOT NULL)")
        run(conn, 1, "the old job", "refresh.py")
        run(conn, 2, "links, TRANSFORM_VERSION still 1", unbumped)
        run(conn, 3, "links, TRANSFORM_VERSION = 2", bumped)
        run(conn, 4, "links, TRANSFORM_VERSION = 2 again", bumped)
    finally:
        conn.execute(f"DROP SCHEMA {SCHEMA} CASCADE")

-- result: e5-ex · python · sha256 b706ba3740bb
run 1 · the old job · exit 0 · INFO refresh: first load: loaded 400 rows · 0 DOIs as links
run 2 · links, TRANSFORM_VERSION still 1 · exit 0 · INFO refresh: nothing changed: skipped · 0 DOIs as links
run 3 · links, TRANSFORM_VERSION = 2 · exit 0 · INFO refresh: transform changed: loaded 400 rows · 400 DOIs as links
run 4 · links, TRANSFORM_VERSION = 2 again · exit 0 · INFO refresh: nothing changed: skipped · 400 DOIs as links

In e5.py and e5.ipynb, with the blocks it runs after.

Check

Run 2 is the failure the version number exists for. The code changed, but the input, the config and both versions didn't, so the key is the same: the job skips, exits 0, and the target keeps 400 bare DOIs under a state row that says it is up to date. Nothing tells a job that its own code changed except what you put in its key. With the bump, run 3 reloads and logs transform changed, and run 4 skips, every DOI now a link. Every one of the 400 datasets has a DOI, so the count goes from 0 to 400 in one run.

Dialects

PostgreSQL: the pg_cron extension schedules SQL inside the database, SELECT cron.schedule('refresh', '17 4 * * 1', 'CALL refresh()'), in cron's syntax, and keeps each run's outcome in cron.job_run_details. It runs SQL only, so fetching a file stays with an outside job like refresh.py. pg_try_advisory_xact_lock takes a bigint or two integers: hashtext turns a name into one, and a fixed number written into the job does as well.
DuckDB: no server and no scheduler: the job is a script that cron, a systemd timer or Task Scheduler starts, as here. A database file opens for writing in one process at a time, and a second process that tries gets a lock error, so the file's lock does the advisory lock's work, except that the second run gets an error where refresh.py skips: catch that error if runs are expected to overlap. BEGIN TRANSACTION; … COMMIT; holds the replacement and the state row together.
SQL Server: SQL Server Agent runs jobs on schedules, records each run's outcome in its history, and notifies an operator by Database Mail on failure; Express edition has no Agent, so it uses Task Scheduler. sp_getapplock with @LockMode = 'Exclusive', @LockOwner = 'Transaction' and @LockTimeout = 0 is pg_try_advisory_xact_lock: it returns a negative number instead of waiting when another session holds the lock.
Python: raise SystemExit(main()) turns the job's result into its exit code; an uncaught exception also exits 1, with the traceback but without the job's own log line. A scheduler inside Python (APScheduler, or the schedule package) runs jobs while its own process lives, so a crash or a restart stops every schedule; orchestrators such as Airflow, Dagster and Prefect add retries, history and alerts, and still run each task as here, so it must be safe to run again.
ArcGIS Pro: a geoprocessing tool or a model can be scheduled from the Geoprocessing pane (the Run button's menu → Schedule), which creates a Windows Task Scheduler task. A Python script runs on a schedule with Pro's own Python, "C:\Program Files\ArcGIS\Pro\bin\Python\Scripts\propy.bat" refresh.py, in a task like the one above; arcpy then needs a Pro license the task's account can use.