The Anatomy of a Batch Job: Discover, Apply, Report #

Problem statement #

Every batch script starts the same way and ends up unmaintainable for the same reason.

import glob, geopandas as gpd

for f in glob.glob("data/*.shp"):
    gdf = gpd.read_file(f)
    gdf = gdf.to_crs(27700)
    gdf["area"] = gdf.area
    gdf.to_file(f.replace(".shp", ".gpkg"), driver="GPKG")

Four lines, and it works. Then the requirements arrive one at a time: skip files already done, keep going when one fails, handle subfolders, log what happened, do a dry run first, run four at a time. Each one gets bolted onto the loop, and after six weeks the script is a hundred lines where discovery, transformation, error handling and reporting are interleaved beyond untangling.

The problem is not any individual requirement. It is that the script has no structure to hang them on. A batch job has a shape, and every one of those requirements belongs to a specific part of it.

Quick answer #

Every batch job is three separable stages plus a policy:

  DISCOVER  →  what items exist, and which still need doing
  APPLY     →  one pure function, run once per item
  REPORT    →  what happened to each item, and the totals
              (+ a failure policy that decides what a bad item does)

from dataclasses import dataclass
from pathlib import Path

@dataclass
class Result:
    item: Path
    status: str          # "ok" | "skipped" | "failed"
    detail: str = ""
    rows: int = 0

def discover(src: Path, dst: Path) -> list[Path]:
    """Everything that exists, minus everything already done."""
    return [p for p in sorted(src.rglob("*.shp"))
            if not (dst / p.relative_to(src)).with_suffix(".gpkg").exists()]

def apply_one(path: Path, dst: Path) -> Result:
    """One item. Knows nothing about the batch, the loop, or the report."""
    gdf = gpd.read_file(path).to_crs(27700)
    out = (dst / path.relative_to(SRC)).with_suffix(".gpkg")
    out.parent.mkdir(parents=True, exist_ok=True)
    gdf.to_file(out, driver="GPKG")
    return Result(path, "ok", rows=len(gdf))

def run(src: Path, dst: Path) -> list[Result]:
    """The batch. Knows nothing about shapefiles."""
    results = []
    for item in discover(src, dst):
        try:
            results.append(apply_one(item, dst))
        except Exception as exc:
            results.append(Result(item, "failed", f"{type(exc).__name__}: {exc}"))
    return results

Every later requirement now has an obvious home:

Not one of them touches the transformation. That is the point of the shape.

The three stages #

Step-by-step solution #

Discover: produce the work list, and nothing else #

Discovery answers one question — what should this run process? — and it should be callable on its own, because that is what makes a dry run trivial.

def discover(src: Path, dst: Path, pattern: str = "*.shp") -> list[Path]:
    candidates = sorted(src.rglob(pattern))
    return [p for p in candidates if not output_for(p, src, dst).exists()]

Two properties matter:

• Deterministic. sorted() is not cosmetic. Unsorted glob output varies by filesystem, which makes two runs incomparable and any "resume from item 400" logic meaningless.
• Side-effect free. Discovery must not create, move or write anything, or you cannot ask it what it would do.

Filtering out completed work here — rather than checking inside the loop — is what makes the job resumable. Re-running after a crash simply discovers less. See how to build a resumable batch GIS job.

Apply: one item, one function, no batch awareness #

def apply_one(path: Path) -> Result:
    ...

The signature is the design. apply_one takes one item and returns one result. It does not know how many items there are, whether it is item 1 or item 4,000, whether other items failed, or whether this is a dry run.

That constraint buys three things at once:

1. It is testable. One fixture file, one call, assertions on the result. No loop, no folder, no mocking.
2. It is parallelisable. A function of one item with no shared state can go into a ProcessPoolExecutor unchanged.
3. It is reusable. The same function serves the batch job and the one-off "just do this file" case.

The moment apply_one needs to know about its neighbours — a running total, a shared connection, a progress counter — one of those three properties is gone. Usually the fix is to return more in the Result and let report do the aggregation.

Report: the run's output is data, not print statements #

def report(results: list[Result]) -> dict:
    by_status = Counter(r.status for r in results)
    return {
        "items": len(results),
        "ok": by_status["ok"],
        "skipped": by_status["skipped"],
        "failed": by_status["failed"],
        "rows": sum(r.rows for r in results),
        "failures": [
            {"item": str(r.item), "detail": r.detail}
            for r in results if r.status == "failed"
        ],
    }

