triplemodel.io.sparql

SPARQL query passthrough and remote endpoint helpers.

class triplemodel.io.sparql.PreparedModelQuery(model_cls, query_str)[source]

Bases: object

Prepared SPARQL query with namespaces from a model’s Rdf.prefixes.

Parameters:
model_cls: type[BaseModel]
query_str: str
property prepared: str
execute(graph, *, initBindings=None, use_store_provided=True, **kwargs)[source]

Run the prepared query on graph.

Return type:

SparqlResult

Parameters:

  • graph (RdfGraph)

  • initBindings (Mapping[Variable, NamedNode | BlankNode | Literal] | None)

  • use_store_provided (bool)

  • kwargs (Any)

as_result(graph, **kwargs)[source]

Alias for execute().

Return type:

SparqlResult

Parameters:

  • graph (RdfGraph)

  • kwargs (Any)

__init__(model_cls, query_str)
Parameters:
Return type:

None

triplemodel.io.sparql.apply_update(graph, update, *, model_cls=None, initNs=None, initBindings=None, use_store_provided=True, **kwargs)[source]

Apply a SPARQL UPDATE to graph (in-memory models may be stale afterward).

Return type:

None

Parameters:
triplemodel.io.sparql.ask(graph, query, *, model_cls=None, initNs=None, initBindings=None, use_store_provided=True, **kwargs)[source]

Execute an ASK query and return the boolean result.

Return type:

bool

Parameters:
triplemodel.io.sparql.construct_models(model_cls, graph, query, *, dispatch=False, graph_out=None, type_uri=None, validate_type=True, on_duplicate='warn', resolver=None, registry=<triplemodel.terms.registry.LiteralRegistry object>, de_skolemize=None, initNs=None, initBindings=None, use_store_provided=True, **kwargs)[source]

Run CONSTRUCT/DESCRIBE and load models from the result graph.

Return type:

list[TypeVar(T, bound= BaseModel)]

Parameters:
triplemodel.io.sparql.detect_query_form(query)[source]

Return the first SPARQL query form keyword in query.

Return type:

Literal['select', 'construct', 'describe', 'ask', 'unknown']

Parameters:

query (str)

triplemodel.io.sparql.graph_from_construct_result(result, graph_out=None)[source]

Merge a CONSTRUCT/DESCRIBE Result graph into graph_out or a new graph.

Return type:

RdfGraph

Parameters:

  • result (SparqlResult)

  • graph_out (RdfGraph | None)

triplemodel.io.sparql.init_bindings_from_model(instance, mapping)[source]

Map SPARQL variable names to RDF terms from model field values.

Return type:

dict[Variable, NamedNode | BlankNode | Literal]

Parameters:
triplemodel.io.sparql.init_ns_from_model(model_cls)[source]

Build prefix map for SPARQL from model_cls Rdf.prefixes.

Return type:

dict[str, str]

Parameters:

model_cls (type[BaseModel])

triplemodel.io.sparql.load_sparql(model_cls, endpoint, query, *, query_form=None, read_only=True, dispatch=False, type_uri=None, validate_type=True, on_duplicate='warn', resolver=None, registry=<triplemodel.terms.registry.LiteralRegistry object>, de_skolemize=None, initNs=None, initBindings=None, use_store_provided=True, **kwargs)[source]

Query a remote SPARQL endpoint and return model instances.

Uses open_sparql_graph(), which raises NotImplementedError in 0.10.0. Query a local Store with construct_models() / select_models() after fetching data, or use SparqlModel for remote endpoints.

Return type:

list[TypeVar(T, bound= BaseModel)]

Parameters:
triplemodel.io.sparql.open_sparql_graph(endpoint, *, read_only=True)[source]

Open a remote SPARQL graph (removed in 0.10.0).

Raises NotImplementedError. Load remote data into a local Store or use SparqlModel for session-level remote stores.

Return type:

RdfGraph

Parameters:
triplemodel.io.sparql.prepare_model_query(model_cls, query)[source]

Prepare query (prefixes applied at execution from model_cls).

Return type:

PreparedModelQuery

Parameters:
triplemodel.io.sparql.run_sparql(graph, query, *, model_cls=None, initNs=None, initBindings=None, use_store_provided=True, use_default_graph_as_union=False, default_graph=None, named_graphs=None, base_iri=None, **kwargs)[source]

Run SPARQL on graph.store with optional prefixes and bindings.

Return type:

SparqlResult

Parameters:

  • graph (RdfGraph)

  • query (str)

  • model_cls (type[BaseModel] | None)

  • initNs (Mapping[str, Any] | None)

  • initBindings (Mapping[Variable, NamedNode | BlankNode | Literal] | Mapping[str, NamedNode | BlankNode | Literal] | None)

  • use_store_provided (bool)

  • use_default_graph_as_union (bool)

  • default_graph (str | NamedNode | DefaultGraph | None)

  • named_graphs (list[str | NamedNode] | None)

  • base_iri (str | None)

  • kwargs (Any)

triplemodel.io.sparql.select_models(model_cls, graph, query, *, field_map=None, subject_var=None, hydrate=False, type_uri=None, validate_type=True, on_duplicate='warn', resolver=None, registry=<triplemodel.terms.registry.LiteralRegistry object>, de_skolemize=None, initNs=None, initBindings=None, use_store_provided=True, **kwargs)[source]

Run SELECT and return model instances from result bindings or hydration.

Return type:

list[TypeVar(T, bound= BaseModel)]

Parameters: