# Mapping DCAT-AP Fields from PostGIS Columns

To map a PostGIS table to DCAT-AP, read the table’s structural metadata from information_schema and geometry_columns, derive a WGS84 bounding box with ST_Extent and ST_Transform, translate each source through an explicit field-mapping dictionary, and emit a dcat:Dataset graph with rdflib serialized to JSON-LD.

A PostGIS database is often the authoritative store behind a spatial data infrastructure, which makes it the ideal place to generate catalog records automatically rather than transcribing them by hand. This guide is a focused task within DCAT-AP Spatial Profile Mapping, which covers the full RDF construction and SHACL validation workflow, and part of the wider Automated Metadata Generation & Schema Mapping pipeline family. The value of driving DCAT-AP straight from the database is that the geometry SRID, extent, and column structure are read from the same tables that serve live queries — the metadata cannot drift away from the data it describes.

# Field mapping reference

The table below is the contract between PostGIS sources and DCAT-AP output. Everything the script produces flows from these rows.

# Automated Python Implementation

The script below is self-contained. It connects with psycopg (v3), reads geometry metadata from geometry_columns, computes a reprojected extent, applies the mapping dictionary above, and serializes a DCAT-AP graph to JSON-LD with rdflib. Connection parameters come from the standard PG* environment variables so no credentials are hard-coded.

#!/usr/bin/env python3
"""
postgis_to_dcat.py — Map a PostGIS table to a DCAT-AP dataset record.

Usage:
    python postgis_to_dcat.py <schema> <table> <out.jsonld>

Environment:
    PGHOST, PGPORT, PGDATABASE, PGUSER, PGPASSWORD  (standard libpq vars)
    DCAT_ACCESS_URL   base URL where the layer is served
    DCAT_LICENSE_URI  absolute license URI
"""
import os
import sys
import uuid

import psycopg
from rdflib import Graph, Namespace, URIRef, Literal, BNode
from rdflib.namespace import RDF, DCTERMS, FOAF

DCAT = Namespace("http://www.w3.org/ns/dcat#")
GSP = Namespace("http://www.opengis.net/ont/geosparql#")

# The mapping contract: PostGIS source -> DCAT-AP handling.
FIELD_MAP = {
    "comment": DCTERMS.description,
    "srid": DCTERMS.conformsTo,
    "extent_wkt": DCAT.bbox,
    "columns": DCAT.keyword,
}


def read_postgis_metadata(conn, schema: str, table: str) -> dict:
    """Read structural + spatial metadata for one PostGIS table."""
    with conn.cursor() as cur:
        # Geometry column + SRID from the standard PostGIS view
        cur.execute(
            """
            SELECT f_geometry_column, srid, type
            FROM geometry_columns
            WHERE f_table_schema = %s AND f_table_name = %s
            LIMIT 1
            """,
            (schema, table),
        )
        geom_row = cur.fetchone()
        if geom_row is None:
            raise ValueError(f"No geometry column registered for {schema}.{table}")
        geom_col, srid, geom_type = geom_row

        # Attribute column names for keywords
        cur.execute(
            """
            SELECT column_name
            FROM information_schema.columns
            WHERE table_schema = %s AND table_name = %s
            ORDER BY ordinal_position
            """,
            (schema, table),
        )
        columns = [r[0] for r in cur.fetchall()]

        # Human-readable table comment for the description
        cur.execute("SELECT obj_description(%s::regclass, 'pg_class')",
                    (f'"{schema}"."{table}"',))
        comment = cur.fetchone()[0] or f"Spatial dataset {schema}.{table}"

        # WGS84 extent as WKT: reproject the aggregate envelope to EPSG:4326.
        # ST_Extent returns a box2d, so cast to geometry before transforming.
        cur.execute(
            f"""
            SELECT ST_AsText(
                       ST_Transform(
                           ST_SetSRID(ST_Extent("{geom_col}"), %s), 4326))
            FROM "{schema}"."{table}"
            """,
            (srid,),
        )
        extent_wkt = cur.fetchone()[0]

    return {
        "schema": schema,
        "table": table,
        "geometry_column": geom_col,
        "srid": srid,
        "geometry_type": geom_type,
        "columns": columns,
        "comment": comment,
        "extent_wkt": extent_wkt,
    }


def build_dcat_graph(meta: dict) -> Graph:
    """Translate PostGIS metadata into a DCAT-AP RDF graph via FIELD_MAP."""
    g = Graph()
    g.bind("dcat", DCAT)
    g.bind("dct", DCTERMS)
    g.bind("gsp", GSP)
    g.bind("foaf", FOAF)

    base = os.environ.get("DCAT_ACCESS_URL", "https://data.example.gov/datasets/")
    dataset = URIRef(f"{base.rstrip('/')}/{meta['schema']}.{meta['table']}")
    g.add((dataset, RDF.type, DCAT.Dataset))

    # Title from the first line of the comment; description from the whole comment
    title = meta["comment"].splitlines()[0].strip()
    g.add((dataset, DCTERMS.title, Literal(title, lang="en")))
    g.add((dataset, FIELD_MAP["comment"], Literal(meta["comment"], lang="en")))

    # CRS conformance from the native SRID
    crs_uri = f"http://www.opengis.net/def/crs/EPSG/0/{meta['srid']}"
    g.add((dataset, FIELD_MAP["srid"], URIRef(crs_uri)))

    # Bounding box as a WKT literal (already reprojected to EPSG:4326)
    # ST_Extent yields a POLYGON; wrap defensively if the table was empty.
    bbox_wkt = meta["extent_wkt"] or "POLYGON EMPTY"
    g.add((dataset, FIELD_MAP["extent_wkt"],
           Literal(bbox_wkt, datatype=GSP.wktLiteral)))

    # Attribute names become discovery keywords
    for col in meta["columns"]:
        if col != meta["geometry_column"]:
            g.add((dataset, FIELD_MAP["columns"], Literal(col)))

    # Publisher as a blank-node organization
    publisher = BNode()
    g.add((publisher, RDF.type, FOAF.Organization))
    g.add((publisher, FOAF.name, Literal(os.environ.get("DCAT_PUBLISHER", "GIS Unit"))))
    g.add((dataset, DCTERMS.publisher, publisher))

    # Distribution with access URL and license
    dist = URIRef(f"{dataset}/dist/{uuid.uuid5(uuid.NAMESPACE_URL, str(dataset))}")
    g.add((dist, RDF.type, DCAT.Distribution))
    g.add((dist, DCAT.accessURL, URIRef(str(dataset))))
    license_uri = os.environ.get("DCAT_LICENSE_URI")
    if license_uri:
        g.add((dist, DCTERMS.license, URIRef(license_uri)))
    g.add((dataset, DCAT.distribution, dist))
    return g


def main() -> int:
    if len(sys.argv) != 4:
        print("Usage: python postgis_to_dcat.py <schema> <table> <out.jsonld>",
              file=sys.stderr)
        return 1
    schema, table, out_path = sys.argv[1], sys.argv[2], sys.argv[3]
    with psycopg.connect() as conn:  # reads PG* env vars
        meta = read_postgis_metadata(conn, schema, table)
    graph = build_dcat_graph(meta)
    graph.serialize(destination=out_path, format="json-ld", indent=2)
    print(f"Wrote {len(graph)} triples to {out_path}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

The deterministic distribution URI (uuid.uuid5 over the dataset URI rather than uuid.uuid4) is intentional: re-running the mapper for the same table produces the same graph, so a downstream triple store overwrites the record instead of accumulating duplicates on every harvest.

# Validation and pipeline integration

Confirm the emitted JSON-LD parses back to a graph with the mandatory properties present before handing it to a portal harvester:

pip install "psycopg[binary]>=3.1" "rdflib>=6.3"

# Round-trip check: the file must re-parse and contain a dcat:Dataset
python - <<'PY'
from rdflib import Graph
from rdflib.namespace import RDF
from rdflib import Namespace
DCAT = Namespace("http://www.w3.org/ns/dcat#")
g = Graph().parse("out.jsonld", format="json-ld")
assert (None, RDF.type, DCAT.Dataset) in g, "no dcat:Dataset produced"
assert (None, DCAT.bbox, None) in g, "missing dcat:bbox"
print(f"OK: {len(g)} triples")
PY

A pytest block asserts the mapping contract holds against a fixture metadata dict, so the graph shape stays stable as the mapper evolves:

# test_postgis_to_dcat.py
from rdflib.namespace import RDF, DCTERMS
from postgis_to_dcat import build_dcat_graph, DCAT, GSP

FIXTURE = {
    "schema": "public", "table": "parcels", "geometry_column": "geom",
    "srid": 27700, "geometry_type": "MULTIPOLYGON",
    "columns": ["gid", "geom", "parcel_ref", "owner"],
    "comment": "Cadastral parcels\nAuthoritative parcel boundaries.",
    "extent_wkt": "POLYGON((-1 51, 0 51, 0 52, -1 52, -1 51))",
}


def test_required_properties_present():
    g = build_dcat_graph(FIXTURE)
    assert (None, RDF.type, DCAT.Dataset) in g
    assert (None, DCAT.bbox, None) in g
    assert (None, DCTERMS.title, None) in g


def test_bbox_is_wkt_literal():
    g = build_dcat_graph(FIXTURE)
    bbox = next(g.objects(predicate=DCAT.bbox))
    assert bbox.datatype == GSP.wktLiteral

# Long-term compliance best practices

• Read the SRID, never assume it. Pull srid from geometry_columns for every table; hard-coding EPSG:4326 silently mislabels any table stored in a national grid and corrupts the reprojected dcat:bbox.
• Keep table comments authoritative. Enforce COMMENT ON TABLE as part of your schema migrations so dct:title and dct:description always have a real source rather than a placeholder.
• Derive distribution URIs deterministically. Use uuid5 over a stable key so repeated harvests update one record instead of multiplying it in the target triple store.
• Validate against the target portal’s SHACL, not just the reference shapes. National profiles add cardinality rules; run the graph through the portal’s published shapes before publishing, as covered in the parent guide.
• Handle empty tables explicitly. ST_Extent over a table with no rows returns null; emit a documented placeholder or skip the record rather than serializing a malformed dcat:bbox.
• Reproject at the database, not in Python. Let PostGIS do ST_Transform; it uses the same PROJ definitions your queries rely on, which keeps the catalog extent consistent with live spatial operations.

# Related

• DCAT-AP Spatial Profile Mapping — the parent guide with full RDF construction, namespaces, and SHACL validation
• Python Scripts for DCAT-AP Spatial Dataset Mapping — companion runnable mapping scripts
• Automating Metadata Extraction from PostGIS Tables — extracting the underlying metadata this mapping consumes
• Automated Metadata Generation & Schema Mapping — the pipeline context for database-driven cataloging