Package

SparqlModel — SPARQL ORM for RDF triple stores.

The SQLModel of SPARQL: typed models, SPARQLSession as the main entry point, and Python queries that compile to SPARQL.

Requires triplemodel>=0.12 and pyoxigraph for in-process graphs; SparqlModel owns the session, query compiler, stores, and cascade policy.

class sparqlmodel.AsyncHttpStore(endpoint, *, read_endpoint=None, write_endpoint=None, graph=None, prefixes=None, auth=None, bearer_token=None, headers=None, timeout=30.0, client=None, mirror_mode='writer', max_retries=2, retry_backoff=0.5, max_triples_per_update=500, query_method='post', graph_store_url=None)[source]

Bases: object

Async SPARQL 1.1 endpoint store with a local triplemodel.Store mirror.

Same semantics as HttpStore, using httpx.AsyncClient for non-blocking remote I/O. Supports optional read_endpoint / write_endpoint, mirror_mode, chunked UPDATE (max_triples_per_update), retries (max_retries / retry_backoff), and query_method post vs get for SELECT.

Mirror limitations: Data written outside this store instance is visible to query / execute but not to graph, get, or cascade until the mirror is updated. Use pull_subjects_into_mirror(), sync_mirror() (when graph_store_url is set), or get (auto-pull) to hydrate from remote. Prefer AsyncMemoryStore for asyncio tests. Assume a single writer per endpoint.

If both auth and bearer_token are set, Basic auth wins for Authorization.

__init__(endpoint, *, read_endpoint=None, write_endpoint=None, graph=None, prefixes=None, auth=None, bearer_token=None, headers=None, timeout=30.0, client=None, mirror_mode='writer', max_retries=2, retry_backoff=0.5, max_triples_per_update=500, query_method='post', graph_store_url=None)[source]

async aclose()[source]

property endpoint: str

property graph: RdfGraph

property graph_store_url: str | None: Graph Store HTTP URL for sync_mirror() (optional).

property mirror_generation: int: Monotonic counter bumped when the local mirror is wholesale or partially replaced.

property mirror_mode: Literal['writer', 'remote_authoritative']: Mirror mode — writer (default) or remote_authoritative.

property namespaces: NamespaceRegistry

async pull_subjects_into_mirror(iris)[source]

Fetch triples for iris from the remote endpoint into the local mirror.

async query(sparql)[source]

Execute SPARQL SELECT against the remote read endpoint.

Return type:: list[dict[str, Any]]

property query_method: Literal['post', 'get']: How remote SELECT queries are sent — post (default) or get.

property read_endpoint: str

async sync_mirror()[source]

Replace the local mirror with the remote default graph (Graph Store HTTP GET).

async update_graph(add=None, remove=None)[source]

Apply graph delta to remote endpoint and local mirror.

property write_endpoint: str

class sparqlmodel.AsyncMemoryStore(graph=None, *, prefixes=None)[source]

Bases: object

In-memory RDF store with async methods delegating to sync MemoryStore.

__init__(graph=None, *, prefixes=None)[source]

async aclose()[source]

property graph: RdfGraph

property namespaces

async query(sparql)[source]

Return type:: list[dict[str, Any]]

async update_graph(add=None, remove=None)[source]

class sparqlmodel.AsyncSPARQLSession(store=None, *, prefixes=None, autoflush=True, close_on_exit=True, rollback_on_error=True)[source]

Bases: object

Async ORM session: CRUD, queries, and graph sync with the backing store.

Use as an async context manager:

async with AsyncSPARQLSession(store=AsyncHttpStore(endpoint)) as session:
    await session.put(model)

Same identity map, hydration, and cascade semantics as SPARQLSession. Not safe to share across concurrent asyncio tasks; use one session per task.

__init__(store=None, *, prefixes=None, autoflush=True, close_on_exit=True, rollback_on_error=True)[source]

async add(model)[source]

Insert model triples into the store (no delete).

Return type:: SPARQLModel

