RDF store

app.rdf

RDF store wrapper: Oxigraph for SPARQL, rdflib for SHACL introspection.

RDFStore(data_path, shapes_path, vocab_path)

In-process ontology store used by every query route.

Oxigraph answers SPARQL; a lazily built rdflib Graph serves SHACL introspection. A single process-wide instance is held on _instance.

Load Turtle files into a fresh Oxigraph store.

Parameters:

Name	Type	Description	Default
data_path	str	Path to compass.ttl (instance data).	required
shapes_path	str	Path to shapes.ttl (SHACL).	required
vocab_path	str	Path to vocab.ttl (SKOS / class labels).	required

Source code in src/backend/app/rdf.py

def __init__(self, data_path: str, shapes_path: str, vocab_path: str):
    """Load Turtle files into a fresh Oxigraph store.

    Args:
        data_path: Path to ``compass.ttl`` (instance data).
        shapes_path: Path to ``shapes.ttl`` (SHACL).
        vocab_path: Path to ``vocab.ttl`` (SKOS / class labels).
    """
    self.store = pyoxigraph.Store()
    self.data_path = data_path
    self.shapes_path = shapes_path
    self.vocab_path = vocab_path
    self._read_graph: Graph | None = None
    self._entity_shapes_cache: list[EntityShape] | None = None
    self.load_data()

read_graph property

Merged rdflib graph of shapes, data, and vocab (parsed once).

Returns:

Type	Description
Graph	Shared Graph used by SHACL projection helpers.

load_data()

Parse the three Turtle files into the Oxigraph store.

Source code in src/backend/app/rdf.py

def load_data(self) -> None:
    """Parse the three Turtle files into the Oxigraph store."""
    with open(self.data_path, "rb") as f:
        self.store.load(f, pyoxigraph.RdfFormat.TURTLE)
    with open(self.shapes_path, "rb") as f:
        self.store.load(f, pyoxigraph.RdfFormat.TURTLE)
    with open(self.vocab_path, "rb") as f:
        self.store.load(f, pyoxigraph.RdfFormat.TURTLE)

query(sparql)

Run a SPARQL SELECT; one dict per row, unbound variables omitted.

Parameters:

Name	Type	Description	Default
sparql	str	Full SELECT query string.	required

Returns:

Type	Description
list[dict[str, Any]]	List of row dicts keyed by variable name.

Raises:

Type	Description
QueryError	When Oxigraph rejects or fails the query. The failing query text is attached for logging.

Source code in src/backend/app/rdf.py

def query(self, sparql: str) -> list[dict[str, Any]]:
    """Run a SPARQL SELECT; one dict per row, unbound variables omitted.

    Args:
        sparql: Full SELECT query string.

    Returns:
        List of row dicts keyed by variable name.

    Raises:
        QueryError: When Oxigraph rejects or fails the query. The failing
            query text is attached for logging.
    """
    start = time.time()
    try:
        results = self.store.query(sparql)
        parsed = []
        for row in results:
            item = {}
            for var in results.variables:
                val = row[var]
                if val is not None:
                    item[var.value] = (
                        f"_:{val.value}"
                        if isinstance(val, pyoxigraph.BlankNode)
                        else val.value
                    )
            parsed.append(item)
        logger.debug("SPARQL query executed in %.4fs", time.time() - start)
        return parsed
    except Exception as exc:
        logger.exception("SPARQL query failed:\n%s", sparql)
        raise QueryError(sparql, exc) from exc

get_entities()

Return cached EntityShape descriptors projected from SHACL.

Returns:

Type	Description
list[EntityShape]	Property descriptors used to build SPARQL and decode GeoJSON.

Source code in src/backend/app/rdf.py

def get_entities(self) -> list[EntityShape]:
    """Return cached ``EntityShape`` descriptors projected from SHACL.

    Returns:
        Property descriptors used to build SPARQL and decode GeoJSON.
    """
    if self._entity_shapes_cache is None:
        self._entity_shapes_cache = get_entity_shape_from_shacl(self.read_graph)
    return self._entity_shapes_cache

validate()

Reject a store that parsed but cannot answer a query.

