Table of Contents

Automated Spatial QA/QC Pipelines: Detecting and Repairing Invalid Geometries with Python

1. Introduction

Automated Spatial QA/QC Pipelines Python

When building Automated Spatial QA/QC Pipelines Python workflows, spatial datasets often look correct on a map while containing geometric defects. Spatial datasets often look correct on a map while containing geometric defects that compromise spatial analysis, database imports, and automated GIS workflows. A polygon may cross its own boundary, a boundary ring may be malformed, or overlapping features may create unintended slivers. Duplicate vertices, empty geometries, inconsistent geometry types, and invalid polygon holes can also undermine downstream processing.

These problems become particularly costly in automated workflows. An invalid parcel boundary can cause a spatial overlay to fail; a malformed administrative polygon can produce incorrect containment results; and inconsistent geometries can interrupt bulk imports into PostGIS or other spatial databases. When a pipeline processes thousands of features, manually inspecting every record is neither efficient nor reproducible.

Automated spatial QA/QC (quality assurance and quality control) addresses this problem by systematically validating geometries, recording the reasons for failure, applying controlled repairs, and exporting only features that meet defined acceptance criteria.

Python provides a practical toolkit for implementing this process:

GeoPandas manages vector datasets as geospatial data frames, supports spatial file I/O, and integrates with pandas.
Shapely performs geometric operations, validates topology, explains invalidity, and repairs many invalid geometries.
Fiona provides an interface to vector data formats supported by GDAL, making it useful for understanding and troubleshooting data ingestion.

A robust pipeline should do more than call a repair function. It should preserve the original data, log every detected problem, distinguish repaired geometries from rejected features, verify the repair outcome, and maintain an auditable record of what changed.

This tutorial builds a reusable Python workflow that reads a raw vector dataset, detects invalid geometries, diagnoses their failure reasons, attempts repairs, screens suspicious polygon fragments, and exports the resulting datasets to GeoPackage or PostGIS.

The central principle is simple: a geometry that can be repaired is not automatically a geometry that should be accepted. Repair must be followed by validation and application-specific quality checks.

2. Prerequisites: Python Libraries and Environment

Use Python 3.10 or newer in a virtual environment. The examples below target Shapely 2.x, which provides the shapely.make_valid() and shapely.is_valid_reason() functions used throughout the pipeline.

Install the required libraries

# Create and activate a virtual environment on Windows.
python -m venv .venv
.venv\Scripts\activate

# Install the spatial data processing libraries.
python -m pip install --upgrade pip
python -m pip install geopandas shapely fiona pyproj pandas

On Linux or macOS, activate the environment with:

source .venv/bin/activate

GeoPandas uses a vector-data I/O engine such as Fiona or Pyogrio to read and write supported formats. Fiona is included explicitly here because it is useful when you need to control or troubleshoot the underlying vector file reader. It handles data access; Shapely handles geometric operations. See the GeoPandas I/O documentation and Fiona documentation.

Verify the installed libraries:

import geopandas as gpdimport shapelyimport fionaprint("GeoPandas:", gpd.__version__)print("Shapely:", shapely.__version__)print("Fiona:", fiona.__version__)

The tutorial focuses on polygon and multipolygon datasets, such as land parcels, building footprints, land-use zones, and administrative boundaries. The repair workflow deliberately preserves polygonal output rather than allowing a repaired polygon to silently become a line or point feature.

Step 1: Load raw vector data and run topology checks

Suppose the source dataset is raw_parcels.gpkg, containing a polygon layer named parcels. Start by reading the layer and checking its coordinate reference system, geometry types, empty features, and validity.

