Scheduled Batch Processing and Pipeline Orchestration with PyQGIS

A script that works when you run it is not the same thing as a job that runs unattended at three in the morning. The difference is entirely in the parts that have nothing to do with geoprocessing: what happens when the input is late, when two runs overlap, when the disk fills halfway through a write, and how anybody finds out. This page, part of the Headless Automation, CI/CD & Testing guide, covers structuring a repeatable job, choosing a scheduler, making publication atomic, handling partial failure, and making a run observable enough to debug from its log alone.

The bootstrap itself — prefix path, application object, Processing registry — is covered in standalone PyQGIS scripts and headless execution; this page assumes you have that working and asks what has to surround it.

Prerequisites Checklist #

• A working standalone script. If it needs a display or a hand-set environment variable, fix that before scheduling it.
• QGIS 3.28 LTR or newer, ideally inside a container so the version is pinned — see containerizing QGIS with Docker.
• Python 3.9+ for the type hints below.
• A writable working directory that is not the output directory. Atomic publication needs both, on the same filesystem.
• Somewhere failures are actually seen — a mailbox, a chat channel, an alerting system. A job whose failures go to a log nobody reads is an unmonitored job.

What a Scheduled Job Has to Do Beyond Processing #

Every robust batch job has the same five stages, and only one of them is the geoprocessing.

The stages exist because each one has a distinct failure mode and a distinct correct response. Acquisition fails when an upstream system is late — usually worth a retry. Validation fails when the data arrived but is wrong — never worth a retry, and never worth publishing. Transformation fails on genuine bugs. Publication must be all-or-nothing. Reporting is what turns any of the above into something a person learns about.

Step-by-Step Implementation #

Step 1 — Make the job a function with an exit code #

A scheduled job’s interface to the outside world is its exit code and its output. Structure the script so both are deliberate rather than incidental.

import logging
import sys

log = logging.getLogger("nightly")

EXIT_OK = 0
EXIT_NOTHING_TO_DO = 0        # not a failure: no new input is a normal outcome
EXIT_BAD_INPUT = 2
EXIT_FAILED = 1


def main(argv: list[str]) -> int:
    """Run one batch. Returns the process exit code."""
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s %(levelname)-7s %(message)s",
    )
    try:
        produced = run_batch()
    except FileNotFoundError as exc:
        log.error("input missing: %s", exc)
        return EXIT_BAD_INPUT
    except Exception:
        log.exception("run failed")
        return EXIT_FAILED

    if produced == 0:
        log.info("no new input; nothing to do")
        return EXIT_NOTHING_TO_DO
    log.info("published %d output(s)", produced)
    return EXIT_OK


if __name__ == "__main__":
    sys.exit(main(sys.argv[1:]))

Distinguishing “nothing to do” from “failed” matters more than it looks. A job that exits non-zero on an empty input directory generates an alert every weekend, and an alert that fires routinely stops being read.

Step 2 — Refuse to run twice at once #

Two overlapping runs writing the same output is the classic overnight failure: the second run starts before the first finishes, both write, and the result is neither. A lock file with an exclusive flock is enough on a single machine.

import fcntl
import os
from contextlib import contextmanager


@contextmanager
def single_instance(lock_path: str):
    """Ensure only one copy of the job runs. Exits quietly if another holds the lock."""
    handle = open(lock_path, "w")
    try:
        fcntl.flock(handle, fcntl.LOCK_EX | fcntl.LOCK_NB)
    except BlockingIOError:
        handle.close()
        raise SystemExit("another run is already in progress")
    try:
        handle.write(str(os.getpid()))
        handle.flush()
        yield
    finally:
        fcntl.flock(handle, fcntl.LOCK_UN)
        handle.close()

The lock is released by the operating system when the process dies, which is exactly what you want: a job killed by an out-of-memory event does not leave a stale lock that blocks every subsequent run.

Step 3 — Validate before transforming #

Validation is where a pipeline earns its reliability. Check what you depend on — the file exists, the layer opens, the fields are present, the row count is within an order of magnitude of yesterday’s — and fail before doing any work rather than after.

from qgis.core import QgsVectorLayer


