sparqlmodel 0.9.2

SPARQL ORM for Python — sessions, queries, and graph persistence on RDF stores

SPARQLModel

The SQLModel of SPARQL — Pydantic v2 entity models mapped to RDF, a persistent session, and Python filters that compile to SPARQL.

Build knowledge-graph and metadata apps with typed SPARQLModel classes, with SPARQLSession() as session:, and ORM-style put, get, nested relationships, and a query builder — on in-memory graphs or remote SPARQL 1.1 endpoints. Same validation ergonomics as FastAPI and SQLModel: invalid data fails at construction and on load, before bad triples reach the store.

Requires Python 3.10+ · Built on TripleModel 0.10 + pyoxigraph · Changelog (0.9.2)

---

Features

Area	What you get
Models	SPARQLModel, Field, Relationship, IRI — Pydantic v2 validation (model_validate, constraints, extra="forbid")
RDF mapping	rdf_type, compact predicates, TripleModel sync_to_graph / from_graph under the hood
Session	add, put, delete, get, identity map, flush / pending queue (sync and async since 0.6)
Queries	session.query(Person).where(Person.name == "x") → SPARQL (&, |, in_, comparisons, multi-hop)
Stores	MemoryStore / AsyncMemoryStore; HttpStore / AsyncHttpStore for Fuseki/Jena ([http])
FastAPI	SessionDep or AsyncSessionDep, lifespan helpers, Turtle/JSON-LD responses
Cascade	Composition on put/delete; Relationship(..., cascade=False) for references

---

Install

pip install sparqlmodel

pip install "sparqlmodel[http]"      # HttpStore + AsyncHttpStore (httpx)
pip install "sparqlmodel[fastapi]"   # FastAPI session + RDF responses
pip install -e ".[dev,http,fastapi]" # development (includes pytest-asyncio for async tests)

For local development with uv, sync dev extras so async tests run: uv sync --extra dev.

---

Quickstart

from sparqlmodel import Field, IRI, Relationship, SPARQLModel, SPARQLSession

class Organization(SPARQLModel):
    rdf_type = "schema:Organization"
    __prefixes__ = {"schema": "https://schema.org/"}

    id: IRI
    name: str = Field("schema:name")

class Person(SPARQLModel):
    rdf_type = "schema:Person"
    __prefixes__ = {"schema": "https://schema.org/"}

    id: IRI
    name: str = Field("schema:name")
    works_for: Organization | None = Relationship(
        "schema:worksFor", model=Organization
    )

acme = Organization(id=IRI("urn:org:acme"), name="Acme Corp")
odos = Person(id=IRI("urn:person:odos"), name="Odos", works_for=acme)

with SPARQLSession() as session:
    session.put(odos)

    found = session.query(Person).where(Person.name == "Odos").first()
    team = session.query(Person).where(Person.works_for.name == "Acme Corp").all()
    full = session.get(Person, odos.id, depth=1)

---

Pydantic models

SPARQLModel subclasses pydantic.BaseModel. You get the same advantages as in FastAPI or SQLModel: typed fields, IDE support, and validation on create and on load.

When	What runs
Person(...) / API body	Pydantic validates types and Field constraints
session.put(model)	Validated instance → sync_to_graph (0.4+: same SPARQLModel instance subclasses TripleModel)
session.get / query hydration	Graph → model_validate → SPARQLModel instance

# Field forwards pydantic.Field kwargs (min_length, ge, description, …)
class Person(SPARQLModel):
    rdf_type = "schema:Person"
    __prefixes__ = {"schema": "https://schema.org/"}
    id: IRI
    name: str = Field("schema:name", min_length=1)

• extra="forbid" — unknown fields on a model raise at validation time (safer for APIs).
• FastAPI — reuse the same SPARQLModel classes for request/response bodies (see FastAPI below).
• JSON-LD — model_dump_jsonld() / model_validate_jsonld() for API dicts (cascade-aware); files and HTTP bodies use model.serialize(format="json-ld") or Person.parse(...).

Details: Models guide · ORM guide

---

Session

SPARQLSession is the unit of work. Use it as a context manager: flush pending writes on success, roll back the pending queue on error, close HTTP stores when done.

Method	Purpose
add(model)	Append triples (no delete of existing subject data)
put(model)	Upsert with cascade and orphan cleanup
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 query; filters compile to SPARQL
execute(sparql)	Raw SPARQL SELECT (auto-prefixes when configured)
flush() / rollback_pending()	Apply or discard put(..., flush=False) queue
expire(Model, iri)	Evict identity map and hydration cache

Nested SPARQLModel values are composition (cascade on put/delete). Use Relationship(..., cascade=False) or an IRI when the target is owned elsewhere.