A batch job that prints as it goes produces a wall of text nobody reads and nothing a machine can check. A batch job that returns structured results can print a summary, write a JSON log, fail CI on a non-zero failure count, and be compared against last night's run — all from the same object.

The single most useful line in any batch job:

summary = report(results)
print(f"{summary['ok']} ok · {summary['skipped']} skipped · {summary['failed']} failed")
if summary["failed"]:
    for f in summary["failures"][:10]:
        print(f"  ✗ {f['item']}: {f['detail']}")

See how to log and summarise errors in a batch GIS job.

The fourth thing: a failure policy #

The three stages describe the happy path. What happens when apply_one raises is a separate decision — fail the whole run, skip the item, or quarantine it — and it is important enough to have its own page. What matters structurally is that the policy lives in run, wrapped around apply_one, and not inside the transformation.

Why the shape survives growth #

# the loop, later — same three stages, more capability
def run(src, dst, *, dry_run=False, workers=1, on_error="skip"):
    items = discover(src, dst)
    if dry_run:
        return [Result(i, "skipped", "dry run") for i in items]

    execute = partial(_guarded, apply_one, on_error=on_error)
    if workers > 1:
        with ProcessPoolExecutor(workers) as pool:
            return list(pool.map(execute, items))
    return [execute(i) for i in items]

Dry run, parallelism and error policy all landed in run. discover and apply_one were not touched. That is what "has a shape" means in practice — the script absorbs new requirements without any of them reaching the code that does the actual GIS work.

Code examples #

Example 1: the complete skeleton #

from concurrent.futures import ProcessPoolExecutor
from dataclasses import dataclass, asdict
from collections import Counter
from functools import partial
from pathlib import Path
import json, geopandas as gpd

@dataclass
class Result:
    item: str
    status: str
    detail: str = ""
    rows: int = 0

def output_for(item: Path, src: Path, dst: Path) -> Path:
    return (dst / item.relative_to(src)).with_suffix(".gpkg")

def discover(src: Path, dst: Path, pattern="*.shp") -> list[Path]:
    return [p for p in sorted(src.rglob(pattern)) if not output_for(p, src, dst).exists()]

def apply_one(item: Path, src: Path, dst: Path) -> Result:
    gdf = gpd.read_file(item)
    if gdf.crs is None:
        raise ValueError("no CRS")
    gdf = gdf.to_crs(27700)
    out = output_for(item, src, dst)
    out.parent.mkdir(parents=True, exist_ok=True)
    gdf.to_file(out, driver="GPKG")
    return Result(str(item), "ok", rows=len(gdf))

def _guarded(fn, item, **kw) -> Result:
    try:
        return fn(item, **kw)
    except Exception as exc:
        return Result(str(item), "failed", f"{type(exc).__name__}: {exc}")

def run(src: Path, dst: Path, *, dry_run=False, workers=1) -> list[Result]:
    items = discover(src, dst)
    if dry_run:
        return [Result(str(i), "skipped", "dry run") for i in items]
    work = partial(_guarded, apply_one, src=src, dst=dst)
    if workers > 1:
        with ProcessPoolExecutor(workers) as pool:
            return list(pool.map(work, items))
    return [work(i) for i in items]

def report(results, path: Path | None = None) -> dict:
    counts = Counter(r.status for r in results)
    summary = {
        "items": len(results), **counts,
        "rows": sum(r.rows for r in results),
        "failures": [asdict(r) for r in results if r.status == "failed"],
    }
    if path:
        path.write_text(json.dumps(summary, indent=2))
    return summary

Sixty lines, and it does everything the hundred-line script did — plus dry run, resume, parallelism and a machine-readable log.

Example 2: testing each stage independently #

def test_discover_skips_completed(tmp_path):
    src, dst = tmp_path / "in", tmp_path / "out"
    (src / "a").mkdir(parents=True); (dst / "a").mkdir(parents=True)
    (src / "a" / "one.shp").touch()
    (src / "a" / "two.shp").touch()
    (dst / "a" / "one.gpkg").touch()

    assert [p.name for p in discover(src, dst)] == ["two.shp"]

def test_discover_is_deterministic(tmp_path):
    ...
    assert discover(src, dst) == discover(src, dst)

def test_apply_one_rejects_missing_crs(tmp_path, crsless_shapefile):
    with pytest.raises(ValueError, match="no CRS"):
        apply_one(crsless_shapefile, tmp_path, tmp_path)

def test_run_records_failure_without_stopping(tmp_path, one_good_one_bad):
    results = run(*one_good_one_bad)
    assert Counter(r.status for r in results) == {"ok": 1, "failed": 1}

