# Topology Validation with Dagster Asset Checks

Invalid geometry is insidious because it passes silently: a self-intersecting parcel or an unclosed ring loads without error, tiles without complaint, and only breaks when a downstream spatial join returns garbage or a renderer throws. Topology validation with Dagster asset checks stops that at the source by attaching an @asset_check to each spatial asset that runs PostGIS ST_IsValid — and, where policy allows, ST_MakeValid — so a materialization carrying invalid polygons is blocked from publication rather than propagated. It is the Dagster implementation of the gating pattern in Orchestrating Spatial Pipelines in Python, under Spatial Pipeline Orchestration & Observability; the Prefect gate is covered in Building Idempotent Spatial DAGs with Prefect.

# Prerequisites

Figure — A materialized geometry asset is gated by an asset check: valid geometry publishes, invalid geometry blocks or routes to a make-valid remediation.

# Step-by-Step Implementation

Each step is verifiable before you proceed. The invariant is that the check is bound to the asset, so publication cannot materialize while an error-severity topology failure stands.

# 1. Define the reprojected geometry asset

The asset loads reprojected features into a domain-scoped PostGIS table. It is the subject the asset check will validate.

# assets.py
import os
from dagster import asset, AssetExecutionContext

@asset
def parcels_geom(context: AssetExecutionContext) -> None:
    # Load the EPSG:4326 reprojected features into the domain schema.
    run_psql(os.environ["DAGSTER_PG_DSN"], """
        CREATE TABLE IF NOT EXISTS domain.parcels_geom
          (id bigint primary key, geom geometry(Geometry, 4326));
        INSERT INTO domain.parcels_geom SELECT id, geom FROM staging.parcels
          ON CONFLICT (id) DO UPDATE SET geom = EXCLUDED.geom;  -- idempotent upsert
    """)
    context.log.info("materialized domain.parcels_geom")

Verify: confirm the table populated in EPSG:4326.

psql "$DAGSTER_PG_DSN" -tAc \
  "SELECT count(*), ST_SRID(geom) FROM domain.parcels_geom GROUP BY 2;"

# 2. Attach an error-severity asset check running ST_IsValid

The @asset_check counts invalid geometries and returns a passing or failing result. Error severity is what makes a failure block the downstream publish asset rather than merely warn.

from dagster import asset_check, AssetCheckResult, AssetCheckSeverity

@asset_check(asset=parcels_geom, blocking=True)
def parcels_topology_valid() -> AssetCheckResult:
    invalid = int(run_psql_scalar(os.environ["DAGSTER_PG_DSN"],
        "SELECT count(*) FROM domain.parcels_geom WHERE NOT ST_IsValid(geom);"))
    return AssetCheckResult(
        passed=(invalid == 0),
        severity=AssetCheckSeverity.ERROR,
        metadata={"invalid_geometries": invalid},
    )

Verify: count invalid geometries the check will see.

psql "$DAGSTER_PG_DSN" -tAc \
  "SELECT count(*) FROM domain.parcels_geom WHERE NOT ST_IsValid(geom);"

# 3. Surface the failure reason for diagnosis

A count is not actionable alone; capture ST_IsValidReason so an operator can locate the offending feature without a separate query.

@asset_check(asset=parcels_geom)
def parcels_topology_reasons() -> AssetCheckResult:
    rows = run_psql_rows(os.environ["DAGSTER_PG_DSN"], """
        SELECT id, ST_IsValidReason(geom) AS reason
        FROM domain.parcels_geom WHERE NOT ST_IsValid(geom) LIMIT 20;""")
    return AssetCheckResult(
        passed=(len(rows) == 0),
        metadata={"sample_invalid": str(rows)},
    )

Verify: read the reasons directly.

psql "$DAGSTER_PG_DSN" -tAc \
  "SELECT id, ST_IsValidReason(geom) FROM domain.parcels_geom
   WHERE NOT ST_IsValid(geom) LIMIT 5;"

# 4. Add an optional ST_MakeValid remediation asset

Where policy permits automatic repair, a downstream asset applies ST_MakeValid and the blocking check re-runs against the repaired geometry before publication.

@asset(deps=[parcels_geom])
def parcels_geom_repaired(context: AssetExecutionContext) -> None:
    # Repair in place only where policy allows automatic remediation.
    run_psql(os.environ["DAGSTER_PG_DSN"], """
        UPDATE domain.parcels_geom SET geom = ST_MakeValid(geom)
        WHERE NOT ST_IsValid(geom);""")
    context.log.info("applied ST_MakeValid to invalid geometries")

Verify: confirm no invalid geometry remains after repair.

psql "$DAGSTER_PG_DSN" -tAc \
  "SELECT count(*) FROM domain.parcels_geom WHERE NOT ST_IsValid(geom);"

# 5. Wire the publish asset behind the blocking check

The publish asset depends on parcels_geom, and because the blocking check gates that asset, an error-severity failure prevents publication automatically.

from dagster import asset, Definitions

@asset(deps=[parcels_geom])
def parcels_published(context: AssetExecutionContext) -> None:
    write_catalog_entry("parcels", version="v1.2.0-crs:EPSG:4326-res:10m")

defs = Definitions(
    assets=[parcels_geom, parcels_geom_repaired, parcels_published],
    asset_checks=[parcels_topology_valid, parcels_topology_reasons],
)

Verify: materialize and confirm the check gated publication.

dagster asset materialize --select parcels_geom+ -f assets.py
dagster asset list-checks -f assets.py | grep parcels_topology_valid

# Configuration Reference

# Common Failure Modes & Fixes

The check passes but invalid geometry still publishes. Root cause: the check was not blocking=True or not ERROR severity, so Dagster recorded a warning without gating the publish asset. Fix: set both, then re-materialize and confirm parcels_published did not run while the check failed.

ST_IsValid reports invalid on geometry that renders fine. Root cause: a self-intersection or ring-orientation issue that renderers tolerate but the OGC validity model rejects. Fix: inspect ST_IsValidReason, and if the geometry is genuinely acceptable, apply ST_MakeValid in the remediation asset rather than relaxing the gate.

ST_MakeValid changes geometry type (polygon to collection). Root cause: repairing a bowtie can split a polygon into a multipolygon or collection. Fix: follow ST_MakeValid with ST_CollectionExtract(geom, 3) to keep only polygonal parts, then re-run the validity check.

The check runs on pre-reprojection data. Root cause: it was bound to a staging asset still in the source CRS, so ST_SRID is not 4326. Fix: bind the check to the reprojected asset and assert ST_SRID(geom) = 4326 before validating topology.

Validation is slow on large tables. Root cause: ST_IsValid runs per row without a spatial index or batching. Fix: validate in bounded batches, ensure a GiST index on geom, and parallelize by tile so the check stays within its SLA window.

# FAQ

# Related

• Up to the parent: Orchestrating Spatial Pipelines in Python
• Building Idempotent Spatial DAGs with Prefect — the zero-retry validation gate in Prefect
• Automating CRS Reprojection in Airflow — the fail-closed quality check in Airflow
• Spatial Pipeline Orchestration & Observability — the section overview and SLA baselines
• Spatial Domain Boundary Design — why topology rules are owned per domain