---

Query DSL

with SPARQLSession() as session:
    session.query(Person).where(Person.name == "Odos").all()

    session.query(Person).where(
        (Person.name == "Odos") | (Person.name == "Ada")
    ).all()

    session.query(Person).where(
        Person.works_for.located_in.name == "Boston"
    ).all(depth=2)

    session.query(Person).where(Person.name.in_(("Odos", "Ada"))).all()

    session.query(Person).where(Person.name != "Other").all()
    # pre-0.5.2 inequality (excludes unbound): .use_inequality_for_ne()

Operators: ==, !=, &, |, <, >, <=, >=, .in_(tuple) or .in_(list) (not a bare string — use ("x",) for one value), multi-hop paths (Person.works_for.name), .limit(n), .offset(n), .order_by(field, desc=False), .count(), .is_(None) / .is_not(None) on nullable relationships, .first() always LIMIT 1 (ignores prior .limit() and .offset()), .use_inequality_for_ne(), .use_optional_for_comparisons() (NE semantics toggle; real OPTIONAL blocks are automatic on nullable hops).

---

Stores

MemoryStore (default) — in-memory triplemodel.Store (pyoxigraph); tests and single-process apps:

with SPARQLSession() as session:
    session.put(model)

HttpStore — SPARQL 1.1 over HTTP with a local mirror for get and cascade (sparqlmodel[http]):

from sparqlmodel import HttpStore, SPARQLSession

with SPARQLSession(store=HttpStore("http://localhost:3030/ds/sparql")) as session:
    session.put(odos)

query / execute use the remote endpoint; get and cascade read the mirror updated by this store’s writes. See the production guide for mirror semantics and deployment notes.

---

FastAPI

Per-request sessions with a shared store — same pattern as SQLModel + SQLAlchemy:

from contextlib import asynccontextmanager

from fastapi import FastAPI, HTTPException, Request
from sparqlmodel import IRI
from sparqlmodel.fastapi import SessionDep, http_store_lifespan, negotiated_response

@asynccontextmanager
async def lifespan(app: FastAPI):
    async with http_store_lifespan(app, "http://localhost:3030/ds/sparql"):
        yield

app = FastAPI(lifespan=lifespan)

@app.get("/person/{iri}")
def person(iri: str, request: Request, session: SessionDep) -> object:
    model = session.get(Person, IRI(iri))
    if model is None:
        raise HTTPException(status_code=404)
    return negotiated_response(request, model)

---

Export

print(odos.serialize(format="turtle"))

# or backward-compatible wrappers:
from sparqlmodel.serializers import export_model
print(export_model(odos, format="turtle"))

File parse/serialize is implemented by TripleModel (parse, serialize, load_graph). See the roadmap.

---

Documentation

Guide	Description
Read the Docs	Full site: install, guides, API reference, troubleshooting
Getting started	Quickstart and first session
Guides	Models (Pydantic), sessions, queries, FastAPI
Real-world examples	Nobel, DCAT, Wikidata, Schema.org (examples/realworld/)
ORM guide	Lifecycle, cascade, hydration, when to use SparqlModel vs TripleModel
Technical specification	Normative API; production checklist
Production guide	HttpStore, sessions, deployment
Roadmap	0.5–1.3 milestones; SQLModel parity
Project plan	Vision and release strategy
Ecosystem	SparqlModel vs TripleModel boundaries

---

Known limitations (0.9.2)

• Multi-valued predicates: first value per predicate on load; prefer put over add for upserts
• HttpStore / AsyncHttpStore (0.9.1+): get and refresh (0.9.2+) pull missing subjects from the remote endpoint into the mirror (CONSTRUCT); full mirror reconciliation and multi-writer sync remain planned 1.0 — see roadmap
• Use merge / refresh / expunge for explicit identity-map control (sessions guide)
• session.graph is a triplemodel.Store (pyoxigraph), not an rdflib Graph — use TripleModel I/O for file round-trip
• Default != uses NOT EXISTS (includes resources with no value); .use_inequality_for_ne() on nullable hops also treats missing links as matching
• ==, <, >, and in_ on optional paths still exclude unbound values (SPARQL-native)
• Nullable relationship filters use OPTIONAL hops; required (non-nullable) hops still use inner-join semantics
• Sessions are not thread-safe; one session per request/task
• Each model field must map to a unique RDF predicate; duplicate predicates raise ConfigurationError at class definition
• Cyclic embedded models raise ConfigurationError on put / model_to_graph
• Shared embedded resources referenced from multiple roots are preserved on put when another subject still links to them

---

License

MIT — see LICENSE.