PyQGIS Engineering & Automation

Processing Only Changed Inputs with a State File

Make a scheduled PyQGIS job incremental: fingerprint each input, keep a small JSON state file, force a full rebuild when the processing logic changes, write…

TL;DR: record a fingerprint per input after each successful run, compare against it on the next run to process only what changed, include a version of your own processing logic in the state so a code change forces a full rebuild, and write the state only after the output has been published. This page is part of the scheduled batch processing and pipeline orchestration guide.

Complete Runnable Code

python
"""Incremental batch processing driven by a small JSON state file."""
import hashlib
import json
import os

LOGIC_VERSION = 4            # bump when the processing changes in a way that invalidates output


def load_state(path: str) -> dict:
    """Read the previous run's state, tolerating a first run and a corrupt file."""
    try:
        with open(path, encoding="utf-8") as handle:
            state = json.load(handle)
    except (FileNotFoundError, json.JSONDecodeError):
        return {"logic_version": LOGIC_VERSION, "inputs": {}}

    if state.get("logic_version") != LOGIC_VERSION:
        return {"logic_version": LOGIC_VERSION, "inputs": {}}      # force a full rebuild
    return state


def save_state(path: str, state: dict) -> None:
    """Write the state atomically so a crash cannot leave it half-written."""
    temp = path + ".tmp"
    with open(temp, "w", encoding="utf-8") as handle:
        json.dump(state, handle, indent=2, sort_keys=True)
    os.replace(temp, path)


def fingerprint(path: str, use_hash: bool = False) -> str:
    """Cheap identity for an input file: size and mtime, or a content hash."""
    stat = os.stat(path)
    if not use_hash:
        return "%d:%d" % (stat.st_size, int(stat.st_mtime))

    digest = hashlib.sha256()
    with open(path, "rb") as handle:
        for chunk in iter(lambda: handle.read(1 << 20), b""):
            digest.update(chunk)
    return digest.hexdigest()


def changed_inputs(paths: list[str], state: dict, use_hash: bool = False) -> list[str]:
    """Return the inputs whose fingerprint differs from the recorded one."""
    known = state.get("inputs", {})
    return [p for p in paths if fingerprint(p, use_hash) != known.get(os.path.basename(p))]


def record(paths: list[str], state: dict, use_hash: bool = False) -> None:
    """Update the state for inputs that were processed successfully."""
    for path in paths:
        state.setdefault("inputs", {})[os.path.basename(path)] = fingerprint(path, use_hash)
One incremental run, end to end A five-step incremental run: load the previous state, list the current inputs, select those whose fingerprint changed, process only those, and write the new state after a successful publish. load state or start empty list inputs as they are now diff by fingerprint process only the changed save state after publishing then compare if any on success

Architecture Breakdown

The state file is a cache, not a source of truth

Everything in it can be discarded at any time, and the only consequence is a slower run. That property is what makes the pattern safe: a corrupt file, a missing file, or a file from a different version of your code all lead to the same correct behaviour, which is to reprocess everything.

Writing it that way — a try that falls back to an empty state — costs three lines and removes an entire class of failure where a scheduled job dies on start-up because its cache is malformed.

logic_version closes the biggest gap

Input fingerprints tell you whether the input changed. They say nothing about whether your processing changed, and a pipeline that only tracks inputs will happily skip every file after you fix a bug in the transformation.

When to ignore the state and reprocess everything A decision tree naming three situations that invalidate incremental state: the processing logic changed, the output schema changed, and the state file is missing or unreadable. Can this run trust the previous state? logic changed reprocess all bump a code version schema changed reprocess all outputs are stale state missing reprocess all treat as first run

An integer you bump by hand is unglamorous and works. It has to be remembered, which is the honest weakness, so bump it in the same commit as the logic change and treat a forgotten bump as the same class of mistake as a forgotten migration.

Choosing the fingerprint

Size and modification time cost one stat call and catch nearly everything. A content hash costs a full read and catches the rest — a file rewritten with identical content, which should not be reprocessed, and an edit that preserved both size and timestamp, which should.

