Sessions and stores

SPARQLSession is the unit of work: all reads and writes go through a session bound to a store (MemoryStore or HttpStore).

Context manager (recommended)

from sparqlmodel import SPARQLSession

with SPARQLSession() as session:
    session.put(model)
    found = session.get(Person, model.id)
# flush pending writes on success; rollback pending queue on error; close store

Exit path	Behavior
Normal	`flush()` pending `put(..., flush=False)` queue, then `close()` store
Exception	`rollback_pending()` (discard queue only; flushed data remains), then `close()`
`rollback_on_error=False`	Keep pending queue after errors
`close_on_exit=False`	Leave store open (shared `HttpStore` on FastAPI `app.state`)

Session API at a glance

Method	Use when
`add(model)`	Append triples; never delete existing subject data
`put(model)`	Upsert with cascade and orphan cleanup (default for apps)
`delete(model)`	Remove owned triples for root + composition tree
`get(Model, iri, depth=0)`	Load one resource; `depth` 0–2 eager-loads relationships
`query(Model).where(...)`	Fluent SELECT compiled to SPARQL
`execute(sparql)`	Raw SPARQL SELECT
`flush()` / `rollback_pending()`	Control `put(..., flush=False)` queue
`expire(Model, iri)`	Evict cache by IRI (also drops matching pending `put`)
`expunge(model)`	Detach one instance from identity / hydration cache (store unchanged)
`expunge_all()`	Clear all identity and hydration entries (pending queue kept)
`refresh(model, *, depth=0)`	Reload from the store into the cached instance when present
`merge(model)`	Attach or reconcile with the session identity map (no store write)

Pending flush queue

Defer writes until commit:

with SPARQLSession() as session:
    session.put(a, flush=False)
    session.put(b, flush=False)
    session.flush()  # or rely on context manager exit

Warning

Pending models are not written to the store until flush() (or successful context-manager exit). get does not return the pending instance; identity for that subject is evicted when the pending put is queued. A failed flush() re-queues remaining models (0.2+). Calling close() with a non-empty pending queue raises RuntimeError.

Identity map

After put, get(Model, iri, depth=0) returns the same Python instance when relationships are not materialized on the object. Different depth values may cache separate hydrated views.

Cache control (0.9+)

Method	When to use
`expire(Model, iri)`	External graph change for one IRI; also removes a queued pending `put` for that subject
`expunge(model)`	Detach a specific instance (e.g. before long-lived work on a copy)
`expunge_all()`	Reset session caches between test cases; does not flush or clear the pending queue
`refresh(model, depth=...)`	Reload attributes from the store (updates the identity-map object in place when cached)
`merge(model)`	Re-attach a detached instance or copy field state onto the canonical session instance

transient → (add|put) → persistent (in store + identity map)
persistent → expunge → detached (may merge again)
persistent → refresh → persistent (reloaded from store)

refresh and merge do not write to the store — use put to persist changes. merge copies only fields you set on the detached instance (unset relationships are left unchanged on the cached object). refresh(..., depth=0) clears relationship attributes on the cached instance; a later get(..., depth≥1) reloads nested data from the store (0.9.1+). On HttpStore / AsyncHttpStore, get and refresh CONSTRUCT-pull when the subject is missing from the mirror (default mirror_mode="writer", 0.9.x+). Since 0.10.0, mirror_mode="remote_authoritative" pulls on every get/refresh, and all pulls use replace-on-pull (no stale predicates per IRI). Use expunge then get at the needed depth if you want a clean reload boundary.

Choosing a store

Store	When to use
`MemoryStore`	Tests, notebooks, single-process tools (default)
`HttpStore`	Fuseki, Jena, or any SPARQL 1.1 endpoint (`sparqlmodel[http]`)

Load an existing RDF file into a session with from_rdf_file() (see Real-world examples for full examples).

from sparqlmodel import HttpStore, SPARQLSession

store = HttpStore(
    "http://localhost:3030/ds/sparql",
    graph_store_url="http://localhost:3030/ds/data",
)
with SPARQLSession(store=store) as session:
    session.put(model)
# After external bulk changes to the remote dataset (0.12+)
store.sync_mirror()  # next get/refresh sees fresh mirror (0.13.1+ clears session cache)

# Multi-reader: re-sync each get from remote (0.10+)
with SPARQLSession(
    store=HttpStore("http://localhost:3030/ds/sparql", mirror_mode="remote_authoritative")
) as session:
    person = session.get(Person, IRI("urn:person:1"))

Composition vs reference

Relationship value	`put` / `delete`
Nested `SPARQLModel`	Composition — cascade + orphan cleanup
`Relationship(..., cascade=False)`	Reference — link only
`IRI`	Reference — update link; do not delete target resource

Threading

Sessions are not thread-safe. Use one session per request, task, or thread. Share the store (e.g. on app.state), not the session.

Next

Query DSL — filter and compile queries
FastAPI integration — wire sessions into ASGI apps
SparqlModel ORM guide — cascade and hydration in depth