import geopandas as gpdimport shapelyINPUT_FILE = "raw_parcels.gpkg"INPUT_LAYER = "parcels"# Use Fiona as the vector-data reading engine.try:    gdf = gpd.read_file(        INPUT_FILE,        layer=INPUT_LAYER,        engine="fiona"    )except Exception as exc:    raise RuntimeError(        f"Could not read {INPUT_FILE}, layer {INPUT_LAYER!r}"    ) from exc# A missing CRS prevents reliable interpretation of coordinates.if gdf.crs is None:    raise ValueError(        "Source data has no CRS. Establish the correct CRS "        "before performing spatial QA/QC."    )print("Features:", len(gdf))print("CRS:", gdf.crs)print("Geometry types:")print(gdf.geom_type.value_counts(dropna=False))# Distinguish missing geometry from empty geometry.missing_mask = gdf.geometry.isna()empty_mask = gdf.geometry.is_empty# Topological validity is a separate test.valid_mask = gdf.geometry.is_validprint("Missing geometries:", missing_mask.sum())print("Empty geometries:", empty_mask.sum())

geometry.is_valid evaluates whether a geometry satisfies the geometry engine’s validity rules. It does not certify that a feature is geographically accurate, correctly classified, free of duplicates, or consistent with neighboring polygons.

That distinction matters. A perfectly valid parcel may be in the wrong county, while two valid administrative polygons may overlap when they should form a non-overlapping coverage.

For broader data cleaning, track at least four categories:

QA category	Example check	Why it matters
Structural integrity	Missing or empty geometry	Prevents unusable spatial records
Topological validity	Self-intersections, invalid holes	Prevents invalid geometric operations
Schema integrity	Unexpected geometry type	Prevents incompatible database loads
Spatial consistency	Overlaps, gaps, duplicate features	Detects dataset-level errors

Step 2: Identify geometry errors and log their types

A Boolean validity flag tells you which records need attention, but it does not explain the problem. Shapely’s is_valid_reason() function returns a diagnostic string for invalid geometries. The alternative shapely.validation.explain_validity() serves a similar purpose. See the Shapely geometry validation documentation.

Create a diagnostic report before modifying the dataset:

import pandas as pdimport shapely# Preserve a source identifier for tracing features back to the input.gdf = gdf.copy()gdf["_qa_source_id"] = gdf.index.astype(str)# Identify invalid, missing, or empty geometries.invalid_mask = ~gdf.geometry.is_validproblem_mask = (    gdf.geometry.isna()    | gdf.geometry.is_empty.fillna(False)    | invalid_mask)problems = gdf.loc[problem_mask].copy()diagnostics = []for _, row in problems.iterrows():    geom = row.geometry    if geom is None:        reason = "Missing geometry"    elif geom.is_empty:        reason = "Empty geometry"    elif not geom.is_valid:        reason = shapely.is_valid_reason(geom)    else:        reason = "Other QA issue"    diagnostics.append({        "source_id": row["_qa_source_id"],        "geometry_type": (            None if geom is None else geom.geom_type        ),        "valid": (            False if geom is None else geom.is_valid        ),        "error_reason": reason,    })diagnostic_df = pd.DataFrame(diagnostics)diagnostic_df.to_csv("geometry_errors.csv", index=False)print(diagnostic_df.head(20))

Typical diagnostic messages include:

Self-intersection — a polygon boundary crosses itself.
Ring Self-intersection — a ring intersects itself, possibly creating an invalid hole or shell.
Hole lies outside shell — an interior ring lies outside the exterior polygon.
Too few points — a ring or geometry does not have enough points to define the expected shape.

The precise wording depends on the geometry engine and version. Treat the reported text as a diagnostic, not as a permanent error-code standard.

What about duplicate vertices and unclosed rings? These need special attention. Repeated consecutive coordinates do not necessarily make a polygon invalid. Likewise, some file readers or geometry constructors may automatically close polygon rings during ingestion, while malformed source files can fail before is_valid is reached. A validity check is therefore only one component of a complete spatial QA/QC pipeline.

For a separate duplicate-vertex cleanup pass, Shapely provides remove_repeated_points(). Use it cautiously: coordinate changes can affect topology, and any modified geometry should be validated again. Do not remove vertices merely because they are duplicated somewhere in a ring; repeated coordinates may have a legitimate geometric role.

Step 3: Repair invalid geometries

