ontocast.onto.sparql_models

Pydantic models for graph mutations.

GraphUpdate / TripleOp are the canonical LLM pipeline mutation abstraction (ordered insert/delete triple patches) used across the extraction pipeline.

Attributes

STANDARD_PREFIXES = {prefix: uri.strip('<>') for prefix, uri in COMMON_PREFIXES.items()} module-attribute

logger = logging.getLogger(__name__) module-attribute

Classes

GraphUpdate

Bases: BaseModel

Structured RDF graph patches for LLM pipeline output.

Each TripleOp in triple_operations is executed in order. SPARQL compilation for rdflib apply happens internally via generate_sparql_queries().

Source code in ontocast/onto/sparql_models.py

class GraphUpdate(BaseModel):
    """Structured RDF graph patches for LLM pipeline output.

    Each ``TripleOp`` in ``triple_operations`` is executed in order. SPARQL compilation
    for rdflib apply happens internally via ``generate_sparql_queries()``.
    """

    triple_operations: list[TripleOp] = Field(
        default_factory=list,
        description="List of graph update operations in execution order. "
        "Each operation should be a TripleOp (insert/delete) with graph encoding "
        "per deployment llm_graph_format and OUTPUT INSTRUCTION.",
    )

    def generate_sparql_queries(self) -> list[str]:
        """Compile triple_operations to SPARQL UPDATE strings for rdflib execution.

        Returns:
            List of SPARQL query strings in operation order.
        """
        queries = []

        for op in self.triple_operations:
            if len(op.graph) > 0:
                prefixes = STANDARD_PREFIXES.copy()

                for prefix, uri in op.graph.namespaces():
                    if prefix:
                        prefixes[prefix] = str(uri)

                prefixes.update(op.prefixes)

                if prefixes:
                    prefix_declarations = []
                    for prefix, uri in prefixes.items():
                        prefix_declarations.append(f"PREFIX {prefix}: <{uri}>")
                    prefix_block = "\n".join(prefix_declarations)
                else:
                    prefix_block = ""

                declared = frozenset(prefixes)
                if op.type == "insert":
                    triple_query = self._generate_insert_query(
                        op.graph, prefix_block, declared
                    )
                else:
                    triple_query = self._generate_delete_query(
                        op.graph, prefix_block, declared
                    )
                queries.append(triple_query)

        return queries

    def count_total_triples(self) -> tuple[int, int]:
        """Count total triples across all operations.

        Returns:
            Tuple of (total_operations, total_triples) where:
            - total_operations: Number of operations
            - total_triples: Total number of triples across all TripleOp operations
        """
        total_triples = sum(len(op.graph) for op in self.triple_operations)
        return (len(self.triple_operations), total_triples)

    def extract_insert_graph(self) -> RDFGraph:
        """Extract RDFGraph of all insert triples from triple_operations.

        Returns:
            RDFGraph containing the union of all insert triples.
        """
        result = RDFGraph()
        for op in self.triple_operations:
            if op.type == "insert" and len(op.graph) > 0:
                copy_triples(
                    op.graph, result, origin="GraphUpdate.extract_insert_graph"
                )
                for prefix, uri in op.graph.namespaces():
                    if prefix:
                        result.bind(prefix, uri)
                for prefix, uri in op.prefixes.items():
                    result.bind(prefix, uri)
        return result

    def generate_diff_summary(self) -> str:
        """Generate a human-readable diff summary of all operations for LLM consumption.

        Returns:
            String representation of all operations showing what will be added, removed, and modified.
            Returns empty string if no operations to perform.
        """
        if not self.triple_operations:
            return ""

        diff_parts = []
        operation_count = 0

        for i, op in enumerate(self.triple_operations, 1):
            if len(op.graph) > 0:
                op_type = op.type.upper()
                diff_parts.append(f"{i}. {op_type} {len(op.graph)} triple(s):")

                graph_prefixes = {
                    prefix: str(uri) for prefix, uri in op.graph.namespaces() if prefix
                }
                all_prefixes = {**graph_prefixes, **op.prefixes}
                if all_prefixes:
                    prefix_list = ", ".join(
                        [f"{k}: {v}" for k, v in all_prefixes.items()]
                    )
                    diff_parts.append(f"   Prefixes: {prefix_list}")

                # Mirror the prefix set generate_sparql_queries declares, so the
                # summary abbreviates exactly what the emitted query does.
                declared = frozenset(STANDARD_PREFIXES) | frozenset(all_prefixes)
                for subject, predicate, obj in self._serializable_triples(op.graph):
                    symbol = "+" if op.type == "insert" else "-"
                    diff_parts.append(
                        f"   {symbol} {self._serialize_rdf_term(subject, declared)} {self._serialize_rdf_term(predicate, declared)} {self._serialize_rdf_term(obj, declared)}"
                    )
                operation_count += 1

        if operation_count == 0:
            return ""

        summary = f"Ontology Update Summary ({operation_count} operation(s)):\n\n"
        summary += "\n".join(diff_parts)

        return summary

    @staticmethod
    def _serializable_triples(graph: RDFGraph) -> list[tuple[Node, Node, Node]]:
        """Return the triples of ``graph`` that SPARQL can express.

        Oxigraph-backed graphs yield RDF 1.2 triple terms as plain tuples, which
        have no SPARQL syntax. Dropping them keeps the rest of the update valid;
        emitting them produced a Python repr inside the query and a
        ``ParseException`` at apply time.
        """
        triples: list[tuple[Node, Node, Node]] = []
        dropped = 0
        for triple in graph:
            if not is_rdflib_triple(triple):
                dropped += 1
                continue
            triples.append(triple)
        if dropped:
            logger.warning(
                "Skipped %d RDF 1.2 triple-term triple(s): not expressible in SPARQL",
                dropped,
            )
        return triples

    def _generate_insert_query(
        self, graph: RDFGraph, prefix_block: str, prefixes: Set[str] = frozenset()
    ) -> str:
        """Generate a SPARQL INSERT query for the given RDFGraph."""
        if len(graph) == 0:
            return ""

        triple_patterns = []
        for subject, predicate, obj in self._serializable_triples(graph):
            triple_patterns.append(
                f"    {self._serialize_rdf_term(subject, prefixes)} {self._serialize_rdf_term(predicate, prefixes)} {self._serialize_rdf_term(obj, prefixes)} ."
            )
        if not triple_patterns:
            return ""

        triples_block = "\n".join(triple_patterns)

        query_parts = []
        if prefix_block:
            query_parts.append(prefix_block)
        query_parts.append("INSERT DATA {")
        query_parts.append(triples_block)
        query_parts.append("}")

        return "\n".join(query_parts)

    def _generate_delete_query(
        self, graph: RDFGraph, prefix_block: str, prefixes: Set[str] = frozenset()
    ) -> str:
        """Generate a SPARQL DELETE query for the given RDFGraph."""
        if len(graph) == 0:
            return ""

        triple_patterns = []
        for subject, predicate, obj in self._serializable_triples(graph):
            triple_patterns.append(
                f"    {self._serialize_rdf_term(subject, prefixes)} {self._serialize_rdf_term(predicate, prefixes)} {self._serialize_rdf_term(obj, prefixes)} ."
            )
        if not triple_patterns:
            return ""

        triples_block = "\n".join(triple_patterns)

        query_parts = []
        if prefix_block:
            query_parts.append(prefix_block)
        query_parts.append("DELETE DATA {")
        query_parts.append(triples_block)
        query_parts.append("}")

        return "\n".join(query_parts)

    @staticmethod
    def _is_declared_prefixed_name(value: str, prefixes: Set[str]) -> bool:
        """True when ``value`` is a ``prefix:local`` name the query declares.

        The LLM emits some terms already abbreviated, and those must pass through
        unbracketed so the ``PREFIX`` block resolves them. Testing for a bare
        colon instead — as this did — also matched absolute IRIs in every
        non-``http`` scheme (``urn:``, ``doi:``, ``file:``, ``mailto:``), which
        then reached the parser as undefined prefixed names. Matching against the
        declared prefixes makes the distinction exact rather than heuristic.
        """
        prefix, separator, local = value.partition(":")
        if not separator or prefix not in prefixes:
            return False
        # A local part carrying IRI structure means the match was a scheme
        # collision, not an abbreviation.
        return local != "" and not any(char in local for char in ":/#[]?@")

    def _serialize_rdf_term(self, term: Node, prefixes: Set[str] = frozenset()) -> str:
        """Serialize an RDF term to its SPARQL string representation.

        Args:
            term: The term to serialize.
            prefixes: Prefix labels declared in the query's ``PREFIX`` block;
                only these are honoured as abbreviations.

        Raises:
            TypeError: If ``term`` is not a term SPARQL can express. Callers
                filter with :meth:`_serializable_triples` first; falling back to
                ``str(term)`` here used to bury a Python repr inside the query,
                which only surfaced as a ``ParseException`` at apply time.
        """
        if isinstance(term, URIRef):
            if self._is_declared_prefixed_name(str(term), prefixes):
                return str(term)
            return f"<{term}>"
        elif isinstance(term, BNode):
            return f"_:{term}"
        elif isinstance(term, Literal):
            # Interpolating into bare quotes, as this did, let any literal
            # containing a quote, backslash or newline -- routine in extracted
            # text -- close the string early and fail the whole update with a
            # ParseException at apply time.
            #
            # Escaped here rather than via ``Literal.n3()``: n3 emits a raw tab
            # inside the quoted form, and rdflib's own SPARQL parser reads that
            # back as spaces, so a tab-bearing literal round-tripped to a
            # different value. Every escape below is a SPARQL ECHAR.
            lexical = f'"{str(term).translate(_LITERAL_ESCAPES)}"'
            if term.language:
                return f"{lexical}@{term.language}"
            if term.datatype:
                return f"{lexical}^^<{term.datatype}>"
            return lexical
        else:
            raise TypeError(
                f"Cannot serialize {type(term).__name__} as a SPARQL term: {term!r}"
            )

