SPARQL and remote endpoints
TripleModel adds thin helpers around a local Store.query / Store.update. There is no Python query DSL — write SPARQL (or use SparqlModel for ORM-style queries and remote endpoints).
When to use which helper
Goal |
Helper |
|---|---|
Full resources from a local graph |
|
Tabular SELECT → flat model fields |
|
Filter subjects already in a local graph |
|
Boolean check |
|
Mutate a graph in place |
|
Remote SPARQL endpoint |
Not in 0.10 — fetch data into a local |
SELECT returns bindings only — not full RDF graphs. Prefer CONSTRUCT for graph → TripleModel round-trip.
open_sparql_graph and load_sparql raise NotImplementedError in 0.10.0 (pyoxigraph has no built-in remote SPARQL graph). See Migrating to TripleModel 0.10.0 (pyoxigraph).
CONSTRUCT on an in-memory graph
from triplemodel import TripleModel, rdf_field
from triplemodel.vocab import FOAF
class Person(TripleModel):
class Rdf:
namespace = "http://example.org/people/"
type_uri = f"{FOAF}Person"
id_field = "slug"
prefixes = {"foaf": str(FOAF)}
slug: str
name: str = rdf_field("foaf:name")
people = Person.construct_from_sparql(
graph,
"""
CONSTRUCT { ?s ?p ?o }
WHERE { ?s a foaf:Person . ?s ?p ?o . }
""",
)
Remote data (pattern)
Load RDF from an endpoint or file into a local Store, then use the helpers above:
from triplemodel import Store, load_graph
graph = load_graph(source="endpoint-export.nt", format="nt")
people = Person.construct_from_sparql(
graph,
"CONSTRUCT { ?s ?p ?o } WHERE { ?s a foaf:Person . ?s ?p ?o }",
)
For live SPARQL sessions against remote endpoints, use SparqlModel.
SELECT projection
Map SPARQL variables to model fields (default: variable name = field name):
rows = Person.select_from_sparql(
graph,
"""
SELECT ?slug ?name WHERE {
?s a foaf:Person .
?s foaf:name ?name .
BIND(REPLACE(STR(?s), "http://example.org/people/", "") AS ?slug)
}
""",
)
With a subject URI variable, set subject_var="s" so Rdf.id_field is filled from the IRI (or the full IRI when IriId is used).
Dataset union and named graphs
run_sparql and PreparedModelQuery.execute forward pyoxigraph dataset options:
from triplemodel import run_sparql
result = run_sparql(
graph,
"SELECT ?s WHERE { ?s ?p ?o }",
use_default_graph_as_union=True,
named_graphs=["http://example.org/graph/g1"],
default_graph="http://example.org/graph/default",
)
Save or reload result documents with parse_query_results and SparqlResult.serialize(format="sparql-results+json").
ASK
from triplemodel import ask
if Person.ask_sparql(graph, "ASK { ?s a foaf:Person }"):
...
SPARQL UPDATE
apply_update wraps graph.update. In-memory TripleModel instances are not updated automatically.
from triplemodel import apply_update
apply_update(
graph,
"""
PREFIX foaf: <http://xmlns.com/foaf/0.1/>
DELETE { ?s foaf:age ?age } WHERE { ?s foaf:age ?age }
""",
model_cls=Person,
)
person = Person.from_graph(graph, person.subject_uri())
For mapped predicates on a known subject, sync_to_graph remains the typed alternative; SPARQL UPDATE is for arbitrary graph edits.
Prepared queries and bindings
from triplemodel import prepare_model_query, init_bindings_from_model
pq = prepare_model_query(Person, "SELECT ?name WHERE { ?s foaf:name ?name . FILTER(?s = ?subj) }")
bindings = init_bindings_from_model(alice, {"subj": "slug"})
result = pq.execute(graph, initBindings=bindings)
run_sparql (and helpers that call it) bind Rdf.prefixes on the same Store you pass in (bind_namespaces with override=True). That mutates the graph for serialization and SPARQL prefix resolution.
Security
graph.query may follow SERVICE clauses when the underlying engine supports them. Only run trusted queries or restrict network access when loading untrusted RDF.