How to Add a Dry-Run Mode to a Batch GIS Script
Problem statement
The script is ready. It will read 900 shapefiles, reproject them, clip them to a boundary, and write 900 GeoPackages into a delivery folder — overwriting whatever is there. You are fairly sure the paths are right.
"Fairly sure" is the problem. Run it and find out, and the failure modes are:
wrote 900 files to /home/gis/data/out # wrong folder — cwd was not what you thought
wrote 12 files, then: DriverError on file 13 # 40 minutes wasted
overwrote last month's delivery # no undoA dry run answers "what would this do?" without doing it. For a batch job that writes hundreds of files, deletes anything, or pushes to a database, it turns an irreversible action into a reviewable plan.
What a dry run needs to show:
- exactly which inputs were matched, and which were skipped
- exactly which outputs would be written, and which already exist
- which operations are destructive (overwrite, delete, upload)
- an estimate of the time and disk the real run would need
- any validation problems that would fail the run anyway
Quick answer
Separate planning from executing, then make the executor optional:
- build a plan: a list of (input, output, action) records — no side effects
- validate the plan: missing inputs, existing outputs, collisions, disk space
- print the plan, and stop there if
--dry-runwas passed - execute the same plan when it was not
- make dry run the default for anything destructive
import argparse
from pathlib import Path
def build_plan(src: Path, out: Path):
plan = []
for shp in sorted(src.rglob("*.shp")):
dest = (out / shp.relative_to(src)).with_suffix(".gpkg")
plan.append({"src": shp, "dest": dest,
"action": "overwrite" if dest.exists() else "create"})
return plan
ap = argparse.ArgumentParser()
ap.add_argument("--dry-run", action="store_true", help="show what would happen, change nothing")
args = ap.parse_args()
plan = build_plan(Path("data/raw"), Path("data/out"))
creates = sum(1 for p in plan if p["action"] == "create")
overwrites = len(plan) - creates
print(f"{len(plan)} operations: {creates} create, {overwrites} overwrite")
if args.dry_run:
for p in plan[:20]:
print(f" [{p['action']:9}] {p['src']} → {p['dest']}")
raise SystemExit(0)
for p in plan:
convert(p["src"], p["dest"])The key structural move is that build_plan() has no side effects. Once planning and doing are separate functions, a dry run is one if — and the plan becomes testable on its own.
Plan, then execute
Step-by-step solution
Make the plan a data structure
A plan is a list of records, not a sequence of actions. That means you can count it, sort it, diff it, save it and test it.
from dataclasses import dataclass, field
from pathlib import Path
@dataclass
class Operation:
src: Path
dest: Path
action: str # create | overwrite | skip | delete
reason: str = ""
size_in: int = 0
warnings: list[str] = field(default_factory=list)
def line(self) -> str:
mark = {"create": "+", "overwrite": "~", "skip": ".", "delete": "-"}[self.action]
warn = f" ⚠ {'; '.join(self.warnings)}" if self.warnings else ""
return f" {mark} {self.src.name:<38} → {self.dest}{warn}"The action field is what makes the report readable at a glance: a plan of 900 + lines is comfortable, and one containing 40 ~ lines deserves a second look.
Build the plan without touching anything
from pathlib import Path
def build_plan(src_root: Path, out_root: Path, overwrite: bool = False) -> list[Operation]:
ops = []
for shp in sorted(src_root.rglob("*.shp")):
dest = (out_root / shp.relative_to(src_root)).with_suffix(".gpkg")
warnings = []
for ext in (".shx", ".dbf"):
if not shp.with_suffix(ext).exists():
warnings.append(f"missing {ext}")
if not shp.with_suffix(".prj").exists():
warnings.append("no .prj (CRS unknown)")
if dest.exists() and not overwrite:
action, reason = "skip", "output exists"
elif dest.exists():
action, reason = "overwrite", "output exists, --overwrite given"
else:
action, reason = "create", ""
ops.append(Operation(shp, dest, action, reason,
size_in=shp.stat().st_size, warnings=warnings))
return opsReading file metadata is fine — it changes nothing. The rule is simply that a planner may read, and must not write.
Validate the plan before showing it
import shutil
from collections import Counter
def validate(ops: list[Operation], out_root: Path) -> list[str]:
problems = []
dest_counts = Counter(op.dest for op in ops if op.action != "skip")
for dest, n in dest_counts.items():
if n > 1:
srcs = [str(o.src) for o in ops if o.dest == dest]
problems.append(f"{n} inputs would write to {dest}: {srcs}")
to_write = sum(op.size_in for op in ops if op.action != "skip")
free = shutil.disk_usage(out_root.parent if out_root.exists() else Path.cwd()).free
if to_write * 1.5 > free:
problems.append(f"needs ~{to_write*1.5/1e9:.1f} GB, {free/1e9:.1f} GB free")
if not ops:
problems.append("no inputs matched — check the source folder and pattern")
return problemsOutput collisions and a full disk are exactly the failures that ruin a long run halfway through. Both are detectable before the first byte is written.
Report the plan usefully
Nobody reads 900 lines. Summarise, then show a sample and everything unusual.
from collections import Counter
def report(ops: list[Operation], problems: list[str], sample: int = 10) -> None:
counts = Counter(op.action for op in ops)
total_in = sum(op.size_in for op in ops if op.action != "skip")
print("── plan ──────────────────────────────────────────")
for action in ("create", "overwrite", "skip", "delete"):
if counts.get(action):
print(f"{action:>10}: {counts[action]:>5}")
print(f"{'input size':>10}: {total_in/1e9:>5.2f} GB")
print(f"\nfirst {sample} operations:")
for op in ops[:sample]:
print(op.line())
if len(ops) > sample:
print(f" … {len(ops) - sample} more")
flagged = [op for op in ops if op.warnings]
if flagged:
print(f"\n{len(flagged)} inputs with warnings:")
for op in flagged[:sample]:
print(op.line())
destructive = [op for op in ops if op.action in ("overwrite", "delete")]
if destructive:
print(f"\n⚠ {len(destructive)} destructive operations")
for op in destructive[:sample]:
print(op.line())
if problems:
print("\n✗ blocking problems:")
for p in problems:
print(f" {p}")Gate the execution
import sys
def main() -> int:
args = parse_args()
ops = build_plan(args.src, args.out, overwrite=args.overwrite)
problems = validate(ops, args.out)
report(ops, problems)
if problems:
print("\nrefusing to run — fix the problems above", file=sys.stderr)
return 2
if args.dry_run:
print("\ndry run: nothing was written")
return 0
destructive = sum(1 for op in ops if op.action in ("overwrite", "delete"))
if destructive and not args.yes and sys.stdin.isatty():
answer = input(f"\n{destructive} files will be overwritten. Continue? [y/N] ")
if answer.strip().lower() not in {"y", "yes"}:
return 1
return execute(ops)The confirmation prompt is guarded by sys.stdin.isatty() so an unattended run never hangs waiting for input that will never come — a genuinely common way to lose a night's processing.
Execute the plan, unchanged
import logging
log = logging.getLogger("batch")
def execute(ops: list[Operation]) -> int:
ok, failed = 0, []
for i, op in enumerate((o for o in ops if o.action != "skip"), start=1):
try:
op.dest.parent.mkdir(parents=True, exist_ok=True)
convert(op.src, op.dest)
ok += 1
except Exception as exc:
failed.append((op.src.name, f"{type(exc).__name__}: {exc}"))
log.warning("failed %s: %s", op.src.name, exc)
log.info("%d written, %d failed", ok, len(failed))
return 1 if failed else 0The executor consumes exactly the plan that was printed. If it iterated the filesystem again, the preview and the run could differ — which would defeat the point.
Wire up the flags
import argparse
from pathlib import Path
def parse_args():
ap = argparse.ArgumentParser(description="Convert shapefiles to GeoPackage")
ap.add_argument("src", type=Path)
ap.add_argument("out", type=Path)
ap.add_argument("-n", "--dry-run", action="store_true",
help="show what would happen and exit")
ap.add_argument("--overwrite", action="store_true",
help="replace existing outputs instead of skipping them")
ap.add_argument("-y", "--yes", action="store_true",
help="skip the confirmation prompt (for unattended runs)")
ap.add_argument("--plan-json", type=Path, help="write the plan as JSON")
return ap.parse_args()-n for dry run follows rsync, make and git clean; users will guess it correctly.
Code examples
Example 1: the complete pattern
#!/usr/bin/env python3
"""convert.py — plan, review, then convert."""
from dataclasses import dataclass, field, asdict
from pathlib import Path
from collections import Counter
import argparse, json, logging, shutil, sys
import geopandas as gpd
log = logging.getLogger("convert")
@dataclass
class Operation:
src: Path
dest: Path
action: str
size_in: int = 0
warnings: list[str] = field(default_factory=list)
def build_plan(src_root: Path, out_root: Path, overwrite: bool) -> list[Operation]:
ops = []
for shp in sorted(src_root.rglob("*.shp")):
dest = (out_root / shp.relative_to(src_root)).with_suffix(".gpkg")
warnings = [f"missing {e}" for e in (".shx", ".dbf") if not shp.with_suffix(e).exists()]
action = "create" if not dest.exists() else ("overwrite" if overwrite else "skip")
ops.append(Operation(shp, dest, action, shp.stat().st_size, warnings))
return ops
def convert(src: Path, dest: Path) -> None:
gdf = gpd.read_file(src)
dest.parent.mkdir(parents=True, exist_ok=True)
tmp = dest.with_name(dest.name + ".part")
gdf.to_file(tmp, driver="GPKG")
tmp.replace(dest)
def main() -> int:
ap = argparse.ArgumentParser()
ap.add_argument("src", type=Path); ap.add_argument("out", type=Path)
ap.add_argument("-n", "--dry-run", action="store_true")
ap.add_argument("--overwrite", action="store_true")
ap.add_argument("-y", "--yes", action="store_true")
ap.add_argument("--plan-json", type=Path)
args = ap.parse_args()
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(message)s")
if not args.src.is_dir():
print(f"source not found: {args.src.resolve()}", file=sys.stderr)
return 2
ops = build_plan(args.src, args.out, args.overwrite)
counts = Counter(o.action for o in ops)
dupes = [d for d, n in Counter(o.dest for o in ops if o.action != "skip").items() if n > 1]
print(f"source : {args.src.resolve()}")
print(f"target : {args.out.resolve()}")
print(f"plan : {dict(counts)} ({sum(o.size_in for o in ops)/1e9:.2f} GB in)")
for op in ops[:10]:
print(f" {op.action:>9} {op.src.name} → {op.dest}")
if len(ops) > 10:
print(f" … {len(ops)-10} more")
if args.plan_json:
args.plan_json.write_text(json.dumps(
[{**asdict(o), "src": str(o.src), "dest": str(o.dest)} for o in ops], indent=2))
if dupes:
print(f"\n✗ {len(dupes)} output collisions", file=sys.stderr)
return 2
if args.dry_run:
print("\ndry run: nothing written")
return 0
if counts.get("overwrite") and not args.yes and sys.stdin.isatty():
answer = input(f"overwrite {counts['overwrite']} files? [y/N] ")
if answer.strip().lower() not in {"y", "yes"}:
return 1
failed = []
for op in (o for o in ops if o.action != "skip"):
try:
convert(op.src, op.dest)
except Exception as exc:
failed.append((op.src.name, str(exc)))
log.warning("failed %s: %s", op.src.name, exc)
written = len(ops) - counts.get("skip", 0) - len(failed)
log.info("done: %d written, %d failed", written, len(failed))
return 1 if failed else 0
if __name__ == "__main__":
raise SystemExit(main())Example 2: dry-run a database write
Filesystem operations are easy to preview; database writes need the same discipline.
from sqlalchemy import create_engine, text
def load_to_postgis(gdf, table, engine, dry_run=False):
with engine.connect() as conn:
exists = conn.execute(text(
"SELECT to_regclass(:t) IS NOT NULL"), {"t": f"public.{table}"}).scalar()
current = conn.execute(text(f"SELECT count(*) FROM {table}")).scalar() if exists else 0
print(f"{'would ' if dry_run else ''}write {len(gdf)} rows to {table} "
f"({'replacing' if exists else 'creating'}, currently {current} rows)")
if dry_run:
return {"rows": len(gdf), "table": table, "existing_rows": current, "applied": False}
gdf.to_postgis(table, engine, if_exists="replace", index=False)
return {"rows": len(gdf), "table": table, "existing_rows": current, "applied": True}Returning the same record shape in both modes means the caller's reporting code does not branch.
Example 3: transaction-style dry run for a whole pipeline
class Runner:
"""Collects intended side effects; applies them only when live."""
def __init__(self, dry_run: bool = True):
self.dry_run = dry_run
self.actions = []
def write(self, gdf, dest, **kwargs):
self.actions.append(("write", str(dest), len(gdf)))
if not self.dry_run:
Path(dest).parent.mkdir(parents=True, exist_ok=True)
gdf.to_file(dest, **kwargs)
def delete(self, path):
self.actions.append(("delete", str(path), None))
if not self.dry_run:
Path(path).unlink(missing_ok=True)
def summary(self):
from collections import Counter
counts = Counter(a[0] for a in self.actions)
prefix = "would " if self.dry_run else ""
return f"{prefix}{dict(counts)}"
runner = Runner(dry_run=args.dry_run)
runner.write(clean, "data/out/parcels.gpkg", driver="GPKG")
runner.delete("data/out/parcels.gpkg.part")
print(runner.summary())One object owns every side effect, so --dry-run is enforced in a single place instead of being remembered at each call site.
Example 4: test the planner without any data
from pathlib import Path
def test_plan_marks_existing_outputs_as_skip(tmp_path):
src, out = tmp_path / "in", tmp_path / "out"
(src / "north").mkdir(parents=True)
for ext in (".shp", ".shx", ".dbf"):
(src / "north" / f"roads{ext}").write_bytes(b"")
(out / "north").mkdir(parents=True)
(out / "north" / "roads.gpkg").write_bytes(b"")
ops = build_plan(src, out, overwrite=False)
assert len(ops) == 1
assert ops[0].action == "skip"
assert ops[0].dest == out / "north" / "roads.gpkg"Because the planner is pure, the whole preview logic is testable with empty files and no GIS stack at all.
Explanation
Most batch scripts interleave three things — discovering work, deciding what to do, and doing it — inside one loop. That structure makes a preview impossible: you cannot show what the script will do without running the loop, and running the loop is the thing you wanted to avoid.
Splitting the script into a planner and an executor fixes it structurally. The planner reads the filesystem and produces a list of intended operations; the executor consumes that list. --dry-run then means "stop after the planner", and because both modes use the same plan, the preview is exactly what will happen rather than an approximation of it.
That separation pays off in several other ways. The plan can be counted and summarised, so you learn that 40 of your 900 outputs already exist before the run rather than after. It can be validated: two inputs mapping to one output is a bug you can catch statically, as is not having the disk space for the result. It can be serialised to JSON and attached to a ticket or diffed against last month's. And it can be unit-tested with empty files, because a pure planner needs no real data.
The last consideration is defaults. For anything reversible, live is a fine default. For a script that overwrites a delivery folder, deletes anything, publishes to a server, or writes to a production database, the safer default is dry run with an explicit --apply to commit — the convention terraform, kubectl and most infrastructure tools converged on for the same reason. Whichever you choose, guard interactive confirmations with sys.stdin.isatty(), or the first scheduled run will hang forever on a prompt nobody can see.
Edge cases or notes
- A dry run must not write anything — including logs into the output folder: Keep run logs somewhere separate, or a "preview" creates directories.
- Plans go stale: If files can change between planning and execution, re-check
dest.exists()at write time, or run both phases back to back. --dry-runin a scheduler is a no-op job: Make sure the exit code says "nothing was done", and never let a scheduled run silently stay in preview mode.- Prompts hang unattended jobs: Guard
input()withsys.stdin.isatty()and provide--yesfor automation. - Deletes deserve extra care: Show the full list of paths that would be removed, not a count, and consider moving to a trash folder rather than unlinking.
- Estimating output size is guesswork: Input bytes are a poor proxy for output bytes when the format changes. State it as an estimate.
- Reproducibility: Sort the plan deterministically so two dry runs of unchanged data produce byte-identical output that can be diffed.
Internal links
- How to Turn a GIS Script into a Command-Line Tool with argparse
- How to Batch Process a Folder of GIS Files in Python: The Complete Workflow
- How to Validate Pipeline Inputs and Outputs Automatically in Python
- Batch Output Files Keep Overwriting Each Other: How to Fix It
- How to Build an Inventory of a GIS Data Folder in Python
- How to Test a GIS Pipeline with pytest
FAQ
What exactly should a dry run print?
Counts by action (create, overwrite, skip, delete), a sample of the operations, every destructive one, every input with a warning, and any blocking problem such as an output collision or insufficient disk space.
Should dry run be the default?
For destructive jobs — overwriting deliveries, deleting files, writing to production databases — yes, with an explicit --apply to commit. For safe, idempotent jobs, live is a reasonable default with -n available.
How do I dry-run a database load?
Query for the table's existence and current row count, report what would be written, and return before calling to_postgis. Keep the return shape identical in both modes so callers do not branch.
How do I stop the preview and the real run drifting apart?
Have the executor consume the plan object rather than re-scanning the filesystem. One list, produced once, used by both paths.
Will a confirmation prompt break my scheduled job?
Yes, unless you guard it. Only prompt when sys.stdin.isatty() is true, and offer --yes so automation can pass explicitly.
Can I keep the plan for the record?
Write it as JSON alongside the run log. It diffs cleanly against previous runs and answers "what did last Tuesday's job actually intend to do?" months later.
How do I test the dry-run logic?
Because the planner has no side effects, you can build a temporary tree of empty files and assert on the operations it produces — no GIS data or dependencies required.