def validated_layer(path: str, required_fields: set[str], min_rows: int) -> QgsVectorLayer:
    """Open `path` and assert the properties the rest of the pipeline assumes."""
    layer = QgsVectorLayer(path, "input", "ogr")
    if not layer.isValid():
        raise ValueError("cannot open %s" % path)

    present = {f.name() for f in layer.fields()}
    missing = required_fields - present
    if missing:
        raise ValueError("%s is missing fields: %s" % (path, ", ".join(sorted(missing))))

    count = layer.featureCount()
    if count < min_rows:
        raise ValueError("%s has %d rows, expected at least %d" % (path, count, min_rows))
    return layer

The row-count floor catches the failure that no schema check can: a technically valid file that arrived truncated. It is the single highest-value assertion in most pipelines.

Step 4 — Run the transformation with feedback #

Processing algorithms report progress and messages through a feedback object. In a scheduled job, route those into the same logger as everything else so the run reads as one narrative.

import logging
from qgis.core import QgsProcessingFeedback

log = logging.getLogger("nightly")


class LoggingFeedback(QgsProcessingFeedback):
    """Forwards Processing's own messages into the job log."""

    def pushInfo(self, info: str) -> None:
        log.info("%s", info)

    def pushWarning(self, warning: str) -> None:
        log.warning("%s", warning)

    def reportError(self, error: str, fatalError: bool = False) -> None:
        log.error("%s", error)

    def setProgressText(self, text: str) -> None:
        log.info("%s", text)

Step 5 — Publish atomically #

Never write directly to the path readers use. Write beside it, verify, then rename — a rename within a filesystem is atomic, so a reader sees either the old file or the new one and never a half-written one.

import os
from qgis.core import QgsVectorLayer


def publish(temp_path: str, live_path: str, min_rows: int = 1) -> None:
    """Verify `temp_path` and move it into place atomically."""
    check = QgsVectorLayer(temp_path, "verify", "ogr")
    if not check.isValid() or check.featureCount() < min_rows:
        raise RuntimeError("refusing to publish %s: output failed verification" % temp_path)
    del check                       # release the provider handle before renaming

    os.replace(temp_path, live_path)     # atomic within one filesystem

Releasing the layer before the rename is not optional on Windows, where an open handle blocks the operation outright.

Choosing a Scheduler #

The scheduler’s job is narrow — start the process on time and notice that it finished — but the four common answers differ sharply in what they give you around that, and in what you have to operate to get it.

Cron is the right answer more often than its reputation suggests: for a single job on a single machine, it is one line and nothing to maintain. Its weaknesses are real, though — no retries, no concurrency control, and output that goes to local mail by default — and the first two are exactly what the lock file and the exit codes above are compensating for.

A systemd timer fixes the logging and gives you Restart= and OnFailure= hooks for free. A CI schedule is worth considering when the job already runs in a container and the pipeline is already defined — see continuous integration for QGIS projects. A workflow orchestrator earns its operational cost only when there are enough interdependent jobs that a dependency graph is genuinely useful.

Advanced Patterns #

Process only what changed #

A nightly job that reprocesses everything takes longer every night. Record what each run consumed — a modification time, a source checksum, a high-water mark — and skip inputs that have not moved.

import json
import os


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


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

Write the state only after a successful publish. Recording progress before the work is confirmed is how a pipeline skips the day it failed on.

Bound the blast radius of a partial failure #

In a job processing many independent inputs, one bad file should not lose the other ninety-nine. Process each independently, collect the failures, and decide at the end whether the run as a whole succeeded.

import logging

log = logging.getLogger("nightly")


def process_all(paths: list[str], tolerance: float = 0.1) -> int:
    """Process every path, tolerating up to `tolerance` of them failing."""
    failures: list[tuple[str, str]] = []
    produced = 0
    for path in paths:
        try:
            process_one(path)
            produced += 1
        except Exception as exc:
            log.exception("failed on %s", path)
            failures.append((path, str(exc)))

    if failures and len(failures) > tolerance * max(1, len(paths)):
        raise RuntimeError("%d of %d inputs failed" % (len(failures), len(paths)))
    for path, reason in failures:
        log.warning("skipped %s: %s", path, reason)
    return produced