Shapely provides two common approaches to repairing invalid polygons: make_valid() and the traditional buffer(0) technique.

Preferred method: `shapely.make_valid()`

import shapelyinvalid_geom = gdf.loc[    ~gdf.geometry.is_valid].geometry.iloc[0]# Repair an invalid polygon using Shapely.repaired_geom = shapely.make_valid(invalid_geom)print("Original type:", invalid_geom.geom_type)print("Original valid:", invalid_geom.is_valid)print("Repaired type:", repaired_geom.geom_type)print("Repaired valid:", repaired_geom.is_valid)print("Repaired empty:", repaired_geom.is_empty)

make_valid() attempts to produce a valid representation of the input geometry. Its result may be a MultiPolygon or a GeometryCollection, rather than a single Polygon. For instance, a self-crossing bow-tie polygon may be split into multiple polygon components. In other cases, repair can return polygonal and lower-dimensional components together. These are documented behaviors, not necessarily repair failures. See Shapely’s make_valid() documentation.

This creates an important design decision: if your target dataset must contain only polygons, you need to extract and verify the polygonal components while logging any discarded line or point remnants.

Alternative method: `buffer(0)`

# Legacy fallback for certain invalid polygon cases.# This is not guaranteed to preserve the intended geometry.fallback_geom = invalid_geom.buffer(0)print("Fallback type:", fallback_geom.geom_type)print("Fallback valid:", fallback_geom.is_valid)

A zero-distance buffer can clean certain self-crossing or self-touching polygon configurations, but it may produce a different result from make_valid(). Shapely documents this technique as a sometimes-useful cleanup approach rather than a universal repair method. See the Shapely buffer documentation.

For a repeatable data pipeline, use make_valid() first and consider buffer(0) only as a controlled fallback. Log which method succeeded so the result remains auditable.

Step 4: Filter small polygon fragments and export clean data

Repairing geometry validity is not the same as assessing whether the resulting polygon is useful. A repaired polygon can have a negligible area, or a self-intersecting source polygon can become a multipolygon whose parts need additional review.

For this tutorial, we will use an area threshold to identify tiny polygon fragments. A small area is a screening criterion, not proof of a sliver. A small building footprint or land parcel may be legitimate. True sliver detection can also use shape compactness, width, overlap with neighboring features, and the positional accuracy expected from the source data.

Area calculations must use an appropriate projected coordinate reference system with metre units. Do not calculate polygon areas directly from longitude and latitude.

# Example only: use a suitable projected CRS for your study area.AREA_CRS = "EPSG:32637"# A threshold of 1 square metre is illustrative, not universal.MIN_AREA_M2 = 1.0# Reproject a copy for area measurement.projected = gdf.to_crs(AREA_CRS)# Calculate area in square metres.gdf["area_m2"] = projected.geometry.area.to_numpy()# Flag very small polygons for review.sliver_candidates = gdf.loc[    gdf["area_m2"] < MIN_AREA_M2].copy()print("Small polygon candidates:", len(sliver_candidates))

This example demonstrates the area test; in the complete pipeline below, apply it to the repaired geometries instead of the raw data. The example CRS is suitable only for areas where the selected projection is appropriate. Choose a local projected CRS for your own study region.

Do not delete candidates without preserving an audit trail. A production workflow should export accepted geometries, quarantine rejected features, and create a report explaining each decision.

GeoPackage is a convenient delivery format for the cleaned data:

# Export accepted polygons to a GeoPackage.cleaned_gdf.to_file(    "cleaned_parcels.gpkg",    layer="parcels_clean",    driver="GPKG",    index=False)

GeoPackage supports layers in a single file and can store spatial metadata. GeoPandas also supports writing vector data to PostGIS through to_postgis(), which requires SQLAlchemy, GeoAlchemy2, and a PostgreSQL driver. See the GeoPandas file I/O guide and to_postgis() reference.

3. Building an Automated QA/QC Function

The next step is to combine ingestion, diagnostics, repairs, area screening, and export into a reusable function.

