End-to-End Spatial Automation Recipes for SpatiaLite & GeoPackage

A single stage of a spatial pipeline — reading a shapefile, creating a GeoPackage, building an index — is easy to demonstrate and hard to run unattended. The engineering challenge is chaining those stages into one pipeline that ingests raw vector data, produces an OGC-conformant GeoPackage, keeps its spatial index consistent, gates on validation, and stages the result for an offline sync push, all while surviving partial failure and safe re-runs.

This guide sits within the Python Integration & Database Workflows section and serves as the recipes hub: it defines the anatomy shared by every automation pipeline on this site, then hands off to focused walkthroughs for the ETL pass, index maintenance, and write-path selection. Read it first to understand how the individual recipes compose, then follow the linked pages for complete runnable implementations.

Prerequisites #

These recipes assume a working Python spatial stack and familiarity with the lower-level building blocks documented elsewhere in this section. Confirm each item before assembling a full pipeline.

• Python 3.9+ with sqlite3 from the standard library
• GDAL/OGR 3.4+ exposing the GPKG and SQLite drivers (verify with ogrinfo --formats)
• Fiona 1.9+ and/or GeoPandas 0.13+ for the ingest stage
• Shapely 2.0+ for geometry envelope computation and validity checks
• mod_spatialite available if any stage targets a SpatiaLite container rather than GeoPackage
• Familiarity with Native sqlite3 Spatial Extensions for direct index and PRAGMA control
• A writable staging directory and an atomic-rename-capable filesystem for the sync push handoff

Concept & Specification Reference #

A pipeline is only as reliable as the contracts between its stages. The table below fixes the vocabulary used across every recipe in this guide.

Idempotency Model #

Every stage must be safe to run twice. The create stage overwrites or truncates its target rather than appending blindly; the index stage rebuilds from scratch rather than incrementing; the validation stage is naturally read-only; and the staging stage uses an atomic rename so a partially copied file is never visible to the consumer. This lets a supervisor restart the pipeline from the last completed stage after any interruption without producing duplicate features or a half-written index. The transaction scoping and rollback strategies guide covers the write-side primitives that make individual stages atomic.

Error Isolation vs Fail-Fast #

Two failure policies coexist in a healthy pipeline. Record-level and file-level errors during ingest are isolated: the offending item is logged and quarantined, and the run continues. Structural errors — a missing driver, an unwritable staging directory, a failed validation gate — are fail-fast: they abort the run because continuing would produce a corrupt or misleading artifact. Deciding which class an error belongs to is a design decision you make per stage, not a runtime accident.

Step-by-Step Implementation #

This section builds a skeleton orchestrator that wires the five stages together. Each recipe page fleshes out one stage; here the goal is the composition pattern and the shared logging and error-isolation scaffolding.

1. Define a stage contract #

Model every stage as a callable that takes a context object and returns it, so stages compose without knowing about each other. A dataclass carries the artifact path, counters, and a quarantine list between stages.

# Shared pipeline context passed between every stage
from dataclasses import dataclass, field
from pathlib import Path

@dataclass
class PipelineContext:
    source_dir: Path
    artifact: Path                       # the GeoPackage being built
    staging_dir: Path
    features_written: int = 0
    quarantined: list[tuple[str, str]] = field(default_factory=list)
    layer_name: str = "features"
    geom_column: str = "geom"

    def quarantine(self, item: str, reason: str) -> None:
        """Record an isolated failure without aborting the run."""
        self.quarantined.append((item, reason))

2. Add structured logging #

Attach one logger configured for structured output. Each stage logs a start and end record with its name and the running feature count, so a failed unattended run leaves a reconstructable timeline.

import logging

def build_logger(name: str = "spatial_pipeline") -> logging.Logger:
    """Configure a logger that emits one structured line per stage event."""
    logger = logging.getLogger(name)
    if logger.handlers:
        return logger
    logger.setLevel(logging.INFO)
    handler = logging.StreamHandler()
    handler.setFormatter(
        logging.Formatter("%(asctime)s stage=%(name)s level=%(levelname)s %(message)s")
    )
    logger.addHandler(handler)
    return logger

3. Wrap each stage with isolation and timing #

A decorator gives every stage uniform logging and turns unexpected exceptions into a single fail-fast abort with context, while leaving record-level isolation to the stage body.

import functools
import time
from collections.abc import Callable