Choosing what to fingerprint an input by A grid of four fingerprints — modification time, size, a content hash and a source version number — showing what each costs to compute and what it misses. costs misses modification time nothing a touched but unchanged file size nothing an edit of the same length content hash a full read nothing source version one query changes the source does not report

For local files, start with size and mtime and move to hashing only if you observe the failure it misses. For inputs arriving over a network, where timestamps are frequently rewritten by the transfer, a hash is usually worth the read from the beginning.

Ordering: Publish, Then Record

The single most important rule in this pattern is that the state is written after the output is published, never before:

python
def run(inputs: list[str], state_path: str, output_path: str) -> int:
    """Process only what changed, publish, then record. Returns the number processed."""
    state = load_state(state_path)
    todo = changed_inputs(inputs, state)
    if not todo:
        return 0                                   # nothing to do is not a failure

    publish_layer(lambda tmp: transform(todo, tmp), output_path)   # atomic publish
    record(todo, state)                            # only now is it safe to remember
    save_state(state_path, state)
    return len(todo)

Recording first means a failure between the two steps loses the work permanently: the state says the input was handled, and no future run will pick it up. Recording last means a failure causes at worst a repeated run, which is the harmless direction to fail in.

Handling Inputs That Disappear

Files are deleted upstream, and a state file that only ever grows becomes a slow leak of stale entries. Pruning is a single set operation:

python
def prune(paths: list[str], state: dict) -> list[str]:
    """Drop state for inputs that no longer exist; return the names removed."""
    present = {os.path.basename(p) for p in paths}
    gone = [name for name in state.get("inputs", {}) if name not in present]
    for name in gone:
        del state["inputs"][name]
    return gone

Whether a disappearing input should also remove its output is a policy question rather than a technical one, and it deserves an explicit answer in the pipeline rather than an accident. Logging the pruned names makes the decision visible either way.

Production Best Practices

  • Write the state after publishing, never before.
  • Include a logic version and bump it with the code that invalidates the output.
  • Fall back to an empty state on anything unreadable.
  • Write the state atomically, with the same temporary-then-rename pattern as the output.
  • Prune entries for inputs that vanished, and log what was pruned.
  • Log how many inputs were skipped as well as how many were processed; a run that skips everything every night is a bug that otherwise looks like success.

Frequently Asked Questions

Where should the state file live?

Beside the output, not beside the code. It describes the state of a particular output directory, so a pipeline run against a different output location should start from an empty state rather than inheriting one from an unrelated run.

Keeping it out of version control matters for the same reason: it is generated, machine-specific and worthless to anyone else.

Can I use a database instead of a file?

Yes, and for a pipeline with many inputs or several machines it is the better answer — a table keyed on input name with a fingerprint column, updated in the same transaction as the output where possible. The pattern is unchanged; only the storage differs.

A file is right when there is one machine and the state is small, which describes most scheduled geoprocessing jobs.

What if an input changes while the job is reading it?

The fingerprint you recorded is from before the read, so the next run sees a difference and reprocesses — which is correct. The subtler problem is the output built from a half-written input, which fingerprinting cannot catch.

Where an upstream system writes inputs in place, the durable fix belongs there: it should publish atomically too. Where you cannot change it, checking that the fingerprint is unchanged after reading, and discarding the result if it moved, is a reasonable defence.

How do I force a full reprocess?

Delete the state file, or bump logic_version. A --rebuild flag that skips the state load is worth adding for the common case, because deleting a file by hand at three in the morning is not a procedure anyone wants to follow.

Does this work when outputs depend on several inputs together?

Only with care. The pattern as written assumes each input maps to its own output. When an output is derived from many inputs, the unit of state should be the output, fingerprinted by the combination of everything that fed it — a hash of the sorted input fingerprints does this well.

Should the state record failures too?

Recording which inputs failed, and why, turns the state file into a small operational record and avoids retrying a permanently broken input on every run. Keep a retry count and a first-seen timestamp so a transient failure is retried and a persistent one is escalated rather than silently retried nightly forever.