Turtle can parse and still be useless — a truncated file, or shapes that no longer describe the data — and a reload that swapped such a store in would take the map down. Deriving entity shapes exercises the SHACL introspection the whole query layer is built on, and counting entities proves the data reached the store.

Raises:

Type	Description
ReloadError	When shapes yield nothing or no entity has coordinates.

Source code in src/backend/app/rdf.py

def validate(self) -> None:
    """Reject a store that parsed but cannot answer a query.

    Turtle can parse and still be useless — a truncated file, or shapes that
    no longer describe the data — and a reload that swapped such a store in
    would take the map down. Deriving entity shapes exercises the SHACL
    introspection the whole query layer is built on, and counting entities
    proves the data reached the store.

    Raises:
        ReloadError: When shapes yield nothing or no entity has coordinates.
    """
    shapes = self.get_entities()
    if not shapes:
        raise ReloadError(
            "the shapes yielded no EntityShape fields, so no filter would work"
        )
    rows = self.query(
        "PREFIX geo: <http://www.w3.org/2003/01/geo/wgs84_pos#> "
        "SELECT (COUNT(DISTINCT ?s) AS ?n) WHERE { ?s geo:lat ?lat . }"
    )
    if not rows or int(rows[0].get("n", 0)) == 0:
        raise ReloadError(
            "no entity in the data has coordinates, so the map would be empty"
        )

from_settings() classmethod

Build a store from the configured ontology and use-case directories.

Returns:

Type	Description
RDFStore	New RDFStore pointing at shapes.ttl under
RDFStore	settings.ontology_dir and compass.ttl / vocab.ttl under
RDFStore	settings.use_case_dir.

Source code in src/backend/app/rdf.py

@classmethod
def from_settings(cls) -> RDFStore:
    """Build a store from the configured ontology and use-case directories.

    Returns:
        New ``RDFStore`` pointing at ``shapes.ttl`` under
        ``settings.ontology_dir`` and ``compass.ttl`` / ``vocab.ttl`` under
        ``settings.use_case_dir``.
    """
    ontology = str(settings.ontology_dir)
    use_case = str(settings.use_case_dir)
    return cls(
        data_path=os.path.join(use_case, "compass.ttl"),
        shapes_path=os.path.join(ontology, "shapes.ttl"),
        vocab_path=os.path.join(use_case, "vocab.ttl"),
    )

instance() classmethod

Return the single live store, creating it from settings if needed.

Returns:

Type	Description
RDFStore	Process-wide RDFStore singleton.

Source code in src/backend/app/rdf.py

@classmethod
def instance(cls) -> RDFStore:
    """Return the single live store, creating it from settings if needed.

    Returns:
        Process-wide ``RDFStore`` singleton.
    """
    if cls._instance is None:
        cls._instance = cls.from_settings()
    return cls._instance

reload_instance() classmethod

Swap in the files currently on disk, keeping the live store on failure.

The candidate is built and validated in full before _instance moves, so a bad edit leaves the last good version serving rather than taking the API down with it.

Returns:

Type	Description
dict[str, Any]	Status dict with reloaded, source, and
dict[str, Any]	replaced_a_running_store.

Raises:

Type	Description
ReloadError	When the on-disk ontology is not usable.

Source code in src/backend/app/rdf.py

@classmethod
def reload_instance(cls) -> dict[str, Any]:
    """Swap in the files currently on disk, keeping the live store on failure.

    The candidate is built and validated in full before ``_instance`` moves,
    so a bad edit leaves the last good version serving rather than taking the
    API down with it.

    Returns:
        Status dict with ``reloaded``, ``source``, and
        ``replaced_a_running_store``.

    Raises:
        ReloadError: When the on-disk ontology is not usable.
    """
    try:
        candidate = cls.from_settings()
        candidate.validate()
    except ReloadError:
        raise
    except Exception as exc:
        raise ReloadError(f"{type(exc).__name__}: {exc}") from exc

    previous = cls._instance
    cls._instance = candidate
    logger.info("ontology reloaded from %s", settings.use_case_dir)
    return {
        "reloaded": True,
        "source": str(settings.use_case_dir),
        "replaced_a_running_store": previous is not None,
    }