SpatiaLite & GeoPackage Automation with Python

Shapefile to GeoPackage ETL Pipeline with Validation

Read every shapefile in a directory with Fiona, stream the features into a single GeoPackage layer, build the R-tree spatial index in one pass, run an…

Read every shapefile in a directory with Fiona, stream the features into a single GeoPackage layer, build the R-tree spatial index in one pass, run an OGC-compliance gate that either passes the file to a sync-staging directory or quarantines it, and log the outcome — all as one re-runnable Python script.

This walkthrough is part of the End-to-End Spatial Automation Recipes guide, which defines the stage-contract pattern this pipeline follows. Here we implement the full ingest-to-staging path end to end with runnable code.

Why This Matters

Field teams rarely receive one clean shapefile. They receive a folder of them — one per survey area, per crew, per week — each a bundle of .shp, .shx, .dbf, and .prj sidecars with inconsistent coordinate reference systems and occasional corruption. A GeoPackage is the destination that makes that pile queryable, portable, and safe to sync to a mobile device, but only if the conversion is validated. An unvalidated pipeline happily produces a file that GDAL wrote but QGIS or a mobile SDK later refuses to open because a mandatory table or trigger is missing.

The pipeline below closes that gap. It does not just convert; it proves the output is conformant before anything downstream can consume it, and it isolates a single bad shapefile instead of letting it abort a 300-file batch.

Prerequisites

  • Python 3.9+ with the standard-library sqlite3 module
  • Fiona 1.9+ (pip install fiona[all] or conda install -c conda-forge fiona gdal)
  • GDAL/OGR 3.4+ with the GPKG and ESRI Shapefile drivers enabled
  • Shapely 2.0+ for geometry envelope computation during index rebuild
  • Source shapefiles with their .shx and .dbf sidecars present
  • Familiarity with the stage-contract pattern in End-to-End Spatial Automation Recipes

Primary Method

The function below is the whole pipeline. It ingests, creates, loads, indexes, validates, and stages, returning a summary that names any quarantined inputs. Each block maps to one stage from the recipe pattern.

python
# Fiona 1.9+ / GDAL 3.4+ — shapefile directory to validated GeoPackage
import os
import sqlite3
from pathlib import Path

import fiona
import fiona.transform

TARGET_CRS = "EPSG:4326"

def shp_dir_to_validated_gpkg(
    src_dir: str,
    out_gpkg: str,
    staging_dir: str,
    layer: str = "features",
    geom_col: str = "geom",
) -> dict:
    """Convert a directory of shapefiles into one validated, sync-staged GeoPackage."""
    src_root = Path(src_dir)
    artifact = Path(out_gpkg)
    quarantined: list[tuple[str, str]] = []

    shapefiles = sorted(src_root.glob("**/*.shp"))
    if not shapefiles:
        raise FileNotFoundError(f"No .shp files under {src_root}")

    # Stage 1+2: ingest reference schema and (re)create the artifact idempotently
    if artifact.exists():
        artifact.unlink()
    with fiona.open(str(shapefiles[0]), driver="ESRI Shapefile") as ref:
        schema = dict(ref.schema)
        if schema["geometry"] in ("Polygon", "LineString", "Point"):
            schema["geometry"] = "Multi" + schema["geometry"]  # accept mixed types

    written = 0
    with fiona.open(
        str(artifact), "w", driver="GPKG", layer=layer,
        schema=schema, crs=TARGET_CRS,
        SPATIAL_INDEX="NO",              # build the index once, after the bulk load
        GEOMETRY_NAME=geom_col,
    ) as dst:
        for shp in shapefiles:           # Stage 3: load with per-file error isolation
            try:
                with fiona.open(str(shp), driver="ESRI Shapefile") as src:
                    for feat in src:
                        geom = fiona.transform.transform_geom(
                            src.crs_wkt, TARGET_CRS, feat["geometry"]
                        )
                        dst.write({**feat, "geometry": geom})
                        written += 1
            except Exception as exc:
                quarantined.append((str(shp), str(exc)))

    build_spatial_index(str(artifact), layer, geom_col)   # Stage 4
    verify_ogc(str(artifact), layer, geom_col)            # Stage 5 gate (raises on fail)
    staged = stage_for_sync(str(artifact), staging_dir)   # Stage 6

    return {"written": written, "quarantined": quarantined, "staged": staged}