None of these tests need a real dataset or a real folder of hundreds of files. That is the return on separating the stages.

Example 3: resume from the last report #

def discover_resumable(src, dst, last_report: Path | None = None):
    items = discover(src, dst)
    if last_report and last_report.exists():
        failed = {f["item"] for f in json.loads(last_report.read_text())["failures"]}
        # retry previous failures even if an output file exists
        items = sorted(set(items) | {Path(f) for f in failed})
    return items

Because discovery is a pure function of the filesystem plus an optional report, resuming is a change to one function rather than a flag threaded through the whole script.

Explanation #

The reason the four-line loop degrades is coupling. In that loop, discovery (glob), transformation (read/to_crs/to_file) and control flow all live in the same block, so any new requirement must be expressed as a change inside it. Add error handling and the transformation is now nested in a try. Add resumability and there is an if before the read. Add parallelism and the whole body has to be extracted into a function anyway — which is the refactor this structure does up front.

Separating the stages is not architecture for its own sake. It is a bet that the transformation is the stable part and the batch machinery is the part that changes, and that bet is almost always right. The GIS operation — reproject, clip, convert — is specified once and rarely changes. Everything around it changes constantly, because that is where operational reality lives: files arriving in subfolders, a supplier sending one corrupt file a week, an overnight window that shrank, a manager who wants a report.

There is a second, quieter benefit. The Result object turns the run from a process into a record. A batch job that returns list[Result] can be diffed against yesterday's, checked in a test, and used to answer "did file X get processed on the 14th?" — questions the print-as-you-go version cannot answer at all. That is the difference between a script and a job you can operate. See how to record run metadata and data lineage.

Edge cases or notes #

• Discovery on a network share is slow. rglob over 200,000 files takes minutes. Cache the work list, or list once and pass it in.
• An output file existing does not mean it is complete. A crashed run leaves a truncated file that discovery will skip. Write to a temp name and rename on success — rename is atomic on the same filesystem.
• Result must be picklable for parallel runs. Dataclasses of primitives are; a GeoDataFrame inside a Result is not worth sending back.
• One item per file is a convention, not a rule. An item can be a layer in a GeoPackage, a database partition, or a tile id — the shape is the same.
• Sorting by size can help. Processing the largest items first gives a better worst-case wall time in a parallel run.
• Do not let apply_one open shared handles. A database connection or an open output file breaks both testability and parallelism.
• Empty discovery is a success, not an error — but it should be visible in the report, because "0 items" usually means the pattern is wrong.

Internal links #

• How to batch process a folder of GIS files in Python: the complete workflow — this shape, fully worked
• Failure policy in batch processing: fail fast, skip or quarantine — the fourth element in depth
• How to build a resumable batch GIS job in Python — discovery as the resume mechanism
• How to log and summarise errors in a batch GIS job — the report stage
• How to add a dry-run mode to a batch GIS script — why discovery must be side-effect free
• How to speed up batch GIS jobs with parallel processing — what apply_one being pure buys you
• What is a GIS data pipeline? The anatomy explained — the same thinking one level up
• Batch script stops at the first bad file — the symptom that sends people here

FAQ #

Is this over-engineering for a ten-file job? #

The skeleton is about forty lines and it is the same forty whether there are ten files or ten thousand. The cost is paid once; the four-line loop charges interest every time a requirement arrives.

Why must discovery be side-effect free? #

Because a dry run is just "call discovery and print the result". If discovery creates output directories or moves files, you cannot ask what the job would do without it doing some of it.

Why should apply_one not know the item count? #

Anything it does with that knowledge — progress bars, running totals — is reporting, and belongs in the report stage. Keeping it out is what allows the function to be tested and parallelised unchanged.

Where does a progress bar go? #

In run, wrapping the loop or the pool. apply_one returns a Result; the caller counts them. See how to add a progress bar and ETA.

What if items depend on each other? #

Then it is not a batch job, it is a pipeline — an ordered sequence of steps over one dataset rather than one step over many items. See how to chain GIS processing steps.

Should the report be written even when everything succeeds? #

Yes. A report that only appears on failure cannot tell you that last Tuesday's run processed 4,000 files instead of the usual 4,200.

How does this relate to Airflow or Luigi? #

Those tools implement this shape at the orchestration level, with items as tasks. Understanding the shape first makes them easier to use — and makes it clear when a sixty-line script is enough. See choosing a scheduler for GIS jobs.