The function below implements several safeguards:

Validates input CRS and polygon geometry types.
Records the original validity reason for each feature.
Tries make_valid() before considering buffer(0).
Extracts polygonal components from repaired geometry collections.
Rechecks the validity of repaired geometries.
Rejects geometries that cannot be repaired into valid polygonal output.
Applies an optional area threshold using an appropriate projected CRS.
Exports clean data, a rejection layer where possible, and a CSV audit report.
Uses logging and exception handling to make failures visible.

Complete reusable script

Save the following as spatial_qa_qc.py.

from pathlib import Pathimport loggingimport geopandas as gpdimport pandas as pdimport shapelyfrom pyproj import CRS# ------------------------------------------------------------# 1. Configure logging# ------------------------------------------------------------logging.basicConfig(    level=logging.INFO,    format="%(asctime)s | %(levelname)s | %(message)s")logger = logging.getLogger("spatial_qa_qc")# ------------------------------------------------------------# 2. Extract polygonal components from a geometry# ------------------------------------------------------------def polygonal_only(geom):    """    Return valid polygonal parts as a Polygon or MultiPolygon.    Lower-dimensional components such as LineStrings and Points    are excluded because this pipeline targets polygon datasets.    Returns None if no polygonal component exists.    """    if geom is None or geom.is_empty:        return None    # Preserve ordinary polygonal geometries without alteration.    if geom.geom_type in {"Polygon", "MultiPolygon"}:        return geom    parts = []    def collect_polygons(item):        # Collect individual polygon components.        if item.geom_type == "Polygon":            parts.append(item)        elif item.geom_type == "MultiPolygon":            parts.extend(list(item.geoms))        # Recursively inspect GeometryCollections.        elif item.geom_type == "GeometryCollection":            for subgeom in item.geoms:                collect_polygons(subgeom)    collect_polygons(geom)    if not parts:        return None    if len(parts) == 1:        return parts[0]    # Combine the polygon pieces within this one source feature.    # The caller will verify the result's validity.    return shapely.union_all(parts)# ------------------------------------------------------------# 3. Attempt controlled geometry repair# ------------------------------------------------------------def repair_polygon(geom):    """    Return (repaired_geometry, method, error_message).    The result is accepted only if it is a non-empty, valid    Polygon or MultiPolygon. Otherwise, return None.    """    errors = []    def validate_candidate(candidate):        if candidate is None or candidate.is_empty:            return None        # Repair may return a GeometryCollection. Retain only        # polygonal components for this polygon-only workflow.        candidate = polygonal_only(candidate)        if candidate is None or candidate.is_empty:            return None        if not candidate.is_valid:            return None        if candidate.geom_type not in {"Polygon", "MultiPolygon"}:            return None        return candidate    # Method 1: use the dedicated Shapely repair function.    try:        candidate = shapely.make_valid(geom)        result = validate_candidate(candidate)        if result is not None:            return result, "make_valid", ""        errors.append("make_valid produced no acceptable polygon")    except Exception as exc:        errors.append(f"make_valid failed: {exc}")    # Method 2: try buffer(0) only as a fallback.    # It can alter geometry, so its use is recorded explicitly.    try:        candidate = geom.buffer(0)        result = validate_candidate(candidate)        if result is not None:            return result, "buffer(0)", "; ".join(errors)        errors.append("buffer(0) produced no acceptable polygon")    except Exception as exc:        errors.append(f"buffer(0) failed: {exc}")    return None, "failed", "; ".join(errors)# ------------------------------------------------------------# 4. Run the complete QA/QC workflow# ------------------------------------------------------------def run_spatial_qa_qc(    input_path,    output_dir,    area_crs,    layer=None,    id_field=None,    min_area_m2=None):    """    Clean a polygon vector dataset and save:      - cleaned.gpkg: accepted polygons      - rejected.gpkg: rejected source features, if writable      - qa_audit.csv: per-feature decisions and diagnostics    Parameters    ----------    input_path : str or Path        Input vector dataset path.    output_dir : str or Path        Directory where outputs will be written.    area_crs : str        Suitable projected CRS with metre units, used for        area-based checks.    layer : str, optional        Input layer name, e.g. for a GeoPackage.    id_field : str, optional        Stable input attribute to include in the audit report.    min_area_m2 : float, optional        Minimum accepted polygon area in square metres.        Set to None to disable automatic area filtering.    """    input_path = Path(input_path)    output_dir = Path(output_dir)    output_dir.mkdir(parents=True, exist_ok=True)    if min_area_m2 is not None and min_area_m2 < 0:        raise ValueError("min_area_m2 cannot be negative.")    # Check that the area CRS uses projected metre units.    metric_crs = CRS.from_user_input(area_crs)    if not metric_crs.is_projected:        raise ValueError(            "area_crs must be a projected CRS with metre units."        )    units = [        axis.unit_name.lower()        for axis in metric_crs.axis_info[:2]    ]    if not units or not all(        "metre" in unit or "meter" in unit        for unit in units    ):        raise ValueError(            "area_crs must use metre units for area screening."        )    # Read the source without overwriting it.    try:        raw = gpd.read_file(            input_path,            layer=layer,            engine="fiona"        )    except Exception as exc:        logger.exception("Failed to read input dataset.")        raise RuntimeError(            f"Could not read source dataset: {input_path}"        ) from exc    if raw.crs is None:        raise ValueError(            "Input CRS is missing. Establish the correct CRS first."        )    if raw.empty:        raise ValueError("Input dataset contains no features.")    geom_col = raw.geometry.name    if "_qa_source_row" in raw.columns:        raise ValueError(            "Input already contains reserved column '_qa_source_row'."        )    if id_field is not None and id_field not in raw.columns:        raise ValueError(f"ID field not found: {id_field}")    logger.info("Loaded %s features.", len(raw))    logger.info("Source CRS: %s", raw.crs)    accepted_records = []    rejected_records = []    audit_rows = []    # Enumerating rows creates a unique run-level source reference    # even where the original DataFrame index contains duplicates.    for position, (_, row) in enumerate(raw.iterrows()):        source_row = str(position)        geom = row[geom_col]        original_type = (            None if geom is None            else getattr(geom, "geom_type", type(geom).__name__)        )        original_valid = (            False if geom is None            else bool(                hasattr(geom, "is_valid") and geom.is_valid            )        )        original_reason = "Missing geometry"        if geom is not None and hasattr(geom, "is_empty"):            if geom.is_empty:                original_reason = "Empty geometry"            elif geom.is_valid:                original_reason = "Valid Geometry"            else:                original_reason = shapely.is_valid_reason(geom)        external_id = (            str(row[id_field])            if id_field is not None            else source_row        )        audit = {            "source_row": source_row,            "source_id": external_id,            "original_type": original_type,            "original_valid": original_valid,            "original_reason": original_reason,            "repair_method": "none",            "output_type": None,            "output_valid": False,            "area_m2": None,            "status": None,            "details": ""        }        # Keep a copy of the original row for the rejection layer.        original_record = row.to_dict()        original_record["_qa_source_row"] = source_row        # Reject missing or empty geometries.        if (            geom is None            or not hasattr(geom, "geom_type")            or geom.is_empty        ):            audit["status"] = "rejected_missing_or_empty"            rejected_records.append(original_record)            audit_rows.append(audit)            continue        # This script explicitly targets polygon datasets.        if geom.geom_type not in {"Polygon", "MultiPolygon"}:            audit["status"] = "rejected_non_polygon_input"            audit["details"] = (                "Expected Polygon or MultiPolygon input."            )            rejected_records.append(original_record)            audit_rows.append(audit)            continue        # Preserve valid geometries; repair only invalid ones.        if geom.is_valid:            fixed = geom            method = "unchanged"            repair_error = ""        else:            fixed, method, repair_error = repair_polygon(geom)        audit["repair_method"] = method        audit["details"] = repair_error        # Reject features that cannot be repaired adequately.        if fixed is None or not fixed.is_valid:            audit["status"] = "rejected_repair_failed"            rejected_records.append(original_record)            audit_rows.append(audit)            continue        # Preserve original attributes and store the repaired geometry.        cleaned_record = row.to_dict()        cleaned_record[geom_col] = fixed        cleaned_record["_qa_source_row"] = source_row        accepted_records.append(cleaned_record)        audit["output_type"] = fixed.geom_type        audit["output_valid"] = bool(fixed.is_valid)        audit["status"] = (            "valid_unchanged"            if method == "unchanged"            else "repaired"        )        audit_rows.append(audit)    # Construct output frames using the original geometry column and CRS.    accepted = gpd.GeoDataFrame(        accepted_records,        geometry=geom_col,        crs=raw.crs    )    audit_by_row = {        record["source_row"]: record        for record in audit_rows    }    # --------------------------------------------------------    # Area-based screening after geometry repair    # --------------------------------------------------------    final_records = []    if not accepted.empty:        projected = accepted.to_crs(metric_crs)        areas = projected.geometry.area.to_numpy()        for position, (_, row) in enumerate(accepted.iterrows()):            record = row.to_dict()            source_row = str(record["_qa_source_row"])            area = float(areas[position])            audit = audit_by_row[source_row]            audit["area_m2"] = area            record["qa_area_m2"] = area            # If configured, quarantine undersized polygons.            if (                min_area_m2 is not None                and area < min_area_m2            ):                audit["status"] = "rejected_below_min_area"                audit["details"] = (                    f"Area {area:.6f} m² is below "                    f"threshold {min_area_m2:.6f} m²."                )                rejected_records.append(record)            else:                final_records.append(record)    cleaned = gpd.GeoDataFrame(        final_records,        geometry=geom_col,        crs=raw.crs    )    rejected = gpd.GeoDataFrame(        rejected_records,        geometry=geom_col,        crs=raw.crs    )    # --------------------------------------------------------    # Verify final output validity    # --------------------------------------------------------    if not cleaned.empty:        if (            cleaned.geometry.isna().any()            or cleaned.geometry.is_empty.any()            or not cleaned.geometry.is_valid.all()        ):            raise RuntimeError(                "Final QA failed: invalid or empty geometry remains."            )        # Remove temporary area measurement from the final data;        # the audit CSV retains the calculated area.        if "qa_area_m2" in cleaned.columns:            cleaned = cleaned.drop(columns="qa_area_m2")        clean_path = output_dir / "cleaned.gpkg"        try:            cleaned.to_file(                clean_path,                layer="cleaned",                driver="GPKG",                index=False            )            logger.info("Clean dataset written to %s", clean_path)        except Exception as exc:            logger.exception("Could not write clean GeoPackage.")            raise RuntimeError(                f"Failed to write {clean_path}"            ) from exc    else:        logger.warning(            "No features passed QA/QC; no clean GeoPackage written."        )    # Export rejected source features separately where possible.    if not rejected.empty:        rejected_path = output_dir / "rejected.gpkg"        try:            rejected.to_file(                rejected_path,                layer="rejected",                driver="GPKG",                index=False            )            logger.info(                "Rejected features written to %s", rejected_path            )        except Exception:            # Some rejected geometries may not be serializable.            # Keep the audit CSV even when this export fails.            logger.exception(                "Could not write rejection geometries. "                "Use the audit CSV to identify affected records."            )    # Always produce an audit report.    audit_df = pd.DataFrame(audit_rows)    report_path = output_dir / "qa_audit.csv"    try:        audit_df.to_csv(report_path, index=False)    except Exception as exc:        logger.exception("Could not write QA audit report.")        raise RuntimeError(            f"Failed to write {report_path}"        ) from exc    logger.info(        "Finished: input=%s, accepted=%s, rejected=%s",        len(raw),        len(cleaned),        len(rejected)    )    return cleaned, rejected, audit_df# ------------------------------------------------------------# 5. Example execution# ------------------------------------------------------------if __name__ == "__main__":    cleaned_gdf, rejected_gdf, audit_df = run_spatial_qa_qc(        input_path="raw_parcels.gpkg",        layer="parcels",        output_dir="qa_output",        # Select a projected CRS appropriate for the study area.        # EPSG:32637 is an example for suitable locations in UTM 37N.        area_crs="EPSG:32637",        # Replace with your actual source attribute, if available.        id_field=None,        # Example threshold only; validate against your dataset.        min_area_m2=1.0    )    print("\nQA/QC summary:")    print(audit_df["status"].value_counts(dropna=False))

