My GIS Pipeline Fails Silently: Fixing Swallowed Errors and Wrong Exit Codes
On this page
Problem statement
The scheduler reports a green run every night. Three weeks later someone notices the output layer has been empty since the 14th. This is how that happens — a real script, run against a real folder whose input has gone missing:
%%writefile nightly_bad.py
import logging, geopandas as gpd, pandas as pd
logging.basicConfig(filename="logs/nightly.log", level=logging.INFO, format="[%(asctime)s] %(message)s")
logging.info("starting nightly build")
frames = []
for name in ["roads", "landuse", "places"]:
try:
frames.append(gpd.read_file(f"input_today/{name}.gpkg")) # the delivery did not arrive
except Exception:
pass
out = gpd.GeoDataFrame(pd.concat(frames) if frames else pd.DataFrame({"geometry": []}), crs=27700)
out.to_file("output/rutland.gpkg")
logging.info("done")import subprocess
run = subprocess.run([sys.executable, "nightly_bad.py"])
print("exit code:", run.returncode)
print(Path("logs/nightly.log").read_text().splitlines()[-2:])
print("features written:", len(gpd.read_file("output/rutland.gpkg")))exit code: 0
['[2026-10-11 00:38:36,756] Created 0 records', '[2026-10-11 00:38:36,759] done']
features written: 0
Nothing crashed, so nothing alerted. "done" was printed by a script that swallowed three exceptions, wrote a zero-row file, and exited 0. The pipeline did not fail — it succeeded at producing nothing, which is much harder to notice.
How a failure gets swallowed on its way to the scheduler:
except Exception: pass, or anexceptblock that logs atDEBUGlevel- an exit code of 0 returned regardless of what happened
- a shell wrapper where only the last command's status is checked (
cmd1; cmd2) - a pipeline
a | bwhose exit status reflectsbalone - output written before validation, so a truncated result looks like a result
- an empty input folder treated as "nothing to do" rather than as a problem
- a
tryaround a whole stage, so a failure skips work without stopping the run
The unifying theme: at every boundary — function, process, shell, scheduler — there is a signal that says "this went wrong", and each of those signals can be dropped independently. Every example on this page runs against Geofabrik's OpenStreetMap layers for Rutland, set up in The data used on this page.
Quick answer
To make failures visible:
- never write
except: pass; log withlogger.exception()and re-raise or record - return a real exit code — non-zero on any failure — from
main() - use
set -euo pipefailin every shell wrapper - assert on the result, not just the absence of exceptions: row counts, CRS, bbox, file size
- emit a machine-readable run summary and alert on it, including "did the job run at all"
log = logging.getLogger("pipeline")
def run_pipeline(input_dir="input"):
frames = [gpd.read_file(p) for p in sorted(Path(input_dir).glob("*.gpkg"))]
out = gpd.GeoDataFrame(pd.concat(frames, ignore_index=True), crs=frames[0].crs) if frames else gpd.GeoDataFrame()
out.to_file("output/rutland.gpkg")
return {"features": len(out), "output": "output/rutland.gpkg"}
def main(input_dir="input") -> int:
try:
result = run_pipeline(input_dir)
except Exception:
log.exception("pipeline failed") # full traceback at ERROR level
return 1
if result["features"] == 0:
log.error("pipeline produced 0 features — treating as failure")
return 1
log.info("wrote %s features to %s", result["features"], result["output"])
return 0
print("all inputs present:", main("input"))
print("empty input folder:", main("input_today"))
print("a corrupt input: ", main("input_with_bad_file"))all inputs present: 0
empty input folder: 1
a corrupt input: 1
raise SystemExit(main()) at the bottom of the script is the whole exit-code contract in one line: whatever main returns becomes the process status the scheduler sees. The empty folder is caught by the zero-features check, the corrupt file by the exception handler — the two different ways the bad script above stayed quiet.
Where the signal gets dropped
Stop swallowing exceptions, return honest exit codes
Stop swallowing exceptions
There are three legitimate reasons to catch an exception: to add context and re-raise, to record a per-item failure in a batch, or to handle a genuinely expected condition. Everything else hides bugs. The three patterns, on the truncated GeoPackage:
path = Path("input_with_bad_file/buildings.gpkg")
failures = []
# 1. hides everything, including typos in your own code
try:
gdf = gpd.read_file(path)
except Exception:
pass
print("1. pass: nothing printed, nothing recorded")
# 2. records the failure and keeps the batch alive — the pattern for per-file loops
try:
gdf = gpd.read_file(path)
except Exception as exc:
failures.append((path.name, f"{type(exc).__name__}: {exc}"))
print("2. recorded:", failures[0][1][:90])
# 3. adds context and re-raises — the pattern for a single-step pipeline
try:
try:
gdf = gpd.read_file(path)
except Exception as exc:
raise RuntimeError(f"could not read {path}") from exc
except RuntimeError as err:
print("3. raised:", err, "| cause:", type(err.__cause__).__name__)1. pass: nothing printed, nothing recorded
2. recorded: DataSourceError: sqlite3_prepare_v2(SELECT COUNT(*) FROM sqlite_master WHERE name IN ('gpk
3. raised: could not read input_with_bad_file/buildings.gpkg | cause: DataSourceError
raise ... from exc keeps the original exception attached as __cause__, so the traceback shows both the context and the cause.
Log exceptions at a level someone reads
A traceback logged at DEBUG in a job configured at INFO is a traceback nobody will ever see:
logging.basicConfig(filename="logs/levels.log", level=logging.INFO, force=True,
format="%(asctime)s %(levelname)-8s %(name)s %(message)s")
job = logging.getLogger("pipeline")
try:
gpd.read_file(path)
except Exception:
job.debug("read failed", exc_info=True) # below the configured level: dropped
job.exception("read failed") # ERROR + traceback
text = Path("logs/levels.log").read_text()
print(text.count("DEBUG"), "DEBUG lines,", text.count("ERROR"), "ERROR line,", text.count("Traceback"), "traceback in the log")0 DEBUG lines, 1 ERROR line, 1 traceback in the log
logger.exception() is logger.error() plus the traceback, and it is meant for use inside an except block — which is exactly where you want it.
Return an honest exit code
Schedulers, CI and make all decide success from the process status. A script that always exits 0 cannot be monitored:
%%writefile nightly.py
import sys, geopandas as gpd
from pathlib import Path
def main() -> int:
folder = Path(sys.argv[1])
if not folder.is_dir():
print(f"config error: {folder} is not a folder", file=sys.stderr)
return 2
files = sorted(folder.glob("*.gpkg"))
if not files:
print("no input files", file=sys.stderr)
return 3
try:
total = sum(len(gpd.read_file(p)) for p in files)
except Exception as exc:
print(f"failed: {type(exc).__name__}: {exc}", file=sys.stderr)
return 1
print(f"ok: {total} features from {len(files)} files")
return 0
if __name__ == "__main__":
raise SystemExit(main())for folder in ["input", "input_with_bad_file", "input_today", "no_such_folder"]:
run = subprocess.run([sys.executable, "nightly.py", folder], capture_output=True, text=True)
print(f"{folder:20} exit {run.returncode} {(run.stdout or run.stderr).strip()[:70]}")input exit 0 ok: 9997 features from 3 files
input_with_bad_file exit 1 failed: DataSourceError: sqlite3_prepare_v2(SELECT COUNT(*) FROM sqlit
input_today exit 3 no input files
no_such_folder exit 2 config error: no_such_folder is not a folder
Distinct codes let a wrapper react differently: retry a transient failure, page someone for a configuration error, ignore "no input" on a public holiday.
Make the shell wrapper strict
By default, a shell script continues after a failing command and reports the status of the last one. Four wrappers around the same failing step:
wrappers = {
"no options": "\"$PY\" nightly.py no_such_folder\necho transform ran on stale data",
"set -e": "set -e\n\"$PY\" nightly.py no_such_folder\necho transform ran on stale data",
"| tee, no pipefail": "set -e\n\"$PY\" nightly.py no_such_folder 2>&1 | tee logs/tee.log",
"| tee, pipefail": "set -eo pipefail\n\"$PY\" nightly.py no_such_folder 2>&1 | tee logs/tee.log",
}
for label, script in wrappers.items():
run = subprocess.run(["bash", "-c", script], capture_output=True, text=True, env=dict(os.environ, PY=sys.executable))
ran_on = "transform ran" in run.stdout
print(f"{label:20} exit {run.returncode}{' (and the next step ran)' if ran_on else ''}")no options exit 0 (and the next step ran)
set -e exit 2
| tee, no pipefail exit 0
| tee, pipefail exit 2
Without -e, the failing step is followed by the next one and the wrapper exits 0. With -e but without pipefail, python job.py | tee log.txt reports tee's status — 0. Only set -euo pipefail passes the failure through. Where you need to continue deliberately, say so:
if ! python optional_step.py; then
echo "optional step failed — continuing" >&2
fiValidate the output, not just the absence of errors
The most dangerous failures raise nothing at all. Check the result against expectations before declaring success:
def assert_output_sane(gdf, previous_count: int | None = None) -> None:
if len(gdf) == 0:
raise ValueError("output has zero features")
if gdf.crs is None:
raise ValueError("output has no CRS")
if gdf.geometry.isna().any():
raise ValueError(f"{gdf.geometry.isna().sum()} null geometries in output")
if not gdf.geometry.is_valid.all():
raise ValueError(f"{(~gdf.geometry.is_valid).sum()} invalid geometries in output")
if previous_count and len(gdf) < previous_count * 0.5:
raise ValueError(f"feature count halved: {previous_count} → {len(gdf)}")
good = gpd.read_file("output/rutland.gpkg")
checks = {"today's output": (good, None), "yesterday had 3× as many": (good, 3 * len(good)),
"the zero-row file": (good.iloc[0:0], None), "no CRS": (good.set_crs(None, allow_override=True), None)}
for label, (gdf, previous) in checks.items():
try:
assert_output_sane(gdf, previous)
print(f"{label:25} ok ({len(gdf):,} features)")
except ValueError as err:
print(f"{label:25} {err}")today's output ok (9,997 features)
yesterday had 3× as many feature count halved: 29991 → 9997
the zero-row file output has zero features
no CRS output has no CRS
Comparing against the previous run catches the class of failure where an upstream source quietly starts returning a fraction of its data — no error, just less of it.
Write a run summary a machine can read
Human-readable logs are for debugging; a structured summary is for alerting:
import json
from datetime import datetime, timezone
def write_summary(path: Path, **fields) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
record = {"finished_at": datetime.now(timezone.utc).isoformat(timespec="seconds"), **fields}
path.write_text(json.dumps(record, indent=2), encoding="utf-8")
write_summary(Path("logs/last_run.json"), status="ok", features=len(good), output="output/rutland.gpkg",
input_files=len(list(Path("input").glob("*.gpkg"))), failures=0)
print(Path("logs/last_run.json").read_text()){
"finished_at": "2026-10-10T22:38:39+00:00",
"status": "ok",
"features": 9997,
"output": "output/rutland.gpkg",
"input_files": 3,
"failures": 0
}
A monitor that reads last_run.json can alert on status != "ok", on a feature count that dropped, and on a finished_at that is older than expected — which is the only way to detect a job that never started.
Detect the job that did not run
A silent failure includes silence itself. Dead-man's-switch monitoring inverts the logic: alert when the success signal is missing. The heartbeat is a request to a monitoring service on success (a hosted cron monitor gives you a URL such as https://hc-ping.com/<uuid>); without one, a check on the summary's age does the same job:
def heartbeat(url: str) -> None:
try:
requests.get(url, timeout=10)
except Exception:
log.warning("heartbeat failed — not fatal")
def is_stale(summary=Path("logs/last_run.json"), hours=26) -> bool:
finished = datetime.fromisoformat(json.loads(summary.read_text())["finished_at"])
return datetime.now(timezone.utc) - finished > timedelta(hours=hours)
heartbeat(HEARTBEAT_URL)
print("heartbeats received:", len(pings), "| summary stale:", is_stale())heartbeats received: 1 | summary stale: False
An entry point with a real contract
The data used on this page
Three folders built from Geofabrik's OpenStreetMap shapefile extract for Rutland, each layer saved as a GeoPackage in British National Grid:
input/— roads, land use and places: a normal night.input_today/— empty: the night the delivery did not arrive.input_with_bad_file/— roads, plus the buildings layer cut to half its length, as an interrupted copy leaves it.
The setup also starts a local HTTP listener that records heartbeat and alert requests, standing in for a monitoring service.
import io
import logging
import os
import shutil
import subprocess
import sys
import threading
import zipfile
from datetime import datetime, timedelta, timezone
from http.server import BaseHTTPRequestHandler, HTTPServer
from pathlib import Path
import geopandas as gpd
import pandas as pd
import requests
for folder in ["input", "input_today", "input_with_bad_file", "output", "logs"]:
Path(folder).mkdir(exist_ok=True)
if not Path("input/roads.gpkg").exists():
reply = requests.get("https://download.geofabrik.de/europe/united-kingdom/england/rutland-260101-free.shp.zip",
timeout=300)
reply.raise_for_status()
zipfile.ZipFile(io.BytesIO(reply.content)).extractall("osm")
for name, layer in [("roads", "roads_free_1"), ("landuse", "landuse_a_free_1"), ("places", "places_free_1"),
("buildings", "buildings_a_free_1")]:
gpd.read_file(f"osm/gis_osm_{layer}.shp").to_crs(27700).to_file(f"input/{name}.gpkg")
shutil.move("input/buildings.gpkg", "input_with_bad_file/buildings_full.gpkg")
shutil.copy("input/roads.gpkg", "input_with_bad_file/roads.gpkg")
data = Path("input_with_bad_file/buildings_full.gpkg").read_bytes()
Path("input_with_bad_file/buildings.gpkg").write_bytes(data[: len(data) // 2]) # an interrupted copy
Path("input_with_bad_file/buildings_full.gpkg").unlink()
pings = []
class Listener(BaseHTTPRequestHandler):
def do_GET(self):
pings.append(self.path)
self.send_response(200)
self.end_headers()
do_POST = do_GET
def log_message(self, *args):
pass
server = HTTPServer(("127.0.0.1", 0), Listener)
threading.Thread(target=server.serve_forever, daemon=True).start()
HEARTBEAT_URL = f"http://127.0.0.1:{server.server_port}/ping/rutland-nightly"
os.environ["ALERT_URL"] = f"http://127.0.0.1:{server.server_port}/alert"
for folder in ["input", "input_today", "input_with_bad_file"]:
print(folder, sorted(p.name for p in Path(folder).glob("*.gpkg")))input ['landuse.gpkg', 'places.gpkg', 'roads.gpkg']
input_today []
input_with_bad_file ['buildings.gpkg', 'roads.gpkg']
Example 1: a pipeline entry point with a real contract
import time
log = logging.getLogger("nightly")
def run(cfg) -> dict:
started = time.perf_counter()
files = sorted(Path(cfg["input_dir"]).glob("*.gpkg"))
if not files:
raise FileNotFoundError(f"no input files in {cfg['input_dir']}")
frames, failures = [], []
for path in files:
try:
frames.append(gpd.read_file(path))
except Exception as exc:
failures.append((path.name, str(exc)))
log.warning("skipping %s: %s", path.name, exc)
if not frames:
raise RuntimeError(f"all {len(files)} inputs failed")
out = gpd.GeoDataFrame(pd.concat(frames, ignore_index=True), crs=frames[0].crs)
assert_output_sane(out)
dest = Path(cfg["output_dir"]) / "rutland.gpkg"
dest.parent.mkdir(parents=True, exist_ok=True)
tmp = dest.with_name(dest.stem + ".part.gpkg") # GDAL picks the driver from the extension
out.to_file(tmp, driver="GPKG")
tmp.replace(dest) # atomic on the same filesystem
return {"features": len(out), "output": str(dest), "inputs": len(files), "failures": len(failures),
"duration_s": round(time.perf_counter() - started, 1)}
def main_contract(input_dir) -> int:
try:
result = run({"input_dir": input_dir, "output_dir": "output"})
except Exception as exc:
log.error("run failed: %s", exc)
write_summary(Path("logs/last_run.json"), status="failed", error=str(exc))
return 1
write_summary(Path("logs/last_run.json"), status="ok", **result)
return 1 if result["failures"] else 0
for folder in ["input", "input_with_bad_file", "input_today"]:
code = main_contract(folder)
summary = json.loads(Path("logs/last_run.json").read_text())
print(f"{folder:20} exit {code} status={summary['status']} features={summary.get('features', '-')}"
f" failures={summary.get('failures', '-')}")input exit 0 status=ok features=9997 failures=0
input_with_bad_file exit 1 status=ok features=7292 failures=1
input_today exit 1 status=failed features=- failures=-
A batch where one input failed still writes the output from the others, but exits 1 and records the failure count, so the partial result is never mistaken for a complete one. The temporary file is named rutland.part.gpkg rather than rutland.gpkg.part: GDAL chooses the driver from the extension, and a .part suffix would leave it guessing.
Example 2: a decorator that guarantees a step is logged
import functools
logging.basicConfig(filename="logs/steps.log", level=logging.INFO, force=True, format="%(levelname)s %(message)s")
steps = logging.getLogger("pipeline")
def step(name):
def decorator(fn):
@functools.wraps(fn)
def wrapper(*args, **kwargs):
steps.info("→ %s", name)
try:
result = fn(*args, **kwargs)
except Exception:
steps.exception("✗ %s failed", name)
raise
steps.info("✓ %s", name)
return result
return wrapper
return decorator
@step("clip to Oakham")
def clip_to_box(gdf, box):
return gpd.clip(gdf, box)
roads = gpd.read_file("input/roads.gpkg")
print(len(clip_to_box(roads, (485000, 307000, 488000, 310000))), "road segments in the box")
try:
clip_to_box(roads, "not a box")
except Exception:
pass
print([line.split("\n")[0] for line in Path("logs/steps.log").read_text().splitlines() if line[:4] in ("INFO", "ERRO")])1138 road segments in the box
['INFO → clip to Oakham', 'INFO ✓ clip to Oakham', 'INFO → clip to Oakham', 'ERROR ✗ clip to Oakham failed']
Every step now reports start, success or failure, and no step can fail quietly — even when a caller further up swallows the exception, as the except: pass here does, the log already has it.
Example 3: guard against a shrinking dataset
STATE = Path("logs/last_run.json")
def check_against_previous(count: int, tolerance: float = 0.2) -> None:
if not STATE.exists():
return
previous = json.loads(STATE.read_text()).get("features")
if previous and count < previous * (1 - tolerance):
raise ValueError(f"feature count dropped {previous} → {count} (>{tolerance:.0%})")
write_summary(STATE, status="ok", features=len(good))
check_against_previous(len(good))
try:
check_against_previous(len(gpd.read_file("input/roads.gpkg"))) # only one of three layers arrived
except ValueError as err:
print("ValueError:", err)ValueError: feature count dropped 9997 → 7292 (>20%)
Example 4: fail the shell wrapper properly
%%writefile nightly.sh
#!/usr/bin/env bash
set -euo pipefail
LOG="logs/nightly-$(date +%Y%m%d).log"
on_error() {
local code=$?
echo "FAILED at line ${BASH_LINENO[0]} with exit $code" >&2
curl -fsS -m 10 "$ALERT_URL?job=nightly&status=failed&code=$code" || true
exit "$code"
}
trap on_error ERR
"$PY" nightly.py "$1" >> "$LOG" 2>&1
curl -fsS -m 10 "$ALERT_URL?job=nightly&status=ok" || truefor folder in ["input", "input_today"]:
before = len(pings)
run = subprocess.run(["bash", "nightly.sh", folder], capture_output=True, text=True,
env=dict(os.environ, PY=sys.executable))
print(f"{folder:12} exit {run.returncode} | sent: {pings[before:]} | {run.stderr.strip()}")input exit 0 | sent: ['/alert?job=nightly&status=ok'] |
input_today exit 3 | sent: ['/alert?job=nightly&status=failed&code=3'] | FAILED at line 14 with exit 3
The trap sends the failure with its exit code — 3 here, "no input" — and the wrapper still exits with that code, so the scheduler sees it too.
Example 5: the edge cases, checked
import warnings
from concurrent.futures import ThreadPoolExecutor
def code_of(snippet, *flags):
return subprocess.run([sys.executable, *flags, "-c", snippet], capture_output=True, text=True)
print("sys.exit('message') ->", code_of("import sys; sys.exit('bad config')").returncode)
print("sys.exit(256) ->", code_of("import sys; sys.exit(256)").returncode)
print("assert under -O ->", code_of("assert False, 'never checked'", "-O").returncode)
with warnings.catch_warnings(record=True) as caught:
warnings.simplefilter("always")
swallow = compile("def f():\n try:\n 1 / 0\n finally:\n return 'ok'\n", "demo", "exec")
namespace = {}
exec(swallow, namespace)
print("return in finally ->", namespace["f"](), "|", [str(w.category.__name__) for w in caught])
with ThreadPoolExecutor() as pool:
future = pool.submit(lambda: gpd.read_file("input_with_bad_file/buildings.gpkg"))
print("exception in a thread -> stored, not raised:", type(future.exception()).__name__)sys.exit('message') -> 1
sys.exit(256) -> 0
assert under -O -> 0
return in finally -> ok | ['SyntaxWarning']
exception in a thread -> stored, not raised: DataSourceError
How far a failure travels before anyone hears
A failure has to travel a long way before a human hears about it. An exception is raised inside a function; a caller either propagates or catches it; main() turns the outcome into an exit code; the process hands that code to a shell; the shell hands its own status to the scheduler; the scheduler decides whether to alert. Each hand-off can drop the signal, and dropping it anywhere makes the whole run look successful.
The most common break is the first one. except Exception: pass is usually written to get past one annoying edge case, and then it quietly absorbs every future bug in the same block. The second most common is the exit code: a script that ends by falling off the bottom of the file exits 0, no matter what it logged on the way. The third lives in the shell, where the default is to keep going after an error and report only the last command's status.
Then there is the failure mode that no exception handling can catch, because nothing raised: the job read an empty folder, wrote an empty file, and finished. Correctness here is not about errors at all — it is about expectations. A pipeline that knows how many features it should produce, what CRS the output must carry, and roughly how long it should take can detect wrongness that Python has no opinion about. That is why output assertions belong in the same tier as exception handling rather than in a "nice to have" tier.
Finally, monitoring needs to cover absence. If the scheduler never fired, or the machine was down, there is no log, no exit code and no error to notice — just silence, which reads exactly like a quiet success. A heartbeat or a freshness check on the run summary is the only thing that distinguishes the two.
Exit codes that do not mean what you think
sys.exit("message")exits with code 1: Passing a string prints it to stderr and uses status 1 (checked in Example 5). Pass an int when you mean a specific code.- Exceptions in threads do not reach the main thread: A
ThreadPoolExecutorstores them in theFuture. Callfuture.result()or the failure is invisible. assertdisappears under-O: Never use bareassertfor production validation; raise a real exception instead.loggingswallows its own errors: A misconfigured handler will not crash the program. Verify the log file is actually being written after a deploy.- Warnings are not failures: A geographic-CRS warning from GeoPandas is easy to lose in a long log. Elevate the ones that matter with
warnings.simplefilter("error", UserWarning)in tests. - Exit codes above 255 wrap: On POSIX only the low 8 bits survive, so
sys.exit(256)exits 0 — a failure code that reads as success. Keep codes between 1 and 125. - A
finallythat returns swallows the exception: Returning fromfinallydiscards the in-flight exception entirely — the check in Example 5 returned'ok'from a division by zero. Python 3.14 now emits aSyntaxWarningfor it (PEP 765); never do it.
Internal links
- How to Get Alerted When an Automated GIS Job Fails
- How to Validate Pipeline Inputs and Outputs Automatically in Python
- How to Log and Summarise Errors in a Batch GIS Job in Python
- How to Build a GIS Data Pipeline in Python: The Complete Workflow
- Batch Script Stops at the First Bad File: How to Keep It Running
- Python GIS Script Works Manually but Not from Cron: How to Fix It
FAQ
Why does my scheduler show a successful run when the job clearly failed?
Because the process exited 0. Either the exception was caught and not re-raised, or main() returned nothing. End the script with raise SystemExit(main()) and return non-zero on failure.
Is except Exception always wrong?
No — it is right at a batch boundary where one item's failure must not stop the others, provided you record the exception. It is wrong when it hides the error, which is why pass is the real problem rather than the broad catch.
How do I make a shell wrapper fail properly?
Start it with set -euo pipefail, and add a trap ... ERR handler if you want to send an alert. Without pipefail, piping output through tee masks the script's exit code entirely.
What should the exit code be when there was nothing to process?
Decide deliberately and document it. Many pipelines treat an empty input as a failure, because "no files today" usually means an upstream delivery is missing rather than genuinely nothing to do.
How do I detect a job that never ran at all?
Write a run summary with a timestamp on every success, and alert when it becomes stale. A heartbeat ping to a cron-monitoring service does the same thing without extra infrastructure.
Should validation raise or just log a warning?
Raise for anything that makes the output unusable — zero rows, missing CRS, invalid geometries. Warn for conditions worth watching that do not invalidate the result, and put those counts in the run summary so a trend is visible.
How do I stop a partially written output from looking like a success?
Write to a temporary path and rename it into place once the write returns. Path.replace() is atomic, so the final file only ever exists in a complete state.