orchestrata_ultimate.domaingen.reader

Reference › orchestrata_ultimate › domaingen › reader

reader reads an OWL or Turtle ontology into the domain-model IR (SPEC-105 FR-001) using rdflib. parse_ontology accepts either a file path or raw text and returns a DomainModel from models. _parse_graph loads the source into a graph. _build_entities creates an Entity per class IRI, capturing its documentation and parent (REQ-105-019). _attach_datatype_properties and _attach_object_properties attach OWL datatype properties as fields and object properties as relationships on their domain class. _apply_restrictions marks fields and relationships as required where an OWL restriction sets a minimum cardinality (REQ-105-017), using _restriction_required_prop. _namespace_of derives the common namespace of the class IRIs. Entities, fields and relationships are sorted by name so the IR, and everything generated from it, is stable. An ontology that cannot be parsed raises OntologyParseError (ERR-105-001). The package imports it, and its output feeds generator and ttl.

At a glance

Class diagram

Dependency diagram

API

Classes

OntologyParseError

Bases: BaseError · Decorators: traces(spec='105', story='US-105-001', reqs=['REQ-105-024']) · Line 30

Raised when an ontology cannot be parsed (ERR-105-001).

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/domaingen/reader.py

"""Ontology (OWL / Turtle) -> DomainModel IR via rdflib (SPEC-105 FR-001).

Deterministic: entities, fields and relationships are sorted by name so the IR
(and everything generated from it) is stable.
"""

from __future__ import annotations

import logging
from pathlib import Path

from rdflib import OWL, RDF, RDFS, Graph, URIRef
from rdflib.term import Node

from orchestrata_ultimate.core.base.error import BaseError
from orchestrata_ultimate.core.traceability import traces
from orchestrata_ultimate.domaingen.models import (
    XSD_TO_PY,
    DomainModel,
    Entity,
    Field,
    Module,
    Relationship,
)

logger = logging.getLogger(__name__)


@traces(spec="105", story="US-105-001", reqs=["REQ-105-024"])
class OntologyParseError(BaseError):
    """Raised when an ontology cannot be parsed (ERR-105-001)."""


@traces(spec="105", story="US-105-001", reqs=["REQ-105-020"])
def _local(iri: object) -> str:
    s = str(iri)
    if "#" in s:
        return s.rsplit("#", 1)[1]
    return s.rstrip("/").rsplit("/", 1)[-1]


@traces(spec="105", story="US-105-001", reqs=["REQ-105-023"])
def _namespace_of(iris: list[str]) -> str:
    """Common namespace prefix (up to '#' or final '/') of the class IRIs."""
    for s in iris:
        if "#" in s:
            return s.rsplit("#", 1)[0] + "#"
    if iris:
        return iris[0].rstrip("/").rsplit("/", 1)[0] + "/"
    return "https://example.org/domain#"


@traces(spec="105", story="US-105-001", reqs=["REQ-105-017"])
def _restriction_required_prop(graph: Graph, restr: Node) -> str | None:
    """Return the property local-name a required owl:Restriction targets, else None.

    A restriction marks a property required when it declares minCardinality (or
    cardinality) >= 1 on an onProperty. Returns ``None`` for anything else
    (not a Restriction, missing onProperty/cardinality, non-numeric, or < 1).
    """
    if (restr, RDF.type, OWL.Restriction) not in graph:
        return None
    on_prop = graph.value(restr, OWL.onProperty)
    card = graph.value(restr, OWL.minCardinality) or graph.value(restr, OWL.cardinality)
    if on_prop is None or card is None:
        return None
    try:
        # card is a Node (typically an rdflib Literal, which is a str
        # subclass at runtime); str() gets its lexical form for int().
        required = int(str(card)) >= 1
    except TypeError, ValueError:
        return None
    return _local(on_prop) if required else None


@traces(spec="105", story="US-105-001", reqs=["REQ-105-017"])
def _apply_restrictions(graph: Graph, by_iri: dict[URIRef, Entity]) -> None:
    """Mark fields/relationships required from owl:Restriction minCardinality (REQ-105-017)."""
    for cls, ent in by_iri.items():
        for restr in graph.objects(cls, RDFS.subClassOf):
            pname = _restriction_required_prop(graph, restr)
            if pname is None:
                continue
            for fld in ent.fields:
                if fld.name == pname:
                    fld.required = True
            for rel in ent.relationships:
                if rel.name == pname:
                    rel.required = True