async close()[source]

Close the backing store when it implements aclose().

async delete(model)[source]

Remove owned triples for the model and cascaded embedded resources.

async execute(sparql)[source]

Execute raw SPARQL SELECT.

Return type:: list[dict[str, Any]]

async expire(model_cls, iri)[source]

Remove a resource from the identity map and hydration cache.

async expunge(model)[source]

Detach model from the identity map and hydration cache (store unchanged).

async expunge_all()[source]

Clear the identity map and hydration cache; pending put queue is kept.

async flush()[source]

Write all pending models queued with put(..., flush=False).

async get(model_cls, iri, *, depth=0)[source]

Load a model by IRI with optional relationship depth.

Return type:: SPARQLModel | None

property graph: RdfGraph

async hydrate_bindings(model_cls, bindings, *, depth=0, polymorphic=False)[source]

Hydrate query results with identity map and session cache.

Return type:: list[SPARQLModel]

async merge(model)[source]

Return the session instance for model’s identity key (no store write).

Return type:: SPARQLModel

property namespaces: NamespaceRegistry

async put(model, *, flush=True)[source]

Upsert model and cascaded embedded resources.

Return type:: SPARQLModel

query(model_cls)[source]

Start a fluent async query for the given model class.

Return type:: AsyncQuery

async refresh(model, *, depth=0)[source]

Reload model from the store at depth (updates cached instance when present).

Return type:: SPARQLModel

async rollback_pending()[source]

Discard pending models without writing to the store.

property store: AsyncStoreProtocol

class sparqlmodel.BackPopulates(field, model)[source]

Bases: object

Link a field to the inverse-side field on another model (SparqlModel back_populates).

__init__(field, model)

field: str

model: type[BaseModel] | str

exception sparqlmodel.ConfigurationError[source]

Bases: SparqlModelError

Raised when a model or session is misconfigured.

sparqlmodel.Field(predicate, *, lang=None, inverse=None, literal_datatype=None, transitive=False, back_populates=None, **kwargs)[source]

Map a model attribute to an RDF predicate.

Parameters:

predicate (str) – Compact or absolute IRI (e.g. schema:name).
lang (str | None) – Default language tag for string literals (Lang metadata).
inverse (str | None) – Inverse predicate IRI for import-only triples.
literal_datatype (str | None) – XSD datatype CURIE for typed literals.
transitive (bool) – Expand multi-valued object URIs transitively on import.
back_populates (Any) – BackPopulates, inverse_pair, or (Model, field) tuple.
**kwargs (Any) – Additional arguments passed to pydantic.Field.

Return type:

Any

exception sparqlmodel.HydrationError[source]

Bases: SparqlModelError

Raised when query results cannot be hydrated into models.

class sparqlmodel.HttpStore(endpoint, *, read_endpoint=None, write_endpoint=None, graph=None, prefixes=None, auth=None, bearer_token=None, headers=None, timeout=30.0, client=None, mirror_mode='writer', max_retries=2, retry_backoff=0.5, max_triples_per_update=500, query_method='post', graph_store_url=None)[source]

Bases: object

SPARQL 1.1 endpoint store with a local triplemodel.Store mirror.

update_graph pushes INSERT DATA / DELETE DATA to the remote endpoint (chunked when max_triples_per_update is exceeded) and applies the mirror delta only after all remote chunks succeed. query executes SELECT against the remote read endpoint (POST by default, optional GET via query_method).

Optional read_endpoint / write_endpoint support Fuseki-style split URLs (defaults to endpoint for both). Transient HTTP failures (502/503/504, connection errors) are retried per max_retries / retry_backoff.

Mirror limitations: Data written outside this store instance (another app, admin UI, or raw SPARQL UPDATE) is visible to query / execute but not to graph, get, or cascade/orphan logic until the mirror is updated. Use pull_subjects_into_mirror(), sync_mirror() (Graph Store HTTP GET when graph_store_url is set), or get (which pulls automatically) to hydrate the mirror from the remote dataset. Prefer MemoryStore for single-process apps and tests. Assume a single writer per endpoint when using HttpStore.

