Geospatial Compliance Metadata Automation Hub

Python Scripts for DCAT-AP Spatial Dataset Mapping

Use rdflib to construct a DCAT-AP Spatial Profile-compliant RDF graph from a Python dictionary, serialize it to Turtle or JSON-LD, and validate it with pySHACL before publishing to any EU open data catalog.

Automating this transformation is non-trivial because the DCAT-AP Spatial Profile adds geospatial-specific constraints on top of the base W3C DCAT vocabulary: bounding boxes must follow a precise WGS84 string format, coordinate reference systems must resolve to authoritative OGC URIs, and licensing terms must be persistent URI references rather than free-text literals. Manual RDF authoring at scale introduces silent constraint violations that only surface when a national portal harvester rejects the feed. This workflow is a direct implementation of the DCAT-AP Spatial Profile Mapping patterns, and sits within the broader Automated Metadata Generation & Schema Mapping discipline that governs production spatial data infrastructures.

The diagram below shows the end-to-end data flow from source metadata to a validated catalog record:

DCAT-AP Spatial Mapping Pipeline Data flows left to right: Source Metadata (Python dict) feeds into rdflib Graph Builder, which produces an RDF Graph. The graph passes through pySHACL Validator. On success the graph is serialised to Turtle or JSON-LD and published to CKAN or GeoNetwork. On failure the violations are reported back. Source Metadata dict rdflib Graph Builder pySHACL Validator Serialize Turtle / JSON-LD Catalog publish violations → fix & rebuild

Automated Python Implementation

Install the two core dependencies before running the script:

pip install rdflib>=6.3 pyshacl>=0.25

The script below is self-contained and runnable. It builds a compliant RDF graph, validates it against a locally-loaded SHACL shapes graph, and serializes the output to Turtle. Inline comments explain every non-obvious mapping decision.

# dcat_ap_spatial_mapper.py
"""
DCAT-AP Spatial Profile mapper.
Requires: rdflib>=6.3, pyshacl>=0.25
"""
from __future__ import annotations

import sys
from typing import Any

from rdflib import Graph, Literal, Namespace, URIRef
from rdflib.namespace import DCAT, DCTERMS, RDF, XSD

# ── Namespace declarations ────────────────────────────────────────────────────
# LOCN is not yet in rdflib's built-ins, so declare it manually.
LOCN = Namespace("http://www.w3.org/ns/locn#")
GEO = Namespace("http://www.opengis.net/ont/geosparql#")
DCATAP = Namespace("http://data.europa.eu/r5r/")
DCT = DCTERMS


def _ogc_crs_uri(epsg_code: str | int) -> URIRef:
    """Return the authoritative OGC EPSG URI for a given EPSG code.

    DCAT-AP Spatial Profile v2/v3 requires this exact URI pattern.
    Bare integer codes (e.g. 4326) are not acceptable as dcat:crs values.
    """
    return URIRef(f"https://www.opengis.net/def/crs/EPSG/0/{epsg_code}")


def build_spatial_dataset(metadata: dict[str, Any]) -> Graph:
    """Construct a DCAT-AP Spatial Profile-compliant RDF graph.

    Parameters
    ----------
    metadata : dict
        Keys documented in the example at the bottom of this file.

    Returns
    -------
    rdflib.Graph
        Populated graph with all mandatory and recommended properties.
    """
    g = Graph()

    # Bind prefixes explicitly — unprefixed URIs bloat serialized output and
    # break cross-catalog SPARQL queries that rely on known prefix declarations.
    g.bind("dcat", DCAT)
    g.bind("dct", DCT)
    g.bind("locn", LOCN)
    g.bind("geo", GEO)
    g.bind("xsd", XSD)
    g.bind("dcatap", DCATAP)

    ds_uri = URIRef(metadata["dataset_uri"])
    g.add((ds_uri, RDF.type, DCAT.Dataset))

    # ── Mandatory properties ─────────────────────────────────────────────────
    g.add((ds_uri, DCT.title, Literal(metadata["title"], lang="en")))
    g.add((ds_uri, DCT.identifier, Literal(metadata["identifier"])))
    g.add((ds_uri, DCT.description, Literal(metadata["description"], lang="en")))

    # ── Spatial coverage ─────────────────────────────────────────────────────
    # dcat:bbox must be a plain string literal in "west south east north" order
    # (WGS84 decimal degrees). Do NOT use a typed xsd:string or a GeoJSON dict.
    if "bbox" in metadata:
        w, s, e, n = metadata["bbox"]
        g.add((ds_uri, DCAT.bbox, Literal(f"{w} {s} {e} {n}")))

    # locn:geometry accepts WKT or GeoJSON as a string literal (optional but
    # recommended for detailed footprints that exceed a bounding rectangle).
    if "geometry_wkt" in metadata:
        g.add((ds_uri, LOCN.geometry, Literal(metadata["geometry_wkt"])))

    # ── Coordinate Reference System ──────────────────────────────────────────
    # Must be an OGC URI, not a bare EPSG integer or an SRID string.
    if "epsg" in metadata:
        g.add((ds_uri, DCAT.crs, _ogc_crs_uri(metadata["epsg"])))

    # ── Licensing and rights ─────────────────────────────────────────────────
    # dct:license must be a URI reference (SPDX or Creative Commons URI).
    # A free-text literal here will fail SHACL validation for DCAT-AP v2+.
    if "license_uri" in metadata:
        g.add((ds_uri, DCT.license, URIRef(metadata["license_uri"])))
    if "rights_statement" in metadata:
        g.add((ds_uri, DCT.rights, Literal(metadata["rights_statement"], lang="en")))
    if "access_rights_uri" in metadata:
        g.add((ds_uri, DCT.accessRights, URIRef(metadata["access_rights_uri"])))

    # ── Temporal extent ──────────────────────────────────────────────────────
    if "temporal_start" in metadata:
        period = URIRef(metadata["dataset_uri"] + "/temporal")
        g.add((ds_uri, DCT.temporal, period))
        g.add((period, RDF.type, DCT.PeriodOfTime))
        g.add((period, DCAT.startDate,
               Literal(metadata["temporal_start"], datatype=XSD.date)))
        if "temporal_end" in metadata:
            g.add((period, DCAT.endDate,
                   Literal(metadata["temporal_end"], datatype=XSD.date)))

    # ── Distributions ────────────────────────────────────────────────────────
    for dist in metadata.get("distributions", []):
        dist_uri = URIRef(dist["distribution_uri"])
        g.add((ds_uri, DCAT.distribution, dist_uri))
        g.add((dist_uri, RDF.type, DCAT.Distribution))
        g.add((dist_uri, DCT.title, Literal(dist["title"], lang="en")))

        # mediaType should be an IANA media type URI, not a bare string, for
        # strict DCAT-AP conformance.
        if "media_type" in dist:
            g.add((dist_uri, DCAT.mediaType,
                   URIRef(f"https://www.iana.org/assignments/media-types/{dist['media_type']}")))
        if "access_url" in dist:
            g.add((dist_uri, DCAT.accessURL, URIRef(dist["access_url"])))
        if "download_url" in dist:
            g.add((dist_uri, DCAT.downloadURL, URIRef(dist["download_url"])))
        if "byte_size" in dist:
            g.add((dist_uri, DCAT.byteSize,
                   Literal(dist["byte_size"], datatype=XSD.nonNegativeInteger)))

    return g


def validate_graph(data_graph: Graph, shapes_ttl_path: str) -> bool:
    """Validate *data_graph* against a local copy of DCAT-AP SHACL shapes.

    Download the shapes file from the SEMIC DCAT-AP repository and pass its
    path here. Returns True when all constraints are satisfied.
    """
    from pyshacl import validate  # type: ignore[import]

    shapes_graph = Graph().parse(shapes_ttl_path, format="turtle")
    conforms, _, report_text = validate(
        data_graph,
        shacl_graph=shapes_graph,
        inference="rdfs",
        abort_on_first=False,
    )
    if not conforms:
        print(report_text, file=sys.stderr)
    return conforms


# ── Example usage ────────────────────────────────────────────────────────────
if __name__ == "__main__":
    import pathlib

    sample_meta: dict[str, Any] = {
        "dataset_uri": "https://data.example.eu/datasets/land-cover-2024",
        "title": "European Land Cover 2024",
        "identifier": "LC-2024-EU",
        "description": (
            "Annual land cover classification for Europe at 10 m resolution, "
            "derived from Sentinel-2 composites."
        ),
        "bbox": [-10.5, 35.0, 35.0, 72.0],  # west south east north, WGS84
        "epsg": 4326,
        "license_uri": "https://creativecommons.org/licenses/by/4.0/",
        "temporal_start": "2024-01-01",
        "temporal_end": "2024-12-31",
        "distributions": [
            {
                "distribution_uri": "https://data.example.eu/dist/lc-2024-geojson",
                "title": "GeoJSON Export",
                "media_type": "application/geo+json",
                "download_url": "https://data.example.eu/files/lc-2024.geojson",
                "byte_size": 94371840,
            },
            {
                "distribution_uri": "https://data.example.eu/dist/lc-2024-turtle",
                "title": "Turtle RDF Export",
                "media_type": "text/turtle",
                "download_url": "https://data.example.eu/files/lc-2024.ttl",
            },
        ],
    }

    graph = build_spatial_dataset(sample_meta)

    # Serialize to Turtle for catalog ingestion or diff review in VCS
    output_path = pathlib.Path("lc-2024-dcat-ap.ttl")
    graph.serialize(destination=str(output_path), format="turtle")
    print(f"Serialized {len(graph)} triples to {output_path}")

    # Optional SHACL validation — supply the path to a local shapes file
    shapes_path = pathlib.Path("dcat-ap-shacl.ttl")
    if shapes_path.exists():
        ok = validate_graph(graph, str(shapes_path))
        sys.exit(0 if ok else 1)