Attributes

triple_operations = Field(default_factory=list, description='List of graph update operations in execution order. Each operation should be a TripleOp (insert/delete) with graph encoding per deployment llm_graph_format and OUTPUT INSTRUCTION.') class-attribute instance-attribute

Methods:

count_total_triples()

Count total triples across all operations.

Returns:

Type	Description
int	Tuple of (total_operations, total_triples) where:
int	total_operations: Number of operations
tuple[int, int]	total_triples: Total number of triples across all TripleOp operations

Source code in ontocast/onto/sparql_models.py

def count_total_triples(self) -> tuple[int, int]:
    """Count total triples across all operations.

    Returns:
        Tuple of (total_operations, total_triples) where:
        - total_operations: Number of operations
        - total_triples: Total number of triples across all TripleOp operations
    """
    total_triples = sum(len(op.graph) for op in self.triple_operations)
    return (len(self.triple_operations), total_triples)

extract_insert_graph()

Extract RDFGraph of all insert triples from triple_operations.

Returns:

Type	Description
RDFGraph	RDFGraph containing the union of all insert triples.

Source code in ontocast/onto/sparql_models.py

def extract_insert_graph(self) -> RDFGraph:
    """Extract RDFGraph of all insert triples from triple_operations.

    Returns:
        RDFGraph containing the union of all insert triples.
    """
    result = RDFGraph()
    for op in self.triple_operations:
        if op.type == "insert" and len(op.graph) > 0:
            copy_triples(
                op.graph, result, origin="GraphUpdate.extract_insert_graph"
            )
            for prefix, uri in op.graph.namespaces():
                if prefix:
                    result.bind(prefix, uri)
            for prefix, uri in op.prefixes.items():
                result.bind(prefix, uri)
    return result