If both auth and bearer_token are set, Basic auth wins for Authorization.

__init__(endpoint, *, read_endpoint=None, write_endpoint=None, graph=None, prefixes=None, auth=None, bearer_token=None, headers=None, timeout=30.0, client=None, mirror_mode='writer', max_retries=2, retry_backoff=0.5, max_triples_per_update=500, query_method='post', graph_store_url=None)[source]

close()[source]

property endpoint: str

property graph: RdfGraph

property graph_store_url: str | None: Graph Store HTTP URL for sync_mirror() (optional).

property mirror_generation: int: Monotonic counter bumped when the local mirror is wholesale or partially replaced.

property mirror_mode: Literal['writer', 'remote_authoritative']: Mirror mode — writer (default) or remote_authoritative.

property namespaces: NamespaceRegistry

pull_subjects_into_mirror(iris)[source]

Fetch triples for iris from the remote endpoint into the local mirror.

query(sparql)[source]

Execute SPARQL SELECT against the remote read endpoint.

Return type:: list[dict[str, Any]]

property query_method: Literal['post', 'get']: How remote SELECT queries are sent — post (default) or get.

property read_endpoint: str

sync_mirror()[source]

Replace the local mirror with the remote default graph (Graph Store HTTP GET).

Requires graph_store_url on construction. Does not modify the remote dataset. For per-subject hydration use pull_subjects_into_mirror() instead.

update_graph(add=None, remove=None)[source]

Apply graph delta to remote endpoint and local mirror.

property write_endpoint: str

class sparqlmodel.IRI[source]

Bases: str

RDF IRI identifier (compact or absolute).

compact(prefixes=None)[source]

Return the most compact form using known prefixes.

Return type:: str

expand(prefixes=None)[source]

Expand a compact IRI (e.g. schema:Person) to an absolute IRI.

Return type:: str

class sparqlmodel.Lang(code, direction=None)[source]

Bases: object

Field metadata: serialize str values with a fixed language tag.

__init__(code, direction=None)

direction: str | None = None

code: str

class sparqlmodel.LangString(value, lang=None, direction=None)[source]

Bases: object

A literal string value with an optional language tag and text direction.

__init__(value, lang=None, direction=None)

direction: str | None = None

lang: str | None = None

value: str

class sparqlmodel.MemoryStore(graph=None, *, prefixes=None)[source]

Bases: object

In-memory RDF store using Store.

__init__(graph=None, *, prefixes=None)[source]

property graph: RdfGraph

property mirror_generation: int: Monotonic counter; 0 for in-memory stores (no wholesale mirror sync).

property namespaces: NamespaceRegistry

query(sparql)[source]

Execute SPARQL SELECT and return variable bindings.

Return type:: list[dict[str, Any]]

update_graph(add=None, remove=None)[source]

Add or remove triples.

class sparqlmodel.MultiLangString(mapping=None, /, **langs)[source]

Bases: object

Multiple language-tagged literals for one predicate, keyed by language code.

Use on a single field when the graph has several @lang objects on the same predicate (for example rdfs:label@en and rdfs:label@fr). Export emits one triple per entry; import collects all language-tagged literals on the field.

__init__(mapping=None, /, **langs)[source]

classmethod from_mapping(mapping)[source]

Build from a language code → value map (None entries are omitted).

Return type:: MultiLangString

get(lang, default=None)[source]

Return type:: LangString | None

values()[source]

Language-tagged literals in arbitrary order (for export).

Return type:: list[LangString]

by_lang: Mapping[str, LangString]

class sparqlmodel.OntologyRegistry(_graph=None, _parent_to_children=<factory>, _registered_inverse_forward=<factory>, _registered_inverse_reverse=<factory>, _graph_inverse_forward=<factory>, _graph_inverse_reverse=<factory>)[source]

Bases: object