The helper functions build_spatial_index, verify_ogc, and stage_for_sync are defined in the walkthrough below. Splitting them out keeps each stage independently testable and lets you reorder or reuse them in other pipelines.

Step-by-Step Walkthrough

1. Establish a reference schema and create the container

The first shapefile sets the geometry type and property schema for the output layer. Promoting single geometry types to their Multi* variant up front avoids a mid-run Record's geometry type does not match collection schema error when a later file mixes Polygon and MultiPolygon. The artifact is deleted first so the run is idempotent — a second invocation produces an identical file, not a doubled one.

python
import fiona

with fiona.open(str(shapefiles[0]), driver="ESRI Shapefile") as ref:
    schema = dict(ref.schema)
    if schema["geometry"] == "Polygon":
        schema["geometry"] = "MultiPolygon"

Creating the layer with SPATIAL_INDEX="NO" is deliberate: maintaining the R-tree incrementally during a bulk load is markedly slower than building it once at the end.

2. Load features with per-file error isolation

The load loop wraps each shapefile in a try/except so one corrupt bundle is quarantined rather than fatal. Every geometry is reprojected to the target CRS before writing, because GeoPackage layers store a single declared CRS and Fiona does not transform geometries on write.

python
import fiona
import fiona.transform

for shp in shapefiles:
    try:
        with fiona.open(str(shp), driver="ESRI Shapefile") as src:
            for feat in src:
                geom = fiona.transform.transform_geom(
                    src.crs_wkt, TARGET_CRS, feat["geometry"]
                )
                dst.write({**feat, "geometry": geom})
    except Exception as exc:
        quarantined.append((str(shp), str(exc)))
        continue

3. Build the R-tree spatial index in one pass

Because the layer was created with the index disabled, build it explicitly after the load. On a GeoPackage the R-tree lives in an rtree_<layer>_<geom> shadow table populated from each geometry’s bounding box.

python
# -- GeoPackage context: populate the R-tree from feature envelopes
import sqlite3

def build_spatial_index(path: str, layer: str, geom_col: str) -> None:
    """Create and fill the GeoPackage R-tree in a single pass after bulk load."""
    conn = sqlite3.connect(path)
    try:
        conn.execute("PRAGMA journal_mode=WAL;")
        conn.execute(
            f"CREATE VIRTUAL TABLE IF NOT EXISTS rtree_{layer}_{geom_col} "
            f"USING rtree(id, minx, maxx, miny, maxy)"
        )
        conn.execute(f"DELETE FROM rtree_{layer}_{geom_col}")
        conn.execute(
            f"INSERT INTO rtree_{layer}_{geom_col} "
            f"SELECT fid, ST_MinX({geom_col}), ST_MaxX({geom_col}), "
            f"ST_MinY({geom_col}), ST_MaxY({geom_col}) "
            f"FROM {layer} WHERE {geom_col} IS NOT NULL"
        )
        conn.commit()
    finally:
        conn.close()

If your load path bypassed OGR triggers or you are unsure whether the index is current, the sibling recipe how to automate R-tree index rebuilds after bulk load covers staleness detection and the equivalent SpatiaLite rebuild.

4. Gate on OGC compliance

This is the stage that separates a real ETL pipeline from a bare converter. The gate confirms the layer is registered in gpkg_contents, its geometry column is registered in gpkg_geometry_columns, and the R-tree row count matches the feature count. Any failure raises, aborting before the file can be staged.

python
# -- GeoPackage context: hard pass/fail compliance gate
import sqlite3

