ontocast.tool.facts_validation.shacl

SHACL execution, shape assembly, autofix repairs, and the catalog lint.

shacl_catalog_contradictions cross-checks the shapes against the term validator: a property the shapes require but the validator would flag as unknown silently destroys extracted data.

Attributes

logger = logging.getLogger(__name__) module-attribute

Classes

ShaclRepairResult

Bases: BaseModel

Outcome of the LLM-free SHACL repair pass.

Source code in ontocast/tool/facts_validation/shacl.py

class ShaclRepairResult(BaseModel):
    """Outcome of the LLM-free SHACL repair pass."""

    model_config = {"arbitrary_types_allowed": True}

    graph: RDFGraph
    records: list[GraphRepairRecord] = Field(default_factory=list)
    violations_before: int = 0
    violations_after: int = 0
    passes_applied: int = 0
    reverted: bool = False
    ran: bool = False

Attributes

graph instance-attribute

model_config = {'arbitrary_types_allowed': True} class-attribute instance-attribute

passes_applied = 0 class-attribute instance-attribute

ran = False class-attribute instance-attribute

records = Field(default_factory=list) class-attribute instance-attribute

reverted = False class-attribute instance-attribute

violations_after = 0 class-attribute instance-attribute

violations_before = 0 class-attribute instance-attribute

ShaclViolation

Bases: BaseModel

One SHACL validation result, in the form the repair pass needs.

FactsValidationFinding is the reporting shape and deliberately flat; this keeps the RDF terms (focus node, path, offending value, constraint component) so a repair can act on them.

Source code in ontocast/tool/facts_validation/shacl.py

class ShaclViolation(BaseModel):
    """One SHACL validation result, in the form the repair pass needs.

    ``FactsValidationFinding`` is the reporting shape and deliberately flat;
    this keeps the RDF terms (focus node, path, offending value, constraint
    component) so a repair can act on them.
    """

    model_config = {"arbitrary_types_allowed": True}

    focus: Node | None = None
    path: URIRef | None = None
    value: Node | None = None
    component: URIRef | None = None
    # Node, not URIRef: the common authoring style is an inline
    # ``sh:property [ sh:path … ; sh:datatype … ]``, whose shape is a blank
    # node. Narrowing to URIRef dropped it and left every such violation
    # unrepairable. pyshacl reports the same BNode the shapes graph holds, so
    # the datatype lookup resolves.
    source_shape: Node | None = None
    severity: TypingLiteral["error", "warning"] = "error"
    message: str = "SHACL constraint violated."

    def as_finding(self) -> FactsValidationFinding:
        """Project onto the reported finding shape."""
        return FactsValidationFinding(
            kind=FactsValidationFindingKind.SHACL,
            severity=self.severity,
            message=self.message,
            subject=str(self.focus) if self.focus is not None else "",
            predicate=str(self.path) if self.path is not None else "",
            values=[str(self.value)] if self.value is not None else [],
            component=str(self.component) if self.component is not None else "",
            source_shape=(
                str(self.source_shape) if self.source_shape is not None else ""
            ),
        )

Attributes

component = None class-attribute instance-attribute

focus = None class-attribute instance-attribute

message = 'SHACL constraint violated.' class-attribute instance-attribute

model_config = {'arbitrary_types_allowed': True} class-attribute instance-attribute

path = None class-attribute instance-attribute

severity = 'error' class-attribute instance-attribute

source_shape = None class-attribute instance-attribute

value = None class-attribute instance-attribute

Methods:

as_finding()

Project onto the reported finding shape.

Source code in ontocast/tool/facts_validation/shacl.py

def as_finding(self) -> FactsValidationFinding:
    """Project onto the reported finding shape."""
    return FactsValidationFinding(
        kind=FactsValidationFindingKind.SHACL,
        severity=self.severity,
        message=self.message,
        subject=str(self.focus) if self.focus is not None else "",
        predicate=str(self.path) if self.path is not None else "",
        values=[str(self.value)] if self.value is not None else [],
        component=str(self.component) if self.component is not None else "",
        source_shape=(
            str(self.source_shape) if self.source_shape is not None else ""
        ),
    )