Subclass and owl:inverseOf hints from an ontology graph and/or static registration.

Use subtypes_of() for descendant type IRIs (subclasses). For ancestor types (superclasses of a given class), use subclass_uris() on a graph that contains rdfs:subClassOf axioms.

__init__(_graph=None, _parent_to_children=<factory>, _registered_inverse_forward=<factory>, _registered_inverse_reverse=<factory>, _graph_inverse_forward=<factory>, _graph_inverse_reverse=<factory>)

classmethod from_graph(graph)[source]

Build a registry backed by graph (typically parsed OWL/RDFS Turtle).

Return type:: OntologyRegistry

classmethod from_ttl(path, *, format=None, base_iri=None)[source]

Parse a Turtle (or other) ontology file into a new registry.

Return type:: OntologyRegistry

inverse_of(predicate_uri)[source]

Return the inverse property IRI for predicate_uri, if known.

Return type:: str | None

load_graph(graph)[source]

Use graph for subtypes_of() and inverse_of() queries.

register_inverse(forward_predicate, inverse_predicate)[source]

Register an owl:inverseOf pair (both directions are queryable).

register_subclasses(base_type_uri, subtype_uris)[source]

Register direct rdfs:subClassOf links without an ontology file.

subtypes_of(type_uri)[source]

Return type_uri and all registered or inferred subclass IRIs.

Return type:: frozenset[str]

exception sparqlmodel.QueryError[source]

Bases: SparqlModelError

Raised when a query cannot be compiled or executed.

sparqlmodel.Relationship(predicate, *, model=None, cascade=True, inverse=None, back_populates=None, **kwargs)[source]

Map a model attribute to an RDF object relationship.

Parameters:

predicate (str) – Compact or absolute IRI (e.g. schema:worksFor).
model (type[Any] | None) – Related SPARQLModel class (inferred from annotation when omitted).
cascade (bool) – When False, nested resources are not included in put/delete cascade.
inverse (str | None) – Inverse predicate for import (not with ref_field).
back_populates (Any) – Paired inverse field on another model (TripleModel 0.12+).
**kwargs (Any) – Additional arguments passed to pydantic.Field.

Return type:

Any

class sparqlmodel.ResourceRef(iri)[source]

Bases: object

Reference to an RDF resource identified by IRI.

__init__(iri)

iri: str

class sparqlmodel.SPARQLModel(*, id=None)[source]

Bases: TripleModel

ORM entity mapped to RDF; persist and query via SPARQLSession.

ensure_id()[source]

Ensure the instance has an IRI id.

Return type:: IRI

classmethod get_prefixes()[source]

Return namespace prefixes for this model (includes built-in RDF prefixes).

Return type:: dict[str, str]

classmethod get_relationship_fields()[source]

Return relationship field definitions.

Return type:: list[tuple[str, FieldInfo, type[SPARQLModel]]]

classmethod get_scalar_fields()[source]

Return scalar (non-relationship) field definitions.

Return type:: list[tuple[str, FieldInfo]]

classmethod iter_sparql_fields()[source]

Yield (name, field_info, annotation) for mapped fields excluding id.

Return type:: list[tuple[str, FieldInfo, Any]]

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'extra': 'forbid', 'str_strip_whitespace': False, 'validate_assignment': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_dump_jsonld()[source]

Serialize to a JSON-LD dict for APIs (cascade-aware ORM view).

For RDF files or HTTP bodies with all triples, use serialize(format="json-ld") (inherited from TripleModel) instead.

Return type:: dict[str, Any]

classmethod model_validate_jsonld(data)[source]

Deserialize from a JSON-LD dict (ORM presentation layer).

For RDF files, use parse() (inherited from TripleModel) or import_graph().

Return type:: Self

classmethod namespace_registry()[source]

Return type:: NamespaceRegistry

subject_uri(*, uri=None)[source]

Return the RDF subject IRI for this instance (expanded when compact).

Return type:: str

rdf_type: ClassVar[str]