Understanding the outputs

After a successful run, the output directory should contain:

File	Contents
`cleaned.gpkg`	Features that passed the repair and acceptance rules
`rejected.gpkg`	Features rejected or quarantined for review, if writable
`qa_audit.csv`	Source identifiers, original validity reasons, repair methods, final statuses, and area measurements

A clean feature may be marked valid_unchanged or repaired. A feature can be rejected for a missing geometry, unexpected geometry type, failed repair, or area below the configured threshold.

The audit report intentionally records repair method and acceptance status separately. This lets an analyst distinguish a polygon that was repaired successfully from one that was already valid and passed unchanged.

Note that the script treats the area threshold as an optional business rule. Set min_area_m2=None to disable it, or configure a threshold appropriate to the mapping scale, legal requirements, and expected size of legitimate features.

Robustness adjustment for completely rejected datasets

One edge case is worth handling before production use: if every feature is rejected, a GeoDataFrame constructed from an empty list of records may not have the geometry column defined. To support datasets where zero features pass QA/QC, add this helper inside run_spatial_qa_qc(), after validating the source dataset:

def build_output_frame(records):    """Build a GeoDataFrame even when no records were accepted."""    if records:        return gpd.GeoDataFrame(            records,            geometry=geom_col,            crs=raw.crs        )    # Preserve the original schema and active geometry column.    empty = raw.iloc[0:0].copy()    empty["_qa_source_row"] = pd.Series(dtype="object")    return empty

