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.12+ + pyoxigraph · Changelog (0.13.1)

Features

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.

# 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.

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",
        graph_store_url="http://localhost:3030/ds/data",
    )
) 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

Known limitations (0.13.1)

• list[SPARQLModel] embed collections are not supported (TripleModel); use set[IRI] / set[ResourceRef] with Relationship(..., model=...) for multi-ref
• Multi-valued scalars and refs (set/list), LangString / MultiLangString, polymorphic query, property paths, and not_() — see Models and Queries guides
• Prefer put over add for upserts (stale literal cleanup)
• HttpStore / AsyncHttpStore: default mirror_mode="writer" pulls only when a subject is missing from the mirror; use mirror_mode="remote_authoritative" (0.10+), pull_subjects_into_mirror, or sync_mirror() (0.12+, requires graph_store_url) when reads must match remote updates. Retries and batched UPDATE: PRODUCTION; GSP mirror sync: PRODUCTION
• 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.