Functions:

apply_shacl_repairs(graph, shapes_graph, ontology_graph, *, mode='prune', passes=1, fact_namespaces=(), code_predicates=(), inference='rdfs', advanced=True, max_triples=0, initial_violations=None)

Repair SHACL violations in code, with no LLM round-trip.

Bounded validate -> repair -> revalidate loop. A pass is kept only when it strictly reduces the violation count: a repair that trades triples for no conformance gain is reverted, the same discipline the un-merge repair uses.

Repairs by constraint component

• sh:datatype: retype a literal that parses as the declared datatype ("2019"^^xsd:string -> "2019"^^xsd:gYear).
• sh:class / sh:nodeKind: replace a string literal with the one catalog IRI declaring it as a surface form (qudt:unit "meV" -> unit:MilliElectronVolt). Ambiguous forms are left reported.
• sh:minCount (mode prune only): drop a focus node that asserts nothing beyond rdf:type/rdfs:label and is referenced by at most one subject, together with that reference.

Everything else -- sh:maxCount (owned by the functional-violation and un-merge machinery), sh:not, sh:qualifiedValueShape, SPARQL constraints -- is reported, never repaired.

Parameters:

Name	Type	Description	Default
graph	RDFGraph	Aggregated facts graph, repaired in place: it may be oxigraph-backed and carry RDF 1.2 triple terms, which a copied rdflib graph would silently drop. A pass that fails the accept test is rolled back triple-for-triple instead.	required
shapes_graph	RDFGraph | None	Shapes to validate against; None disables the pass.	required
ontology_graph	RDFGraph | None	Merged ontology context, indexed for surface forms.	required
mode	str	off | rewrite (rewrites only) | prune (also prunes).	'prune'
passes	int	Maximum repair rounds.	1
fact_namespaces	Sequence[str]	Only nodes under these namespaces are repaired.	()
code_predicates	Sequence[str]	Code-bearing predicates for surface resolution.	()
inference	str	pyshacl pre-inference mode.	'rdfs'
advanced	bool	Enable SHACL Advanced Features.	True
max_triples	int	Skip validation above this graph size; 0 disables.	0
initial_violations	Sequence[ShaclViolation] | None	Violations already computed for graph with the same parameters (e.g. by the reporting pass), reused to skip the redundant first validation.	None

Returns:

Type	Description
ShaclRepairResult	The repaired graph, the applied repair records, and fact-scoped
ShaclRepairResult	violation counts before and after (the population conforms is
ShaclRepairResult	judged on; the loop's accept test uses the raw count internally).

Source code in ontocast/tool/facts_validation/shacl.py

def apply_shacl_repairs(
    graph: RDFGraph,
    shapes_graph: RDFGraph | None,
    ontology_graph: RDFGraph | None,
    *,
    mode: str = "prune",
    passes: int = 1,
    fact_namespaces: Sequence[str] = (),
    code_predicates: Sequence[str] = (),
    inference: str = "rdfs",
    advanced: bool = True,
    max_triples: int = 0,
    initial_violations: Sequence[ShaclViolation] | None = None,
) -> ShaclRepairResult:
    """Repair SHACL violations in code, with no LLM round-trip.

    Bounded ``validate -> repair -> revalidate`` loop. A pass is kept only when
    it strictly reduces the violation count: a repair that trades triples for
    no conformance gain is reverted, the same discipline the un-merge repair
    uses.

    Repairs by constraint component:
        - ``sh:datatype``: retype a literal that parses as the declared
          datatype (``"2019"^^xsd:string`` -> ``"2019"^^xsd:gYear``).
        - ``sh:class`` / ``sh:nodeKind``: replace a string literal with the one
          catalog IRI declaring it as a surface form (``qudt:unit "meV"`` ->
          ``unit:MilliElectronVolt``). Ambiguous forms are left reported.
        - ``sh:minCount`` (mode ``prune`` only): drop a focus node that asserts
          nothing beyond ``rdf:type``/``rdfs:label`` and is referenced by at
          most one subject, together with that reference.

    Everything else -- ``sh:maxCount`` (owned by the functional-violation and
    un-merge machinery), ``sh:not``, ``sh:qualifiedValueShape``, SPARQL
    constraints -- is reported, never repaired.

    Args:
        graph: Aggregated facts graph, repaired **in place**: it may be
            oxigraph-backed and carry RDF 1.2 triple terms, which a copied
            rdflib graph would silently drop. A pass that fails the accept
            test is rolled back triple-for-triple instead.
        shapes_graph: Shapes to validate against; ``None`` disables the pass.
        ontology_graph: Merged ontology context, indexed for surface forms.
        mode: ``off`` | ``rewrite`` (rewrites only) | ``prune`` (also prunes).
        passes: Maximum repair rounds.
        fact_namespaces: Only nodes under these namespaces are repaired.
        code_predicates: Code-bearing predicates for surface resolution.
        inference: pyshacl pre-inference mode.
        advanced: Enable SHACL Advanced Features.
        max_triples: Skip validation above this graph size; 0 disables.
        initial_violations: Violations already computed for ``graph`` with the
            same parameters (e.g. by the reporting pass), reused to skip the
            redundant first validation.

    Returns:
        The repaired graph, the applied repair records, and fact-scoped
        violation counts before and after (the population ``conforms`` is
        judged on; the loop's accept test uses the raw count internally).
    """
    if mode == "off" or shapes_graph is None or not len(shapes_graph) or passes <= 0:
        return ShaclRepairResult(graph=graph)

    def _validate(target: RDFGraph) -> list[ShaclViolation] | None:
        return run_shacl(
            target,
            shapes_graph,
            ontology_graph=ontology_graph,
            inference=inference,
            advanced=advanced,
            max_triples=max_triples,
        )

    violations = (
        list(initial_violations) if initial_violations is not None else _validate(graph)
    )
    if violations is None:
        return ShaclRepairResult(graph=graph)

    def _scoped_count(candidates: Sequence[ShaclViolation]) -> int:
        return len(_fact_scope_violations(graph, candidates, fact_namespaces))

    result = ShaclRepairResult(
        graph=graph,
        violations_before=_scoped_count(violations),
        violations_after=_scoped_count(violations),
        ran=True,
    )
    surface_index = build_surface_index(ontology_graph, code_predicates)

    def _rollback(added: Sequence[tuple], removed: Sequence[tuple]) -> None:
        for triple in added:
            graph.remove(triple)
        for triple in removed:
            graph.add(triple)

    for _ in range(passes):
        if not violations:
            break
        plan = _shacl_repairs_for(
            graph,
            shapes_graph,
            violations,
            mode=mode,
            surface_index=surface_index,
            fact_namespaces=fact_namespaces,
        )
        records = plan.records
        if not records:
            break

        applied_removals: list[tuple] = []
        applied_removal_set: set[tuple] = set()
        seen_removals: set[tuple] = set()
        for triple in plan.removals:
            if triple in seen_removals:
                continue
            seen_removals.add(triple)
            if triple in graph:
                graph.remove(triple)
                applied_removals.append(triple)
                applied_removal_set.add(triple)
        applied_additions: list[tuple] = []
        for triple in plan.additions:
            if triple not in graph:
                graph.add(triple)
                applied_additions.append(triple)

        candidate_violations = _validate(graph)
        if candidate_violations is None:
            _rollback(applied_additions, applied_removals)
            break
        if len(candidate_violations) >= len(violations):
            logger.warning(
                "SHACL autofix: pass did not reduce violations (%d -> %d); "
                "keeping the pre-repair graph",
                len(violations),
                len(candidate_violations),
            )
            _rollback(applied_additions, applied_removals)
            result.reverted = True
            break

        # Only once the pass is accepted, and deliberately after the accept test
        # rather than alongside the removals: reification quads cannot travel
        # through _rollback (rdflib cannot add or remove a triple-term triple),
        # and validation runs on a copy with triple terms stripped, so this
        # changes no count and needs no undo.
        #
        # Retarget before sweeping. A statement that is retyped *and* then
        # pruned in the same pass has to be swept at its new triple term, which
        # the retarget has already installed -- so prune still wins, by the
        # sweep matching rather than by the retarget happening to miss.
        retargeted = retarget_reifiers(
            graph,
            {
                removed: replacement
                for removed, replacement in plan.retargets.items()
                if removed in applied_removal_set and replacement in graph
            },
        )
        swept = drop_reifiers_mentioning(graph, plan.pruned)

        provenance_note = ", ".join(
            note
            for note in (
                f"{retargeted} provenance quad(s) retargeted" if retargeted else "",
                f"{swept} orphaned provenance quad(s) swept" if swept else "",
            )
            if note
        )
        logger.info(
            "SHACL autofix: %d repair(s) applied, violations %d -> %d%s",
            len(records),
            len(violations),
            len(candidate_violations),
            f", {provenance_note}" if provenance_note else "",
        )
        violations = candidate_violations
        result.records.extend(records)
        result.passes_applied += 1
        result.violations_after = _scoped_count(candidate_violations)

    return result

collect_shacl_shapes(ontology_graph, stored_shapes)

Assemble the SHACL shapes graph for the validation gate.

Sources: the deployment's shapes partition (stored_shapes, resolved by :class:~ontocast.tool.shapes_catalog.ShapesCatalog -- seeded from FACTS_SHAPES_DIR and mutable over /shapes), plus the ontology context itself when it already carries sh:NodeShape declarations inline -- the zero-config path for catalogs that ship shapes next to their schema.

Parameters:

Name	Type	Description	Default
ontology_graph	RDFGraph | None	Ontology context offered to the renderer.	required
stored_shapes	RDFGraph | None	Merged shapes graph from the shapes partition.	required

Returns:

Type	Description
RDFGraph | None	RDFGraph | None: The shapes to validate against, or None when there
RDFGraph | None	are none -- which is what keeps shacl_evaluated at None
RDFGraph | None	("never checked") rather than reporting a clean run.

Source code in ontocast/tool/facts_validation/shacl.py

def collect_shacl_shapes(
    ontology_graph: RDFGraph | None, stored_shapes: RDFGraph | None
) -> RDFGraph | None:
    """Assemble the SHACL shapes graph for the validation gate.

    Sources: the deployment's shapes partition (``stored_shapes``, resolved by
    :class:`~ontocast.tool.shapes_catalog.ShapesCatalog` -- seeded from
    ``FACTS_SHAPES_DIR`` and mutable over ``/shapes``), plus the ontology
    context itself when it already carries ``sh:NodeShape`` declarations inline
    -- the zero-config path for catalogs that ship shapes next to their schema.

    Args:
        ontology_graph: Ontology context offered to the renderer.
        stored_shapes: Merged shapes graph from the shapes partition.

    Returns:
        RDFGraph | None: The shapes to validate against, or ``None`` when there
        are none -- which is what keeps ``shacl_evaluated`` at ``None``
        ("never checked") rather than reporting a clean run.
    """
    shapes = RDFGraph()
    if stored_shapes is not None and len(stored_shapes):
        shapes += stored_shapes
    node_shape = SH.NodeShape
    if ontology_graph is not None and (None, RDF.type, node_shape) in ontology_graph:
        shapes += ontology_graph
    return shapes if len(shapes) else None

run_shacl(graph, shapes_graph, *, ontology_graph=None, inference='rdfs', advanced=True, max_triples=0)

Validate graph against shapes_graph, returning the violations.

Reaching here means shapes were found, so the caller expects validation to happen: a missing extra or a skipped run is reported at warning level, not debug. Silently returning "no violations" is indistinguishable from "conforms", so those cases return None.

The ontology context is mixed in (ont_graph) rather than left out. A facts graph states that a value uses unit:DAY; that the individual is a qudt:Unit is stated only in the catalog. Validating the facts alone therefore fails every sh:class constraint pointing at a catalog individual — violations that describe the missing schema, not the data.

RDFS inference is the default for the same reason. SHACL resolves class targets through rdfs:subClassOf on its own, but property paths carry no entailment: a shape on obs:hasResult does not see the life:hasStorageResult the renderer emitted, and reports the more specific statement as a missing one, so turning inference off raises the violation count rather than lowering it.

Parameters:

Name	Type	Description	Default
graph	RDFGraph	Data graph to validate.	required
shapes_graph	RDFGraph	Shapes to validate against.	required
ontology_graph	RDFGraph | None	Schema mixed into the data graph for validation.	None
inference	str	pyshacl pre-inference (none / rdfs / owlrl).	'rdfs'
advanced	bool	Enable SHACL Advanced Features.	True
max_triples	int	Skip validation above this graph size; 0 disables.	0

Returns:

Type	Description
list[ShaclViolation] | None	Violations in report order, or None when validation did not run.

Source code in ontocast/tool/facts_validation/shacl.py

def run_shacl(
    graph: RDFGraph,
    shapes_graph: RDFGraph,
    *,
    ontology_graph: RDFGraph | None = None,
    inference: str = "rdfs",
    advanced: bool = True,
    max_triples: int = 0,
) -> list[ShaclViolation] | None:
    """Validate ``graph`` against ``shapes_graph``, returning the violations.

    Reaching here means shapes were found, so the caller expects validation to
    happen: a missing extra or a skipped run is reported at warning level, not
    debug. Silently returning "no violations" is indistinguishable from
    "conforms", so those cases return ``None``.

    The ontology context is mixed in (``ont_graph``) rather than left out. A
    facts graph states that a value uses ``unit:DAY``; that the individual *is*
    a ``qudt:Unit`` is stated only in the catalog. Validating the facts alone
    therefore fails every ``sh:class`` constraint pointing at a catalog
    individual — violations that describe the missing schema, not the data.

    RDFS inference is the default for the same reason. SHACL resolves class
    targets through ``rdfs:subClassOf`` on its own, but property paths carry no
    entailment: a shape on ``obs:hasResult`` does not see the
    ``life:hasStorageResult`` the renderer emitted, and reports the more
    specific statement as a missing one, so turning inference off raises the
    violation count rather than lowering it.

    Args:
        graph: Data graph to validate.
        shapes_graph: Shapes to validate against.
        ontology_graph: Schema mixed into the data graph for validation.
        inference: pyshacl pre-inference (``none`` / ``rdfs`` / ``owlrl``).
        advanced: Enable SHACL Advanced Features.
        max_triples: Skip validation above this graph size; 0 disables.

    Returns:
        Violations in report order, or ``None`` when validation did not run.
    """
    try:
        import pyshacl
    except ImportError:
        logger.warning(
            "SHACL shapes are configured but pyshacl is not installed; "
            "skipping SHACL validation. Install the extra: uv sync --extra shacl"
        )
        return None

    if max_triples and len(graph) > max_triples:
        logger.warning(
            "Skipping SHACL validation: %d triples exceeds "
            "FACTS_SHACL_MAX_TRIPLES=%d. The graph is unvalidated, not conformant.",
            len(graph),
            max_triples,
        )
        return None

    # pyshacl clones and mixes the data graph through plain rdflib graphs,
    # which cannot hold the RDF 1.2 triple terms an oxigraph-backed aggregated
    # graph carries (rdflib ``Graph.add`` asserts on them). Hand pyshacl a
    # sanitised copy; the dropped reification provenance carries no shape
    # targets, so validation loses nothing.
    data_graph = RDFGraph()
    copy_triples(graph, data_graph, origin="run_shacl")
    for prefix, namespace in graph.namespaces():
        data_graph.bind(prefix, namespace, override=True)

    conforms, results_graph, _ = pyshacl.validate(
        data_graph,
        shacl_graph=shapes_graph,
        ont_graph=(
            ontology_graph
            if ontology_graph is not None and len(ontology_graph)
            else None
        ),
        inference=inference,
        advanced=advanced,
        abort_on_first=False,
    )
    if conforms:
        return []

    violations: list[ShaclViolation] = []
    for result in results_graph.subjects(RDF.type, SH.ValidationResult):
        severity_iri = results_graph.value(result, SH.resultSeverity)
        message = results_graph.value(result, SH.resultMessage)
        path = results_graph.value(result, SH.resultPath)
        component = results_graph.value(result, SH.sourceConstraintComponent)
        source_shape = results_graph.value(result, SH.sourceShape)
        violations.append(
            ShaclViolation(
                focus=results_graph.value(result, SH.focusNode),
                path=path if isinstance(path, URIRef) else None,
                value=results_graph.value(result, SH.value),
                component=component if isinstance(component, URIRef) else None,
                source_shape=source_shape,
                severity=("error" if severity_iri == SH.Violation else "warning"),
                message=str(message) if message else "SHACL constraint violated.",
            )
        )
    return violations

shacl_catalog_contradictions(shapes_graph, ontology_graph, *, policy=None, catalog_terms=None)

Property paths the shapes require but the unit validator would flag.

A SHACL property shape with sh:minCount >= 1 demands a property that the deterministic UNKNOWN_TERM check — same closure rules, same exemptions — would report as not existing. Data cannot satisfy both: the renderer is ordered to remove exactly what validation requires. Found live in practice, where shapes required qudt:numericValue while the validator's mandatory findings drove repair renders to delete it. Callers log the returned IRIs as configuration errors.

Parameters:

Name	Type	Description	Default
shapes_graph	RDFGraph | None	The shapes the gate validates against.	required
ontology_graph	RDFGraph | None	The ontology context the facts were rendered against.	required
policy	ValidationPolicy | None	The exemptions the unit validator applies. Pass the same policy the unit loop uses -- including its shapes-contract exemptions -- or the check reports contradictions the validator never raises.	None
catalog_terms	Collection[str] | None	The whole catalog's term inventory, when ontology_graph is only a retrieved subset of it. The unit validator judges unknown terms against the whole catalog, so a term the snapshot omitted is not a contradiction.	None

Source code in ontocast/tool/facts_validation/shacl.py

def shacl_catalog_contradictions(
    shapes_graph: RDFGraph | None,
    ontology_graph: RDFGraph | None,
    *,
    policy: ValidationPolicy | None = None,
    catalog_terms: Collection[str] | None = None,
) -> list[str]:
    """Property paths the shapes require but the unit validator would flag.

    A SHACL property shape with ``sh:minCount >= 1`` demands a property that
    the deterministic UNKNOWN_TERM check — same closure rules, same
    exemptions — would report as not existing. Data cannot satisfy both: the
    renderer is ordered to remove exactly what validation requires. Found live
    in practice, where shapes required ``qudt:numericValue``
    while the validator's mandatory findings drove repair renders to delete
    it. Callers log the returned IRIs as configuration errors.

    Args:
        shapes_graph: The shapes the gate validates against.
        ontology_graph: The ontology context the facts were rendered against.
        policy: The exemptions the unit validator applies. Pass the same
            policy the unit loop uses -- including its shapes-contract
            exemptions -- or the check reports contradictions the validator
            never raises.
        catalog_terms: The whole catalog's term inventory, when
            ``ontology_graph`` is only a retrieved subset of it. The unit
            validator judges unknown terms against the whole catalog, so a
            term the snapshot omitted is not a contradiction.
    """
    if shapes_graph is None or ontology_graph is None:
        return []
    catalog_terms = set(collect_catalog_terms(ontology_graph)) | set(
        catalog_terms or ()
    )
    if not catalog_terms:
        return []
    policy = policy or ValidationPolicy()
    declared_namespaces = collect_declared_namespaces(ontology_graph)
    standard_namespaces = policy.standard_namespaces()
    fallback_terms = policy.exempt_terms(shapes_graph, ontology_graph)
    required: set[str] = set()
    for shape, path in shapes_graph.subject_objects(SH.path):
        if not isinstance(path, URIRef):
            continue
        min_count = next(shapes_graph.objects(shape, SH.minCount), None)
        try:
            if min_count is None or int(str(min_count)) < 1:
                continue
        except ValueError:
            continue
        required.add(str(path))
    contradictions = [
        term
        for term in sorted(required)
        if _namespace_of(term) in declared_namespaces
        and term not in catalog_terms
        and term not in fallback_terms
        and not _namespace_of(term).startswith(standard_namespaces)
    ]
    return contradictions