Rdf: ClassVar[type]

id: Annotated[IRI | None, IriId()]

class sparqlmodel.SPARQLSession(store=None, *, prefixes=None, autoflush=True, close_on_exit=True, rollback_on_error=True)[source]

Bases: object

ORM session: CRUD, queries, and graph sync with the backing store.

Use as a context manager to flush pending writes on success, discard the pending queue on error, and close the backing store when it supports HttpStore.close() when using HttpStore:

with SPARQLSession(store=HttpStore(endpoint)) as session:
    session.put(model)

Already-flushed writes are not rolled back on error; only the pending queue from put(..., flush=False) is affected. Full transactional rollback may be added in a future release.

Not thread-safe; use one session per thread or asyncio task.

__init__(store=None, *, prefixes=None, autoflush=True, close_on_exit=True, rollback_on_error=True)[source]

add(model)[source]

Insert model triples into the store (no delete).

Return type:: SPARQLModel

close()[source]

Close the backing store when it implements close().

delete(model)[source]

Remove owned triples for the model and cascaded embedded resources.

execute(sparql)[source]

Execute raw SPARQL SELECT.

Return type:: list[dict[str, Any]]

expire(model_cls, iri)[source]

Remove a resource from the identity map and hydration cache.

expunge(model)[source]

Detach model from the identity map and hydration cache (store unchanged).

expunge_all()[source]

Clear the identity map and hydration cache; pending put queue is kept.

flush()[source]

Write all pending models queued with put(..., flush=False).

classmethod from_rdf_file(path, *, format='turtle', prefixes=None, autoflush=True, close_on_exit=True, rollback_on_error=True)[source]

Open a session over an on-disk RDF file.

Uses an in-memory MemoryStore. Parses path with TripleModel load_graph (pass a Path, not file contents as a string). Use for tutorials, fixtures, and ETL that starts from Turtle, TriG, or other supported formats before put / query.

Parameters:

path (str | Path) – File path to parse.
format (str) – RDF format hint (e.g. "turtle", "trig"); inferred from extension when omitted on load_graph.
prefixes (Mapping[str, str] | None) – Namespace prefixes merged into the session registry and graph.
autoflush (bool) – Passed to __init__().
close_on_exit (bool) – Passed to __init__().
rollback_on_error (bool) – Passed to __init__().

Return type:

Self

get(model_cls, iri, *, depth=0)[source]

Load a model by IRI with optional relationship depth.

Return type:: SPARQLModel | None

property graph: RdfGraph

hydrate_bindings(model_cls, bindings, *, depth=0, polymorphic=False)[source]

Hydrate query results with identity map and session cache.

Return type:: list[SPARQLModel]

merge(model)[source]

Return the session instance for model’s identity key (no store write).

Return type:: SPARQLModel

property namespaces: NamespaceRegistry

put(model, *, flush=True)[source]

Upsert model and cascaded embedded resources.

Return type:: SPARQLModel

query(model_cls)[source]

Start a fluent query for the given model class.

Return type:: Query

refresh(model, *, depth=0)[source]

Reload model from the store at depth (updates cached instance when present).

Return type:: SPARQLModel

rollback_pending()[source]

Discard pending models without writing to the store.

property store: StoreProtocol

sparqlmodel.SchemaRegistry: alias of OntologyRegistry

exception sparqlmodel.SparqlModelError[source]

Bases: Exception

Base exception for SparqlModel.

exception sparqlmodel.StaleTripleWarning[source]

Bases: UserWarning

Warn when add() may leave stale triples on an existing subject.

class sparqlmodel.TypedLiteral(value, datatype=None)[source]

Bases: object

One RDF literal value with its own XSD (or custom) datatype IRI.

Use as the element type of set[TypedLiteral] or list[TypedLiteral] when several objects on one predicate may carry different ^^datatype IRIs. Field-level literal_datatype= applies one datatype to every object; this type preserves each graph literal’s datatype independently.

__init__(value, datatype=None)[source]