Then replace the three GeoDataFrame constructors in the main function with:

accepted = build_output_frame(accepted_records)# ...perform the area screening as shown above...cleaned = build_output_frame(final_records)rejected = build_output_frame(rejected_records)

This allows the pipeline to finish its audit reporting even when every feature is quarantined. The export section can then skip creating an empty clean GeoPackage while still reporting why no features passed.

4. Exporting QA/QC Results to PostGIS

For multi-user GIS applications, a spatial database is often a more useful destination than a standalone file. After verifying that the cleaned output meets your acceptance criteria, you can load it into PostGIS.

Install the additional dependencies:

python -m pip install sqlalchemy geoalchemy2 "psycopg[binary]"

Store the PostgreSQL connection string in an environment variable named POSTGIS_URL. For example, a connection string may follow this pattern:

postgresql+psycopg://USER:PASSWORD@HOST:5432/DATABASE

Do not commit real credentials to a source-code repository.

import osimport loggingfrom sqlalchemy import create_enginelogger = logging.getLogger("spatial_qa_qc")database_url = os.environ.get("POSTGIS_URL")if not database_url:    raise RuntimeError(        "POSTGIS_URL is missing from the environment."    )engine = create_engine(database_url)try:    if cleaned_gdf.empty:        logger.warning(            "No accepted features are available for PostGIS."        )    else:        cleaned_gdf.to_postgis(            name="parcels_clean",            con=engine,            schema="staging",            if_exists="fail",            index=False        )        logger.info(            "Clean spatial data successfully loaded into PostGIS."        )except Exception:    logger.exception("PostGIS export failed.")

