Skip to content

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, graph_version_manager.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 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
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 = ""

                if op.type == "insert":
                    triple_query = self._generate_insert_query(op.graph, prefix_block)
                else:
                    triple_query = self._generate_delete_query(op.graph, prefix_block)
                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:
                for triple in op.graph:
                    result.add(triple)
                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}")

                for subject, predicate, obj in op.graph:
                    symbol = "+" if op.type == "insert" else "-"
                    diff_parts.append(
                        f"   {symbol} {self._serialize_rdf_term(subject)} {self._serialize_rdf_term(predicate)} {self._serialize_rdf_term(obj)}"
                    )
                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

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

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

        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) -> str:
        """Generate a SPARQL DELETE query for the given RDFGraph."""
        if len(graph) == 0:
            return ""

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

        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)

    def _serialize_rdf_term(self, term: Node) -> str:
        """Serialize an RDF term to its SPARQL string representation."""
        if isinstance(term, URIRef):
            if ":" in str(term) and not str(term).startswith("http"):
                return str(term)
            else:
                return f"<{term}>"
        elif isinstance(term, BNode):
            return f"_:{term}"
        elif isinstance(term, Literal):
            if term.language:
                return f'"{term}"@{term.language}'
            elif term.datatype:
                return f'"{term}"^^<{term.datatype}>'
            else:
                return f'"{term}"'
        else:
            return str(term)

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 
137
138
139
140
141
142
143
144
145
146
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 
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
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:
            for triple in op.graph:
                result.add(triple)
            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 
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
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}")

            for subject, predicate, obj in op.graph:
                symbol = "+" if op.type == "insert" else "-"
                diff_parts.append(
                    f"   {symbol} {self._serialize_rdf_term(subject)} {self._serialize_rdf_term(predicate)} {self._serialize_rdf_term(obj)}"
                )
            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 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
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 = ""

            if op.type == "insert":
                triple_query = self._generate_insert_query(op.graph, prefix_block)
            else:
                triple_query = self._generate_delete_query(op.graph, prefix_block)
            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 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
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 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
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#'}",
    )