Choosing the tolerance is a business decision, not a technical one — and writing it as an explicit number in the code is far better than the implicit tolerance of zero or one that a naive loop gives you.

Keep the run observable #

Log a single structured summary line at the end of every run: inputs seen, outputs produced, failures, elapsed time. It is what turns a directory of logs into something you can grep for a trend, and it answers “when did this start getting slower” without any additional tooling.

Pitfalls and Debugging #

• Cron’s environment is not your shell’s. No PATH you recognise, no PROJ_LIB, no virtual environment. Set every variable the job needs explicitly in the script or the unit file, and never rely on a login profile having run.

• Relative paths. A cron job’s working directory is not the script’s directory. Resolve every path against os.path.dirname(os.path.abspath(__file__)) or an explicit configuration value.

• Output written in place. A reader that opens the file mid-write gets a truncated dataset. Always write-then-rename.

• Overlapping runs. Without a lock, a job that occasionally takes longer than its interval will eventually run twice at once, and the failure is silent data corruption rather than an error.

• Alerts that fire routinely. Any alert that goes off on a normal day trains everyone to ignore it. Distinguish “nothing to do” from “failed” in the exit code.

• exitQgis() skipped on the error path. A crash during teardown turns a clean failure into a confusing one. Use a context manager so teardown runs whatever happens.

• Unbounded temporary files. Processing writes intermediates to a temporary directory that nothing cleans up in a headless run. Set the temporary folder explicitly and clear it at the start of each run.

Frequently Asked Questions #

Should a scheduled job run in a container? #

Almost always, yes. The value is not isolation but pinning: a container fixes the QGIS, GDAL and PROJ versions together with the datum grids they depend on, so a run in six months produces the same numbers as a run today. An apt upgrade on the host can otherwise change a coordinate transform underneath a job nobody has touched.

The exception is a job that needs direct access to hardware or a filesystem awkward to mount — and even there, the reproducibility argument usually wins.

How do I stop a long job from being killed mid-write? #

Handle SIGTERM and finish the current unit of work rather than the whole batch. Because publication is a rename, a job interrupted between units leaves the previous output intact and the temporary file orphaned — which the next run’s cleanup removes. That is the whole benefit of the write-then-rename pattern: there is no state in which a reader sees something incomplete.

Where should logs go? #

To standard output, and let the scheduler capture them. A job that manages its own log files has to solve rotation, permissions and disk limits — problems systemd’s journal and every CI system have already solved. Under cron, redirect to a file and rotate it with logrotate.

Whatever the destination, log one summary line per run with counts and elapsed time. It is the line you will actually read.

Can I run several Processing algorithms in parallel? #

Not inside one QGIS process — a process holds exactly one QgsApplication, and the Processing registry is not designed for concurrent use. The workable pattern is process-level parallelism: several processes, each with its own application, each handling a disjoint set of inputs.

That also gives you memory isolation, which matters for raster work, where a single job can grow until the operating system intervenes.

How do I test a scheduled job? #

Split it. The geoprocessing is testable with pytest-qgis against small fixtures. The orchestration — locking, validation, exit codes, atomic publish — is ordinary Python with no QGIS in it at all, and should be tested that way, including the failure paths that are hard to trigger in production.

Then run the whole thing once against a copy of real input before scheduling it. Most first-night failures are environmental, and only an end-to-end run finds those.

Conclusion #

The geoprocessing is the easy part of a scheduled job. What makes one dependable is everything around it: a lock so runs cannot overlap, validation that fails before any work is done, atomic publication so readers never see a partial file, exit codes that distinguish an empty day from a broken one, and a log that explains a failure without a rerun. Build those once, in a shape you can reuse, and the next pipeline is mostly a matter of writing the transformation.

Related #

• Headless Automation, CI/CD & Testing for PyQGIS — the parent guide covering the whole delivery pipeline
• Standalone PyQGIS Scripts and Headless Execution — the bootstrap every scheduled job starts from
• Containerizing QGIS with Docker — pinning the runtime a scheduled job depends on
• Continuous Integration for QGIS Projects — running the same job on a schedule in CI
• Testing PyQGIS with pytest-qgis — testing the transformation half of a pipeline