ontocast.onto.sparql_models

Pydantic models for graph mutations and tool SPARQL operations.

GraphUpdate / TripleOp are the canonical LLM pipeline mutation abstraction (ordered insert/delete triple patches). SPARQLOperationModel is used by tooling (tool/sparql.py) — a separate path.

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}"
            )

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

SPARQLOperationModel

Bases: BaseModel

Pydantic model for a single SPARQL operation.

Attributes:

Name	Type	Description
operation_type	SPARQLOperationType	Type of SPARQL operation (INSERT, UPDATE, DELETE)
query	str	The SPARQL query string
description	str	Optional description of the operation
metadata	dict[str, Any]	Optional metadata dictionary

Source code in ontocast/onto/sparql_models.py

class SPARQLOperationModel(BaseModel):
    """Pydantic model for a single SPARQL operation.

    Attributes:
        operation_type: Type of SPARQL operation (INSERT, UPDATE, DELETE)
        query: The SPARQL query string
        description: Optional description of the operation
        metadata: Optional metadata dictionary
    """

    operation_type: SPARQLOperationType = Field(
        description="Type of SPARQL operation: INSERT, UPDATE, or DELETE"
    )
    query: str = Field(
        description="The complete SPARQL query string with proper syntax"
    )
    description: str = Field(
        default="", description="Optional description of the operation"
    )
    metadata: dict[str, Any] = Field(
        default_factory=dict,
        description="Optional metadata dictionary for the operation",
    )

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#'}",
    )