The staging schema must already exist, and the database user must have suitable permissions. The if_exists="fail" setting prevents this example from silently replacing an existing table.

After the import, perform a database-side validity check. If your geometry column is called geometry, run:

SELECT
    id,
    ST_IsValid(geometry) AS is_valid,
    ST_IsValidReason(geometry) AS validity_reason
FROM staging.parcels_clean
WHERE NOT ST_IsValid(geometry);

Replace id and geometry with the actual column names in your table. A successful query returning no rows indicates that no invalid geometries were found by this check. It does not establish that the features are geographically correct or free from overlaps.

PostGIS provides another layer of validation close to the system that will consume the data. This is useful when the pipeline feeds spatial joins, database constraints, web services, or other automated geoprocessing tasks.

5. Additional Topology Checks for Production Pipelines

Individual geometry validity is necessary, but it is not sufficient for every spatial dataset. A production pipeline should also test the relationships between features and preserve a clear distinction between structural defects and business-rule violations.

Detect overlapping polygons and coverage errors

Consider a dataset of adjacent administrative areas. Every polygon might pass geometry.is_valid, yet the dataset could still contain unintended gaps or overlaps.

In versions of GeoPandas and GEOS that support polygon coverage validation, you can run:

# Run this after individual geometries have been validated.if hasattr(cleaned_gdf.geometry, "is_valid_coverage"):    coverage_ok = cleaned_gdf.geometry.is_valid_coverage()    print("Coverage valid:", coverage_ok)    if not coverage_ok:        # Inspect edges associated with coverage errors.        bad_edges = cleaned_gdf.geometry.invalid_coverage_edges()        print("Coverage issues:", (~bad_edges.is_empty).sum())