Key design decisions in the script:

  • _ogc_crs_uri() centralises the EPSG-to-OGC-URI conversion so bare integer codes never reach the graph.
  • dcat:bbox is always a plain string literal. Typed xsd:string values pass serialization but fail SEMIC SHACL shapes that check the literal’s datatype tag.
  • dcat:mediaType is resolved to an IANA URI rather than a bare MIME string, matching the DCAT-AP v2.1 normative requirement.
  • dct:temporal creates a dct:PeriodOfTime blank node with typed xsd:date literals rather than untyped strings.

Validation and Pipeline Integration

Verify the Turtle output is parseable

# rdflib must be able to round-trip the file without parse errors
python - <<'EOF'
from rdflib import Graph
g = Graph().parse("lc-2024-dcat-ap.ttl", format="turtle")
print(f"Parsed {len(g)} triples OK")
EOF

Run mandatory-property checks with pytest

Add a fixture-driven test to your test suite. This catches regressions when the metadata dictionary evolves:

# tests/test_dcat_ap_mapper.py
import pytest
from rdflib.namespace import DCAT, DCTERMS, RDF
from rdflib import URIRef
from dcat_ap_spatial_mapper import build_spatial_dataset

SAMPLE = {
    "dataset_uri": "https://example.org/ds/1",
    "title": "Test Dataset",
    "identifier": "TEST-001",
    "description": "A test spatial dataset.",
    "bbox": [-5.0, 50.0, 2.0, 59.0],
    "epsg": 4326,
    "license_uri": "https://creativecommons.org/licenses/by/4.0/",
    "distributions": [
        {
            "distribution_uri": "https://example.org/ds/1/dist/1",
            "title": "GeoJSON",
            "media_type": "application/geo+json",
            "access_url": "https://example.org/ds/1/dist/1",
        }
    ],
}


@pytest.fixture()
def graph():
    return build_spatial_dataset(SAMPLE)


def test_type(graph):
    ds = URIRef(SAMPLE["dataset_uri"])
    assert (ds, RDF.type, DCAT.Dataset) in graph


def test_bbox_format(graph):
    ds = URIRef(SAMPLE["dataset_uri"])
    bbox_vals = list(graph.objects(ds, DCAT.bbox))
    assert len(bbox_vals) == 1
    parts = str(bbox_vals[0]).split()
    assert len(parts) == 4, "bbox must be 'west south east north'"


def test_license_is_uri(graph):
    ds = URIRef(SAMPLE["dataset_uri"])
    license_vals = list(graph.objects(ds, DCTERMS.license))
    assert all(isinstance(v, URIRef) for v in license_vals), \
        "dct:license must be a URI, not a Literal"


def test_crs_ogc_uri(graph):
    ds = URIRef(SAMPLE["dataset_uri"])
    crs_vals = list(graph.objects(ds, DCAT.crs))
    assert len(crs_vals) == 1
    assert "opengis.net/def/crs/EPSG" in str(crs_vals[0])

Run the suite with:

python -m pytest tests/test_dcat_ap_mapper.py -v

GitHub Actions CI gate

Add this step to your workflow to block merges that produce invalid DCAT-AP output:

- name: Validate DCAT-AP spatial metadata
  run: |
    pip install rdflib>=6.3 pyshacl>=0.25
    python dcat_ap_spatial_mapper.py
    python -m pytest tests/test_dcat_ap_mapper.py -v

For SHACL validation in CI, cache the DCAT-AP shapes file as a repository artifact rather than downloading it on every run. This keeps the pipeline deterministic and avoids failures caused by upstream availability.

Long-Term Compliance Best Practices

  • Pin the DCAT-AP version your catalog targets. The EU has published DCAT-AP v2.0, v2.1, and v3.0; each version changes which properties are mandatory versus recommended. Record the target version in a COMPLIANCE.md file alongside your mapper and review it when upgrading catalog software.

  • Store the shapes graph in version control alongside the mapper. SEMIC publishes updated SHACL shapes with each DCAT-AP release. Pinning the shapes file to a commit hash ensures that a shapes update upstream does not silently change your validation results between pipeline runs.

  • Use persistent, resolvable license URIs. Creative Commons URIs (https://creativecommons.org/licenses/by/4.0/) and SPDX identifiers resolve reliably. Internal license registry URLs behind a corporate VPN will cause harvester failures when national portals attempt to dereference dct:license.

  • Validate coordinate order before inserting bbox values. The most common mapping failure is swapped latitude/longitude order. Add an assertion assert -90 <= s <= n <= 90 and -180 <= w <= e <= 180 before calling build_spatial_dataset. Silent coordinate order errors produce bounding boxes that pass SHACL but place datasets on the wrong continent in spatial search interfaces.

  • Design the mapper for idempotent updates. When refreshing existing catalog records, call graph.remove((ds_uri, None, None)) before adding updated triples. This prevents duplicate dct:title or dcat:bbox statements that accumulate in incremental update workflows and corrupt SPARQL query results.

  • Automate CRS URI resolution checks in integration tests. OGC EPSG URIs follow a stable pattern, but EPSG codes for deprecated or regional CRS definitions can return non-200 responses from the OGC registry. A periodic integration test that HEAD-checks a sample of your dcat:crs URIs catches deprecated codes before they surface in harvester logs. See Automating Metadata Extraction from PostGIS Tables for a pattern that combines SRID extraction with CRS URI construction.