def stage(name: str) -> Callable:
    """Decorate a stage function with timing and fail-fast error context."""
    def wrapper(fn: Callable[[PipelineContext], PipelineContext]) -> Callable:
        @functools.wraps(fn)
        def inner(ctx: PipelineContext) -> PipelineContext:
            log = build_logger()
            start = time.perf_counter()
            log.info("start", extra={"name": name})
            try:
                ctx = fn(ctx)
            except Exception as exc:  # structural failure -> abort the run
                log.error(f"aborted reason={exc}", extra={"name": name})
                raise
            elapsed = time.perf_counter() - start
            log.info(
                f"done features={ctx.features_written} elapsed={elapsed:.2f}s",
                extra={"name": name},
            )
            return ctx
        return inner
    return wrapper

4. Compose the orchestrator #

The orchestrator threads a single context through the stage sequence. Because each stage returns the context, the pipeline is a left fold over the stage list — trivial to reorder, test, or truncate.

from collections.abc import Sequence

def run_pipeline(
    ctx: PipelineContext,
    stages: Sequence[Callable[[PipelineContext], PipelineContext]],
) -> PipelineContext:
    """Run stages in order, threading one context through each."""
    for stage_fn in stages:
        ctx = stage_fn(ctx)
    return ctx

5. Provide idempotent stage bodies #

The skeleton bodies below show the shape each recipe fills in. The create stage truncates its target; the index and validation stages are covered in depth by the linked recipes. Note that every body is safe to run twice.

import sqlite3

@stage("create")
def create_geopackage(ctx: PipelineContext) -> PipelineContext:
    """Idempotently (re)create the target GeoPackage container."""
    if ctx.artifact.exists():
        ctx.artifact.unlink()          # truncate for a clean, repeatable run
    from osgeo import ogr, osr
    driver = ogr.GetDriverByName("GPKG")
    ds = driver.CreateDataSource(str(ctx.artifact))
    srs = osr.SpatialReference()
    srs.ImportFromEPSG(4326)
    ds.CreateLayer(ctx.layer_name, srs=srs, geom_type=ogr.wkbUnknown,
                   options=[f"GEOMETRY_NAME={ctx.geom_column}"])
    ds.FlushCache()
    ds = None
    return ctx

@stage("stage_sync")
def stage_for_sync(ctx: PipelineContext) -> PipelineContext:
    """Atomically move the validated artifact into the sync push directory."""
    ctx.staging_dir.mkdir(parents=True, exist_ok=True)
    target = ctx.staging_dir / ctx.artifact.name
    # os.replace is atomic on the same filesystem: consumers never see a partial file
    import os
    os.replace(ctx.artifact, target)
    return ctx

Wire the ingest, index, and validation stages from the recipe pages into the stages list and call run_pipeline. The write-path choice for the ingest and create stages — GDAL/OGR versus raw sqlite3 — is a decision covered in its own recipe below, because it changes how much of the metadata and index bookkeeping your code must own.

Validation & Verification #

The validation stage is the pipeline’s safety gate. At minimum it confirms the artifact is a readable GeoPackage with the expected layer, a populated gpkg_contents registration, and a synchronized R-tree. Run these checks before the staging stage ever sees the file.

# Verify structural integrity of the artifact before it advances to sync staging
import sqlite3

def verify_artifact(path: str, layer: str, geom_col: str) -> dict:
    """Return counts proving the layer, its registration, and its index agree."""
    conn = sqlite3.connect(path)
    try:
        features = conn.execute(f"SELECT COUNT(*) FROM {layer}").fetchone()[0]
        registered = conn.execute(
            "SELECT COUNT(*) FROM gpkg_contents WHERE table_name = ?", (layer,)
        ).fetchone()[0]
        indexed = conn.execute(
            f"SELECT COUNT(*) FROM rtree_{layer}_{geom_col}"
        ).fetchone()[0]
    finally:
        conn.close()
    assert registered == 1, f"{layer} missing from gpkg_contents"
    assert features == indexed, f"index out of sync: {features} rows, {indexed} indexed"
    return {"features": features, "indexed": indexed}

A row-count match between the feature table and its rtree_* shadow table is the fastest proxy for a healthy index. For the full OGC conformance ruleset — mandatory tables, trigger presence, SRID resolution — run the queries in how to validate GeoPackage OGC compliance as part of this same gate.

# Complementary command-line gate: confirms GDAL can open every layer cleanly
ogrinfo -al -so staged/survey.gpkg | grep -E "Layer name|Feature Count"