Use coverage validation only where the features are expected to form a polygon coverage. For example, a dataset of neighboring administrative boundaries may be expected to have no overlaps, whereas a dataset of building footprints may legitimately contain gaps. The acceptance rules must reflect the spatial model.

Handle duplicate vertices and unclosed rings carefully

Repeated consecutive vertices may be removable without materially changing a polygon, but repeated coordinates are not automatically evidence of invalid topology. Shapely supports remove_repeated_points() for this purpose. If you add it to your cleaning workflow, record that the coordinates were changed and validate the result again.

Unclosed rings require a different approach. A malformed source file may fail during ingestion before any Shapely validity function can inspect its geometry. The main pipeline therefore catches read failures, but it cannot repair a feature that never reached the in-memory GeoDataFrame. Such cases need source-file inspection or a separate ingestion-recovery procedure.

A useful optional diagnostic for any suspicious polygon is:

# Only run this when a feature has actually been identified# as having repeated consecutive coordinates.candidate = shapely.remove_repeated_points(invalid_geom)if not candidate.is_empty:    print("Candidate valid:", candidate.is_valid)    print("Candidate type:", candidate.geom_type)

This is a diagnostic example, not a replacement for the main repair function. Check the output before accepting it, particularly if the geometry was already invalid.

Harden the pipeline for scheduled execution

For recurring data deliveries, incorporate the workflow into a scheduled job or continuous integration process. Useful operational controls include recording the input filename and processing timestamp, retaining the original source, tracking repaired and rejected feature counts, and failing the job when the proportion of invalid features exceeds an agreed tolerance.

For instance, a pipeline can reject an entire delivery if too many records require repair. This prevents a severely degraded source dataset from being accepted simply because every individual feature was eventually converted into a valid geometry.

6. Summary

An automated spatial QA/QC pipeline turns geometry validation from a manual troubleshooting exercise into a repeatable data engineering process.

GeoPandas manages vector datasets, Shapely identifies and repairs invalid geometry, and Fiona provides a reliable vector-data access option. Together, they support a workflow that detects topology failures, records error reasons, applies controlled repairs, checks repaired results, screens small polygon fragments, and exports accepted data to GeoPackage or PostGIS.

The most important engineering practices are to preserve the original data, validate every repaired result, use a suitable projected CRS for area calculations, and maintain an audit trail for rejected features. Do not assume that every invalid polygon can be repaired safely or that every small polygon is a sliver.

A reliable spatial data cleaning pipeline does not merely make geometries valid. It makes its decisions traceable, its acceptance criteria explicit, and its outputs suitable for the next stage of analysis.