@traces(
    spec="105", story="US-105-001", reqs=["REQ-105-012", "REQ-105-013", "REQ-105-024"]
)
def _parse_graph(source: str | Path) -> Graph:
    """Parse an OWL/Turtle source (file path or raw text) into an rdflib Graph."""
    graph = Graph()
    src = str(source)
    try:
        if isinstance(source, Path) or (len(src) < 4096 and Path(src).exists()):
            graph.parse(str(source), format="turtle")
        else:
            graph.parse(data=src, format="turtle")
    except OntologyParseError:
        raise
    except Exception as exc:  # rdflib raises a variety of parse errors
        raise OntologyParseError(f"failed to parse ontology: {exc}") from exc
    return graph


@traces(
    spec="105", story="US-105-001", reqs=["REQ-105-001", "REQ-105-016", "REQ-105-019"]
)
def _build_entities(graph: Graph, classes: set[URIRef]) -> dict[URIRef, Entity]:
    """Build an Entity per class IRI, capturing doc and parent (REQ-105-019)."""
    by_iri: dict[URIRef, Entity] = {}
    for cls in classes:
        ent = Entity(name=_local(cls), iri=str(cls))
        # doc: prefer rdfs:comment, fall back to rdfs:label (REQ-105-019). The
        # class identifier stays the IRI local name (identifier-safe).
        comment = graph.value(cls, RDFS.comment)
        label = graph.value(cls, RDFS.label)
        if comment is not None:
            ent.doc = str(comment)
        elif label is not None:
            ent.doc = str(label)
        parent = graph.value(cls, RDFS.subClassOf)
        if isinstance(parent, URIRef) and parent in classes:
            ent.parent = _local(parent)
        by_iri[cls] = ent
    return by_iri


@traces(spec="105", story="US-105-001", reqs=["REQ-105-014", "REQ-105-018"])
def _attach_datatype_properties(graph: Graph, by_iri: dict[URIRef, Entity]) -> None:
    """Attach owl:DatatypeProperty declarations as fields on their domain entity."""
    for prop in graph.subjects(RDF.type, OWL.DatatypeProperty):
        domain = graph.value(prop, RDFS.domain)
        if not isinstance(domain, URIRef) or domain not in by_iri:
            continue
        rng = graph.value(prop, RDFS.range)
        pytype = XSD_TO_PY.get(_local(rng), "str") if rng is not None else "str"
        if rng is not None and _local(rng) not in XSD_TO_PY:
            logger.warning("Unknown datatype %s -> str", rng)
        by_iri[domain].fields.append(
            Field(name=_local(prop), type=pytype, iri=str(prop))
        )


@traces(spec="105", story="US-105-001", reqs=["REQ-105-015", "REQ-105-025"])
def _attach_object_properties(graph: Graph, by_iri: dict[URIRef, Entity]) -> None:
    """Attach owl:ObjectProperty declarations as relationships on their domain."""
    for prop in graph.subjects(RDF.type, OWL.ObjectProperty):
        domain = graph.value(prop, RDFS.domain)
        if not isinstance(domain, URIRef) or domain not in by_iri:
            continue
        rng = graph.value(prop, RDFS.range)
        if not isinstance(rng, URIRef):  # ERR-105-002
            logger.warning("Object property %s has no range — skipped", prop)
            continue
        functional = (prop, RDF.type, OWL.FunctionalProperty) in graph
        inverse = graph.value(prop, OWL.inverseOf)
        by_iri[domain].relationships.append(
            Relationship(
                name=_local(prop),
                target=_local(rng),
                iri=str(prop),
                many=not functional,
                inverse=_local(inverse) if inverse is not None else None,
            )
        )


@traces(
    spec="105",
    story="US-105-001",
    reqs=[
        "REQ-105-001",
        "REQ-105-004",
        "REQ-105-012",
        "REQ-105-013",
        "REQ-105-015",
        "REQ-105-016",
        "REQ-105-017",
        "REQ-105-021",
        "REQ-105-023",
        "REQ-105-025",
    ],
)
def parse_ontology(source: str | Path) -> DomainModel:
    """Parse an OWL/Turtle ontology (file path or raw text) into a DomainModel."""
    graph = _parse_graph(source)

    classes = {
        c
        for c in set(graph.subjects(RDF.type, OWL.Class))
        | set(graph.subjects(RDF.type, RDFS.Class))
        if isinstance(c, URIRef)
    }
    by_iri = _build_entities(graph, classes)
    _attach_datatype_properties(graph, by_iri)
    _attach_object_properties(graph, by_iri)
    _apply_restrictions(graph, by_iri)

    namespace = _namespace_of(sorted(str(c) for c in classes))
    entities = sorted(by_iri.values(), key=lambda e: e.name)
    for ent in entities:
        ent.fields.sort(key=lambda f: f.name)
        ent.relationships.sort(key=lambda r: r.name)
    name = namespace.rstrip("#/").rsplit("/", 1)[-1] or "domain"
    return DomainModel(
        name=name, namespace=namespace, modules=[Module(name=name, entities=entities)]
    )