def verify_ogc(path: str, layer: str, geom_col: str) -> None:
    """Raise if the artifact is not a conformant, index-consistent GeoPackage."""
    conn = sqlite3.connect(path)
    try:
        contents = conn.execute(
            "SELECT COUNT(*) FROM gpkg_contents WHERE table_name = ?", (layer,)
        ).fetchone()[0]
        geom_reg = conn.execute(
            "SELECT COUNT(*) FROM gpkg_geometry_columns WHERE table_name = ?",
            (layer,),
        ).fetchone()[0]
        features = conn.execute(f"SELECT COUNT(*) FROM {layer}").fetchone()[0]
        indexed = conn.execute(
            f"SELECT COUNT(*) FROM rtree_{layer}_{geom_col}"
        ).fetchone()[0]
    finally:
        conn.close()
    if contents != 1:
        raise ValueError(f"{layer} not registered in gpkg_contents")
    if geom_reg != 1:
        raise ValueError(f"{layer} missing from gpkg_geometry_columns")
    if features != indexed:
        raise ValueError(f"index out of sync: {features} rows vs {indexed} indexed")

For the complete conformance ruleset, including trigger presence and SRID resolution against gpkg_spatial_ref_sys, run the queries in how to validate GeoPackage OGC compliance inside this same gate.

5. Stage the validated artifact atomically

Only a file that passed the gate reaches this stage. Checkpoint the WAL so no changes linger in the sidecar, then move the file into the staging directory with an atomic rename so a downstream sync process never observes a partial copy.

python
import os
import sqlite3
from pathlib import Path

def stage_for_sync(path: str, staging_dir: str) -> str:
    """Checkpoint the WAL and atomically move the artifact into the push queue."""
    conn = sqlite3.connect(path)
    conn.execute("PRAGMA wal_checkpoint(TRUNCATE);")
    conn.close()
    dest_dir = Path(staging_dir)
    dest_dir.mkdir(parents=True, exist_ok=True)
    target = dest_dir / Path(path).name
    os.replace(path, target)             # atomic within one filesystem
    return str(target)

Verification

Run the pipeline against a sample directory and confirm the staged file reports the expected feature count and a synchronized index. The command-line check confirms GDAL agrees.

python
result = shp_dir_to_validated_gpkg(
    "/data/field_surveys", "/tmp/survey.gpkg", "/data/sync_queue"
)
print(f"written={result['written']} staged={result['staged']}")
for path, reason in result["quarantined"]:
    print(f"QUARANTINED {path}: {reason}")
bash
# Confirm the staged GeoPackage opens cleanly and reports its layers
ogrinfo -al -so /data/sync_queue/survey.gpkg | grep -E "Layer name|Feature Count"

A clean run prints one Layer name and a Feature Count equal to written, with an empty quarantine list.

Alternative Approaches or Edge Cases

GeoPandas for smaller batches. If the entire dataset fits comfortably in memory, geopandas.read_file() followed by GeoDataFrame.to_file(driver="GPKG") is fewer lines. It loses the per-record streaming that keeps Fiona safe on constrained field hardware, so reserve it for datasets below a few hundred thousand features. See converting shapefiles to GeoPackage with GeoPandas.

Multiple layers instead of one merged layer. The pipeline above merges every shapefile into one layer. To keep each source as its own named layer instead, open the destination in append mode per file with a distinct layer= argument, as shown in how to batch convert shapefiles to GeoPackage with Fiona. The validation gate then loops over fiona.listlayers() rather than a single layer name.

Troubleshooting

ValueError: Record's geometry type does not match collection schema

Cause: A source file contains MultiPolygon features but the destination layer was declared as Polygon. GeoPackage layers are strictly typed.

Fix: Promote the schema geometry to its Multi* variant before creating the layer, as done in step 1. This accepts both singular and multi geometries into one layer.

sqlite3.OperationalError: no such function: ST_MinX

Cause: The index-build SQL ran on a plain sqlite3 connection that never loaded a spatial extension, so the envelope functions are undefined.

Fix: Either load mod_spatialite on the connection before building the index, or compute envelopes in Python with Shapely and insert the four bounds directly. The native sqlite3 spatial extensions guide shows the extension-loading sequence.

ValueError: index out of sync from the gate

Cause: Features were written after the index build, or a load error left rows the R-tree never saw.

Fix: Ensure build_spatial_index runs after the last write, and treat the gate failure as intended — the file is correctly quarantined. Re-run the pipeline; because it is idempotent, the rebuild restores agreement.