Session

ORM unit of work over a graph store (SPARQLSession).

class sparqlmodel.session.SPARQLSession(store=None, *, prefixes=None, autoflush=True, close_on_exit=True, rollback_on_error=True)

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)

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

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

flush()

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

rollback_pending()

Discard pending models without writing to the store.

close()

Close the backing store when it implements close().

expire(model_cls, iri)

Remove a resource from the identity map and hydration cache.

expunge(model)

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

expunge_all()

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

refresh(model, *, depth=0)

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

Return type:

SPARQLModel

merge(model)

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

Return type:

SPARQLModel

add(model)

Insert model triples into the store (no delete).

Return type:

SPARQLModel

put(model, *, flush=True)

Upsert model and cascaded embedded resources.

Return type:

SPARQLModel

delete(model)

Remove owned triples for the model and cascaded embedded resources.

get(model_cls, iri, *, depth=0)

Load a model by IRI with optional relationship depth.

Return type:

SPARQLModel | None

hydrate_bindings(model_cls, bindings, *, depth=0, polymorphic=False)

Hydrate query results with identity map and session cache.

Return type:

list[SPARQLModel]

query(model_cls)

Start a fluent query for the given model class.

Return type:

Query

execute(sparql)

Execute raw SPARQL SELECT.

Return type:

list[dict[str, Any]]

Session-level identity map, hydration cache, and pending writes.

class sparqlmodel.session_state.SessionState

Bases: object

Identity map, hydration cache, and pending put queue for a session.

__init__()

invalidate_hydration_for_iri(iri_key)

Drop hydration cache entries for all model classes at iri_key.

remove_pending_for(model_cls, iri_key)

Drop queued put(..., flush=False) writes for (model_cls, iri_key).

expunge_all()

Clear identity map and hydration cache (pending queue unchanged).

ORM eager-load: hydrate query results and get by relationship depth.

sparqlmodel.hydration.validate_depth(depth)

Raise if hydration depth is outside the supported range (0–2).

sparqlmodel.hydration.hydrate_from_bindings(model_cls, bindings, store, *, depth=0)

Hydrate models from SPARQL SELECT bindings.

Return type:

list[SPARQLModel]

sparqlmodel.hydration.hydrate_one(model_cls, iri, store, *, depth=0, polymorphic=False)

Load a single model by IRI from the store.

Return type:

SPARQLModel | None

Async ORM unit of work over a graph store (AsyncSPARQLSession).

class sparqlmodel.async_session.AsyncSPARQLSession(store=None, *, prefixes=None, autoflush=True, close_on_exit=True, rollback_on_error=True)

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)

async flush()

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

async rollback_pending()

Discard pending models without writing to the store.

async close()

Close the backing store when it implements aclose().

async expire(model_cls, iri)

Remove a resource from the identity map and hydration cache.

async expunge(model)

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

async expunge_all()

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

async refresh(model, *, depth=0)

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

Return type:

SPARQLModel

async merge(model)

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

Return type:

SPARQLModel

async add(model)

Insert model triples into the store (no delete).

Return type:

SPARQLModel

async put(model, *, flush=True)

Upsert model and cascaded embedded resources.

Return type:

SPARQLModel

async delete(model)

Remove owned triples for the model and cascaded embedded resources.

async get(model_cls, iri, *, depth=0)

Load a model by IRI with optional relationship depth.

Return type:

SPARQLModel | None

async hydrate_bindings(model_cls, bindings, *, depth=0, polymorphic=False)

Hydrate query results with identity map and session cache.

Return type:

list[SPARQLModel]

query(model_cls)

Start a fluent async query for the given model class.

Return type:

AsyncQuery

async execute(sparql)

Execute raw SPARQL SELECT.

Return type:

list[dict[str, Any]]

Async ORM query builder; compiles Python filters to SPARQL.

class sparqlmodel.async_query.AsyncQuery(session, model_cls)

Bases: object

Async ORM query builder for a SPARQLModel class.

__init__(session, model_cls)

Shared session CRUD, hydration, and graph-sync logic (sync and async).

sparqlmodel.session_core.sparql_has_prefix_declarations(sparql)

Return True if sparql already declares at least one PREFIX.

Return type:

bool

sparqlmodel.session_core.depth_satisfied(model, depth)

Return whether model has relationships loaded through depth.

Return type:

bool

sparqlmodel.session_core.invalidate_subjects(state, subjects)

Expire identity and hydration cache entries for cascade subjects.

sparqlmodel.session_core.check_stale_add(store_graph, model)

Warn when add may duplicate triples for an existing subject.

sparqlmodel.session_core.expunge_impl(state, model)

Remove model from the identity map and hydration cache.

sparqlmodel.session_core.expunge_all_impl(state)

Clear identity map and hydration cache (pending queue unchanged).

sparqlmodel.session_core.refresh_impl(state, store, model, *, depth)

Reload model from the store graph at depth.

Return type:

SPARQLModel

async sparqlmodel.session_core.refresh_impl_async(state, store, model, *, depth)

Reload model from the async store mirror at depth.

Return type:

SPARQLModel

sparqlmodel.session_core.merge_impl(state, model)

Attach or reconcile model with the session identity map (no store write).

Return type:

SPARQLModel