Automating Attribute Schema Migration with Fiona

A focused, driver-level recipe for rewriting a vector layer so its attribute table conforms to a declared target schema — the executable core of Schema Drift Reconciliation for Spatial Layers.

Concept & Context

Once schema drift reconciliation has decided what the reconciled schema should be, something has to physically rewrite each branch layer to match it. That rewrite is a schema migration: a deterministic transform that reads records under one schema and writes them under another, adding new columns, dropping retired ones, applying declared renames, and coercing types within safe bounds. This page implements that transform with Fiona, which is the right tool precisely because it does not hide the driver’s field definitions — you see and control int:9, float:24.15, and str:80 directly, which matters when the destination is a format-sensitive container like Shapefile.

Working at the Fiona layer keeps memory bounded: features stream one at a time from reader to writer, so a multi-gigabyte layer migrates in constant memory rather than materializing a full GeoDataFrame. It also keeps the operation honest about widths and precision, which is where value-level attribute reconciliation for tabular spatial data and higher-level DataFrame tooling tend to lose fidelity. Geometry is passed through untouched here; when the geometry type itself has drifted, that is a separate concern handled by reconciling geometry type changes across branches.

Core Migration Pipeline

The migration runs as five ordered steps, each with a strict contract:

  1. Read both schemas. Open the source layer and a target-schema reference (another layer, or a schema dict from the reconciliation step) and extract geometry, properties, and crs.
  2. Compute the migration plan. For every target field, decide its origin: kept as-is, renamed from a source field, coerced from a source field of a different type, or added new. Every source field not consumed is a drop.
  3. Open the writer. Create the output with the target schema, the source CRS, and the chosen driver, so the container’s field widths are fixed before any record is written.
  4. Transform each record. Stream source features; for each, build a new properties dict from the plan — rename, fill, cast, drop — and write it with its geometry unchanged.
  5. Verify convergence. Re-open the output, assert its schema equals the target, and confirm the feature count matches the source.

Working Implementation

"""Migrate a vector layer's attribute schema to a declared target schema using Fiona.

Streams records source -> target in constant memory, applying a field-level plan:
add (typed null), drop, rename, and widening-only coercion with range checks.
"""
from __future__ import annotations
from dataclasses import dataclass
import fiona


# ── Widening-only coercion table: (from_base, to_base) -> caster ──────────────
def _to_float(v):  return None if v is None else float(v)
def _to_str(v):    return None if v is None else str(v)
def _int_check(v):
    if v is None:
        return None
    if abs(v - round(v)) > 1e-9:              # float carrying a fraction
        raise ValueError(f"non-integral value {v!r} cannot widen to int losslessly")
    return int(round(v))

WIDENERS = {
    ("int", "float"):  _to_float,
    ("int", "str"):    _to_str,
    ("float", "str"):  _to_str,
    ("bool", "int"):   lambda v: None if v is None else int(v),
    ("bool", "str"):   _to_str,
    ("date", "str"):   _to_str,
    ("datetime", "str"): _to_str,
}


def _base(type_token: str) -> str:
    """'float:24.15' -> 'float'; normalize driver spellings to a base type."""
    b = type_token.split(":")[0].strip().lower()
    return {"integer": "int", "integer64": "int", "string": "str"}.get(b, b)


def _width(type_token: str) -> str:
    return type_token.split(":")[1].strip() if ":" in type_token else ""


@dataclass
class FieldOp:
    target_field: str
    source_field: str | None      # None => added field
    op: str                       # "keep" | "rename" | "coerce" | "add"
    caster: callable | None       # set for "coerce"
    target_type: str              # full target type token, e.g. "str:80"


def build_plan(source_schema: dict, target_schema: dict,
               renames: dict[str, str] | None = None) -> list[FieldOp]:
    """renames maps SOURCE field name -> TARGET field name (declared, not inferred)."""
    renames = renames or {}
    src_props = {k.lower(): v for k, v in source_schema["properties"].items()}
    tgt_props = {k.lower(): v for k, v in target_schema["properties"].items()}
    # Invert the rename map: target name -> source name
    tgt_from_src = {tgt: src for src, tgt in renames.items()}

    plan: list[FieldOp] = []
    for tgt_name, tgt_type in tgt_props.items():
        origin = tgt_from_src.get(tgt_name, tgt_name)   # renamed or same-named
        if origin in src_props:
            s_base, t_base = _base(src_props[origin]), _base(tgt_type)
            if s_base == t_base:
                op = "rename" if origin != tgt_name else "keep"
                plan.append(FieldOp(tgt_name, origin, op, None, tgt_type))
            elif (s_base, t_base) in WIDENERS:
                plan.append(FieldOp(tgt_name, origin, "coerce",
                                    WIDENERS[(s_base, t_base)], tgt_type))
            else:
                raise ValueError(
                    f"unsupported (narrowing) cast {s_base}->{t_base} for '{tgt_name}'")
        else:
            plan.append(FieldOp(tgt_name, None, "add", None, tgt_type))   # typed null
    return plan