generate_diff_summary()

Generate a human-readable diff summary of all operations for LLM consumption.

Returns:

Type	Description
str	String representation of all operations showing what will be added, removed, and modified.
str	Returns empty string if no operations to perform.

Source code in ontocast/onto/sparql_models.py

def generate_diff_summary(self) -> str:
    """Generate a human-readable diff summary of all operations for LLM consumption.

    Returns:
        String representation of all operations showing what will be added, removed, and modified.
        Returns empty string if no operations to perform.
    """
    if not self.triple_operations:
        return ""

    diff_parts = []
    operation_count = 0

    for i, op in enumerate(self.triple_operations, 1):
        if len(op.graph) > 0:
            op_type = op.type.upper()
            diff_parts.append(f"{i}. {op_type} {len(op.graph)} triple(s):")

            graph_prefixes = {
                prefix: str(uri) for prefix, uri in op.graph.namespaces() if prefix
            }
            all_prefixes = {**graph_prefixes, **op.prefixes}
            if all_prefixes:
                prefix_list = ", ".join(
                    [f"{k}: {v}" for k, v in all_prefixes.items()]
                )
                diff_parts.append(f"   Prefixes: {prefix_list}")

            # Mirror the prefix set generate_sparql_queries declares, so the
            # summary abbreviates exactly what the emitted query does.
            declared = frozenset(STANDARD_PREFIXES) | frozenset(all_prefixes)
            for subject, predicate, obj in self._serializable_triples(op.graph):
                symbol = "+" if op.type == "insert" else "-"
                diff_parts.append(
                    f"   {symbol} {self._serialize_rdf_term(subject, declared)} {self._serialize_rdf_term(predicate, declared)} {self._serialize_rdf_term(obj, declared)}"
                )
            operation_count += 1

    if operation_count == 0:
        return ""

    summary = f"Ontology Update Summary ({operation_count} operation(s)):\n\n"
    summary += "\n".join(diff_parts)

    return summary