Common Failure Modes & Fixes #

Non-idempotent create leaves stale features #

Symptom: Re-running the pipeline doubles the feature count or raises table already exists.

Cause: The create stage appended to an existing artifact instead of truncating it, so a second run stacked a new copy on top of the first.

Fix: Delete or os.replace the target file at the start of the create stage, as in the skeleton above. Never rely on CREATE TABLE IF NOT EXISTS for the artifact itself when the whole run is meant to be reproducible.

Stale R-tree after a direct bulk load #

Symptom: Spatial filters return too few rows even though SELECT COUNT(*) shows every feature is present.

Cause: The ingest stage inserted geometries with raw sqlite3, bypassing the OGR triggers that maintain the R-tree spatial index, leaving the index empty or partial.

Fix: Add an explicit index-rebuild stage after any trigger-bypassing load. The dedicated recipe below shows the rebuild SQL for both SpatiaLite and GeoPackage and how to detect the staleness automatically.

Validation gate skipped under exception #

Symptom: A malformed GeoPackage reaches the sync queue despite a validation stage being present.

Cause: The validation function caught and swallowed its own assertion errors, or the orchestrator continued past a failed stage instead of aborting.

Fix: Let validation failures propagate as exceptions so the fail-fast decorator aborts the run. Only the ingest stage may isolate errors; downstream structural gates must be fail-fast.

Partial file visible to the sync consumer #

Symptom: A downstream sync process reads a truncated GeoPackage and reports database disk image is malformed.

Cause: The staging stage copied bytes into the push directory in place, so the consumer read the file mid-copy.

Fix: Write to a temporary path on the same filesystem, then os.replace into the final name. Rename within one filesystem is atomic, so the consumer sees either the old file or the complete new one — never a partial. See how to build an offline sync push script for GeoPackage for the consumer side of this handoff.

WAL sidecar files break the artifact handoff #

Symptom: The staged file opens but is missing the most recent features that the pipeline logged as written.

Cause: The artifact was left in WAL mode with un-checkpointed changes still living in the -wal sidecar, and only the main .gpkg file was moved to staging.

Fix: Run PRAGMA wal_checkpoint(TRUNCATE); and close every connection before the staging stage, or keep the artifact in rollback-journal mode until staging. Connection pooling and lifecycle management covers deterministic teardown that guarantees no open handle survives into the staging stage.

Performance Notes #

The dominant cost in these pipelines is the index build, and its timing relative to the bulk load determines total throughput. Building the R-tree incrementally as features are inserted is far slower than loading all geometry first with the index disabled and building it once over sorted data. For any artifact above roughly 100,000 features, split ingest and indexing into separate stages: load with SPATIAL_INDEX=NO, then run the index stage in a single pass. On a 500,000-feature dataset this two-phase ordering typically cuts total build time by 30 to 50 percent.

Wrap the bulk load in a single explicit transaction and set PRAGMA synchronous=NORMAL with journal_mode=WAL for the duration; committing once per feature is the most common cause of an ingest stage that takes minutes instead of seconds. Increase PRAGMA cache_size to cover the working set of geometry pages so the index build reads from memory rather than disk. When the pipeline runs on a constrained field device, favor a streaming ingest that never materializes the full dataset in RAM, as documented in managing large spatial datasets in memory.

Child Pages #

• Shapefile to GeoPackage ETL Pipeline with Validation — a complete Python pipeline that reads shapefiles, creates a GeoPackage, loads features, builds the R-tree, runs the OGC compliance gate, and stages for sync
• How to Automate R-tree Index Rebuilds After Bulk Load — detect and repair a stale spatial index for both SpatiaLite and GeoPackage after a trigger-bypassing insert
• GDAL vs sqlite3 Module for GeoPackage Writes — a decision guide for choosing the write path that owns the right amount of metadata and index bookkeeping for your pipeline

Frequently Asked Questions #

Related #

• Python Integration & Database Workflows — parent section: connection lifecycles, serialization, extension loading, and concurrency control for the full Python spatial SQLite stack
• Fiona & OGR Driver Configuration — driver binding, schema enforcement, and creation options for the ingest and create stages
• Native sqlite3 Spatial Extensions — direct sqlite3 access for index management and PRAGMA tuning inside a stage
• Transaction Scoping & Rollback Strategies — the atomicity primitives that make each stage safe to re-run
• How to Build an Offline Sync Push Script for GeoPackage — the consumer of the staging stage’s validated artifact