def _fits(value, type_token: str) -> None:
    """Range/length guard against the target field definition."""
    if value is None:
        return
    base, width = _base(type_token), _width(type_token)
    if base == "str" and width and len(str(value)) > int(width):
        raise ValueError(f"string '{value}' exceeds width {width}")
    if base == "int" and width and abs(int(value)) >= 10 ** int(width):
        raise ValueError(f"integer {value} exceeds {width}-digit field width")


def migrate_layer(src_path: str, out_path: str, target_schema: dict,
                  renames: dict[str, str] | None = None,
                  driver: str = "GPKG") -> int:
    with fiona.open(src_path) as src:
        plan = build_plan(src.schema, target_schema, renames)
        # Writer uses the TARGET schema so field widths are fixed up front.
        out_schema = {"geometry": target_schema["geometry"],
                      "properties": dict(target_schema["properties"])}
        written = 0
        with fiona.open(out_path, "w", driver=driver,
                        crs=src.crs, schema=out_schema) as dst:
            for feat in src:
                props_in = {k.lower(): v for k, v in feat["properties"].items()}
                props_out: dict[str, object] = {}
                for fop in plan:
                    if fop.op == "add":
                        props_out[fop.target_field] = None          # typed null
                        continue
                    raw = props_in.get(fop.source_field)
                    val = fop.caster(raw) if fop.op == "coerce" else raw
                    _fits(val, fop.target_type)
                    props_out[fop.target_field] = val
                dst.write({"type": "Feature",
                           "geometry": feat["geometry"],      # geometry passthrough
                           "properties": props_out})
                written += 1
    return written


if __name__ == "__main__":
    # Target schema comes from the reconciliation step (or a reference layer).
    target = {
        "geometry": "MultiPolygon",
        "properties": {
            "parcel_id":  "int:9",
            "land_use":   "str:40",      # renamed from source 'landuse'
            "population": "float:24.15", # widened from source int
            "risk_band":  "str:12",      # newly added field
        },
    }
    n = migrate_layer(
        "branches/source/parcels.gpkg",
        "migrated/parcels.gpkg",
        target_schema=target,
        renames={"landuse": "land_use"},
        driver="GPKG",
    )
    print(f"migrated {n} features")

Validation & Output Verification

Migration is only trustworthy if the output schema and record count are proven, not assumed. Re-open the written layer and check it against the target.

def verify_migration(out_path: str, target_schema: dict, source_count: int) -> None:
    with fiona.open(out_path) as out:
        got = {k.lower(): _base(v) for k, v in out.schema["properties"].items()}
        want = {k.lower(): _base(v) for k, v in target_schema["properties"].items()}
        assert got == want, f"schema mismatch: {got} != {want}"
        assert out.schema["geometry"] == target_schema["geometry"], "geometry token drift"
        assert len(out) == source_count, f"count {len(out)} != source {source_count}"
    print("migration verified")

Run a spot check on a coerced column to confirm values survived the cast — for a widened population field, assert that a known source integer now reads back as the equal float. For a schema-registry workflow, fingerprint the output and compare it to the reconciled fingerprint recorded upstream; a match closes the loop that the parent schema drift reconciliation gate depends on.

Failure Modes

  • schema mismatch on verification — the source contained a field not covered by the plan, or a target field spelled differently in case. Root cause: an undeclared rename or mixed-case field names. Fix: add the mapping to renames and lower-case all names before building the plan.

  • string exceeds width mid-stream — a value is longer than the target str:N definition, common when migrating into Shapefile. Root cause: the target width was sized from the schema, not the data. Fix: widen the target field to the observed max length, or move the destination to GeoPackage, which has no fixed string cap.

  • unsupported (narrowing) cast — the plan requested float→int or int64→int32. Root cause: the target schema narrowed a field that must only widen. Fix: escalate as a breaking change to the reconciliation step; never migrate a narrowing automatically.

  • Added column reads back all-null but the field is missing — the driver dropped an unwritten field. Root cause: a None written without the field being in the writer schema. Fix: ensure every added field is present in out_schema so the typed null is materialized as a real column.