generate_sparql_queries()

Compile triple_operations to SPARQL UPDATE strings for rdflib execution.

Returns:

Type	Description
list[str]	List of SPARQL query strings in operation order.

Source code in ontocast/onto/sparql_models.py

def generate_sparql_queries(self) -> list[str]:
    """Compile triple_operations to SPARQL UPDATE strings for rdflib execution.

    Returns:
        List of SPARQL query strings in operation order.
    """
    queries = []

    for op in self.triple_operations:
        if len(op.graph) > 0:
            prefixes = STANDARD_PREFIXES.copy()

            for prefix, uri in op.graph.namespaces():
                if prefix:
                    prefixes[prefix] = str(uri)

            prefixes.update(op.prefixes)

            if prefixes:
                prefix_declarations = []
                for prefix, uri in prefixes.items():
                    prefix_declarations.append(f"PREFIX {prefix}: <{uri}>")
                prefix_block = "\n".join(prefix_declarations)
            else:
                prefix_block = ""

            declared = frozenset(prefixes)
            if op.type == "insert":
                triple_query = self._generate_insert_query(
                    op.graph, prefix_block, declared
                )
            else:
                triple_query = self._generate_delete_query(
                    op.graph, prefix_block, declared
                )
            queries.append(triple_query)

    return queries

TripleOp

Bases: BaseModel

Operation to modify triples in the RDF graph.

This operation can insert or delete triples. Prefixes are automatically extracted from the RDFGraph's namespace bindings (from @prefix declarations in Turtle).

Source code in ontocast/onto/sparql_models.py

class TripleOp(BaseModel):
    """Operation to modify triples in the RDF graph.

    This operation can insert or delete triples. Prefixes are automatically extracted
    from the RDFGraph's namespace bindings (from @prefix declarations in Turtle).
    """

    type: TypingLiteral["insert", "delete"] = Field(
        description="Type of operation: 'insert' to add triples, 'delete' to remove triples"
    )

    @field_validator("type", mode="before")
    @classmethod
    def normalize_op_type(cls, v: object) -> str:
        if isinstance(v, str) and v.lower() == "update":
            return "insert"
        if isinstance(v, str):
            return v
        raise TypeError(f"TripleOp.type must be a string, got {type(v).__name__}")

    graph: LLMGraphWire = Field(
        default_factory=RDFGraph,
        description=(
            "RDF triples for this insert or delete operation. "
            "Encoding is defined by deployment llm_graph_format and OUTPUT INSTRUCTION."
        ),
    )
    prefixes: dict[str, str] = Field(
        default_factory=dict,
        description="Optional: Additional or override prefixes. "
        "Prefixes are automatically extracted from the RDFGraph's namespace bindings. "
        "Standard prefixes from COMMON_PREFIXES in constants.py (rdf, rdfs, owl, xsd, dc, dcterms, skos, foaf, schema, prov, ex) are automatically available. "
        "This field can be used to add or override prefixes if needed. "
        "Mapping format: {'prefix_name': 'namespace_uri'}. Example: {'fca': 'http://example.org/ontologies/fca#'}",
    )

Attributes

graph = Field(default_factory=RDFGraph, description='RDF triples for this insert or delete operation. Encoding is defined by deployment llm_graph_format and OUTPUT INSTRUCTION.') class-attribute instance-attribute

prefixes = Field(default_factory=dict, description="Optional: Additional or override prefixes. Prefixes are automatically extracted from the RDFGraph's namespace bindings. Standard prefixes from COMMON_PREFIXES in constants.py (rdf, rdfs, owl, xsd, dc, dcterms, skos, foaf, schema, prov, ex) are automatically available. This field can be used to add or override prefixes if needed. Mapping format: {'prefix_name': 'namespace_uri'}. Example: {'fca': 'http://example.org/ontologies/fca#'}") class-attribute instance-attribute

type = Field(description="Type of operation: 'insert' to add triples, 'delete' to remove triples") class-attribute instance-attribute

Methods:

normalize_op_type(v) classmethod

Source code in ontocast/onto/sparql_models.py

@field_validator("type", mode="before")
@classmethod
def normalize_op_type(cls, v: object) -> str:
    if isinstance(v, str) and v.lower() == "update":
        return "insert"
    if isinstance(v, str):
        return v
    raise TypeError(f"TripleOp.type must be a string, got {type(v).__name__}")

Functions: