API reference

TripleModel — Pydantic models backed by RDF graphs via pyoxigraph.

Install: pip install triplemodel

triplemodel.Store

alias of RdfGraph

triplemodel.canonicalize_quads(quads, *, algorithm=<CanonicalizationAlgorithm unstable>)[source]

Return a canonical ordering of quads using pyoxigraph.Dataset.canonicalize().

Use for diff snapshots of in-memory data. Blank-node labels from a persistent Store are not stable when graph shape changes.

Return type:

list[Quad]

Parameters:

  • quads (Iterable[Quad])

  • algorithm (CanonicalizationAlgorithm)

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

Bases: object

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

Parameters:

  • code (str)

  • direction (str | None)

__init__(code, direction=None)
Parameters:

  • code (str)

  • direction (str | None)

Return type:

None

direction: str | None = None
code: str
triplemodel.all_from_dataset(dataset, model_cls, **kwargs)[source]

Load every resource of this model’s RDF type from its named graph context.

Return type:

list[TypeVar(T, bound= BaseModel)]

Parameters:

  • dataset (RdfDataset)

  • model_cls (type[T])

  • kwargs (Any)

triplemodel.all_from_dataset_dispatch(dataset, *, model_classes=None, validate_type=True, on_duplicate='warn', resolver=None, registry=<triplemodel.terms.registry.LiteralRegistry object>, de_skolemize=None)[source]

Load all subjects whose rdf:type maps to a registered class (per-class graph).

When model_classes is set, only those classes are loaded (recommended when the process has other registered models from unrelated modules).

Return type:

list[BaseModel]

Parameters:
triplemodel.all_from_graph_dispatch(graph, *, validate_type=True, on_duplicate='warn', resolver=None, registry=<triplemodel.terms.registry.LiteralRegistry object>, de_skolemize=None)[source]

Load all subjects that resolve to a registered model class.

Return type:

list[BaseModel]

Parameters:
triplemodel.dump_dataset(dataset, destination=None, *, format='trig', jsonld_context=None, **format_kwargs)[source]

Serialize dataset to a string, bytes, or file.

Return type:

str | bytes | None

Parameters:
triplemodel.dump_graph(graph, destination=None, *, format='turtle', jsonld_context=None, **format_kwargs)[source]

Serialize graph to a string, bytes, or file.

Return type:

str | bytes | None

Parameters:
triplemodel.dump_model(model, path, **kwargs)[source]

Write a model instance to an RDF file (alias for model.serialize).

Return type:

str | bytes | None

Parameters:
triplemodel.get_graph_context(container, graph_iri=None)[source]

Return the named-graph view for triple I/O within container.

Return type:

RdfGraph

Parameters:

  • container (RdfGraph | RdfDataset)

  • graph_iri (str | None)

triplemodel.graph_to_model_dispatch(graph, uri, *, validate_type=True, on_duplicate='warn', resolver=None, registry=<triplemodel.terms.registry.LiteralRegistry object>, de_skolemize=None)[source]

Hydrate using the most specific registered class for the subject’s types.

Return type:

BaseModel

Parameters:
triplemodel.graph_to_model_dispatch_from_dataset(dataset, uri, *, validate_type=True, on_duplicate='warn', resolver=None, registry=<triplemodel.terms.registry.LiteralRegistry object>, de_skolemize=None)[source]

Hydrate using the most specific registered class for the subject’s named graph.

Return type:

BaseModel

Parameters:
triplemodel.graph_to_model_from_dataset(dataset, model_cls, uri, *, graph_iri=None, config=None, validate_type=True, on_duplicate='warn', resolver=None, registry=<triplemodel.terms.registry.LiteralRegistry object>, de_skolemize=None)[source]

Construct an instance from triples in the model’s named graph context.

Return type:

TypeVar(T, bound= BaseModel)

Parameters:
triplemodel.graph_to_models_from_dataset(dataset, model_cls, *, graph_iri=None, type_uri=None, config=None, validate_type=True, on_duplicate='warn', resolver=None, registry=<triplemodel.terms.registry.LiteralRegistry object>, de_skolemize=None)[source]

Load all resources of this type from the model’s named graph context.

Return type:

list[TypeVar(T, bound= BaseModel)]

Parameters:
triplemodel.infer_format(hint, explicit_format=None)[source]

Resolve a parser/serializer format name for pyoxigraph.

Return type:

str

Parameters:

  • hint (str | Path | None)

  • explicit_format (str | None)

triplemodel.iter_model_quads(model, *, uri=None, graph_iri=None, config=None, resolver=None, registry=<triplemodel.terms.registry.LiteralRegistry object>)[source]

Yield (subject, predicate, object, graph_iri) rows for model.

Return type:

Iterator[tuple[str | NamedNode | BlankNode | Literal, str, Any, str | None]]

Parameters:
triplemodel.is_quad_format(fmt)[source]

Return True when fmt is a quad-aware format (TriG, N-Quads).

Return type:

bool

Parameters:

fmt (str)

triplemodel.load_dataset(source=None, *, data=None, format=None, base=None, bind_prefixes=None, jsonld_context=None, **format_kwargs)[source]

Parse RDF into an in-memory dataset (alias for parse_into_dataset()).

Return type:

RdfDataset

Parameters:
triplemodel.load_graph(source=None, *, data=None, format=None, base=None, bind_prefixes=None, jsonld_context=None, **format_kwargs)[source]

Parse RDF into an in-memory graph (alias for parse_into_graph()).

Return type:

RdfGraph

Parameters:
triplemodel.load_models(path, *model_classes, **kwargs)[source]

Load one or more model classes from a file (single parse for multiple classes).

Return type:

list[TypeVar(TModel, bound= BaseModel)] | dict[type[TypeVar(TModel, bound= BaseModel)], list[TypeVar(TModel, bound= BaseModel)]]

Parameters:
triplemodel.load_models_from_dataset(dataset, *model_classes, **kwargs)[source]

Load multiple model classes from one dataset (each uses its Rdf.graph_iri).

Return type:

dict[type[TypeVar(TModel, bound= BaseModel)], list[TypeVar(TModel, bound= BaseModel)]]

Parameters:

  • dataset (RdfDataset)

  • model_classes (type[TModel])

  • kwargs (Any)

triplemodel.load_models_from_graph(graph, *model_classes, **kwargs)[source]

Load multiple model classes from one graph without re-parsing.

Return type:

dict[type[TypeVar(TModel, bound= BaseModel)], list[TypeVar(TModel, bound= BaseModel)]]

Parameters:

  • graph (RdfGraph)

  • model_classes (type[TModel])

  • kwargs (Any)

triplemodel.load_models_streaming(path, *model_classes, store=None, chunk_size=500, store_identifier=None, **kwargs)[source]

Load models from a large file using chunked hydration (optional on-disk store).

For N-Triples / N-Quads, pass store='disk' (and optional store_identifier) to avoid holding the full graph in memory. Omit store for an in-memory parse. Turtle/TriG still require a full parse.

Return type:

list[TypeVar(TModel, bound= BaseModel)] | dict[type[TypeVar(TModel, bound= BaseModel)], list[TypeVar(TModel, bound= BaseModel)]]

Parameters:

  • path (str | Path)

  • model_classes (type[TModel])

  • store (str | None)

  • chunk_size (int)

  • store_identifier (str | None)

  • kwargs (Any)

triplemodel.open_graph(store, identifier='', *, create=True, read_only=False, ephemeral_store_path=None, **kwargs)[source]

Open an in-memory or on-disk graph backed by pyoxigraph.

store may be memory / default or disk. For disk, identifier is a directory path passed to pyoxigraph.Store.

When read_only=True, opens with pyoxigraph.Store.read_only() (undefined behavior if another process writes the same path). When create=False, raises FileNotFoundError if the directory does not exist.

Return type:

RdfGraph

Parameters:

  • store (str)

  • identifier (str)

  • create (bool)

  • read_only (bool)

  • ephemeral_store_path (str | None)

  • kwargs (Any)

triplemodel.parse_into_store_graph(path, *, store='disk', identifier=None, format=None, base=None, bind_prefixes=None, jsonld_context=None, **format_kwargs)[source]

Parse a document into a store-backed Graph (recommended for large N-Triples/N-Quads).

Defaults to an on-disk pyoxigraph.Store (store='disk'). Pass identifier for a persistent directory, or omit it to use a temporary directory removed by close().

Return type:

RdfGraph

Parameters:
triplemodel.model_to_dataset(model, dataset=None, *, uri=None, config=None, graph_iri=None, mode=None, bind=None, resolver=None, registry=<triplemodel.terms.registry.LiteralRegistry object>, skolemize=None)[source]

Add triples for model to the appropriate named graph in dataset.

Return type:

RdfDataset

Parameters:
triplemodel.models_to_dataset(models, dataset=None, *, mode='add', resolver=None, registry=<triplemodel.terms.registry.LiteralRegistry object>)[source]

Serialize multiple model instances into named graphs by resolve_graph_iri.

Return type:

RdfDataset

Parameters:
triplemodel.parse_into_dataset(source=None, *, data=None, format=None, base=None, bind_prefixes=None, jsonld_context=None, lenient=False, without_named_graphs=False, rename_blank_nodes=False, **format_kwargs)[source]

Parse RDF into a new in-memory RdfDataset.

Return type:

RdfDataset

Parameters:
triplemodel.parse_into_graph(source=None, *, data=None, format=None, base=None, bind_prefixes=None, jsonld_context=None, lenient=False, without_named_graphs=False, rename_blank_nodes=False, **format_kwargs)[source]

Parse RDF into a new in-memory RdfGraph.

Return type:

RdfGraph

Parameters:
triplemodel.ref_field(predicate, *, model, inverse=None, default=Ellipsis, **field_kwargs)[source]

Foreign-key field: import hydrates model from the object URI in the graph.

Return type:

TypeVar(_T)

Parameters:

  • predicate (str)

  • model (type[BaseModel])

  • inverse (str | None)

  • default (_T | EllipsisType)

  • field_kwargs (Unpack[RdfFieldKwargs])

triplemodel.parse_url_into_dataset(url, *, format=None, base=None, timeout=30.0, bind_prefixes=None, jsonld_context=None, lenient=False, without_named_graphs=False, rename_blank_nodes=False, **format_kwargs)[source]

Parse RDF from a URL into a RdfDataset.

Return type:

RdfDataset

Parameters:
triplemodel.parse_url_into_graph(url, *, format=None, base=None, timeout=30.0, bind_prefixes=None, jsonld_context=None, lenient=False, without_named_graphs=False, rename_blank_nodes=False, **format_kwargs)[source]

Parse RDF from a URL.

Return type:

RdfGraph

Parameters:
triplemodel.quads_in_context(dataset, graph_iri)[source]

Iterate quads in a named graph (or default graph when graph_iri is None).

Return type:

Iterator[tuple[NamedNode | BlankNode | Literal, NamedNode | BlankNode | Literal, NamedNode | BlankNode | Literal, NamedNode | BlankNode | Literal]]

Parameters:

  • dataset (RdfDataset)

  • graph_iri (str | None)

triplemodel.resolve_graph_iri(model, cfg=None)[source]

Return the named graph IRI for model (class config, then instance override).

Return type:

str | None

Parameters:
triplemodel.resolve_model_class(graph, subject, *, use_subclass=None)[source]

Pick the most specific registered class for subject’s rdf:type values.

Return type:

type[BaseModel]

Parameters:

  • graph (RdfGraph)

  • subject (NamedNode | BlankNode | Literal)

  • use_subclass (bool | None)

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

Bases: object

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

Parameters:

  • value (str)

  • lang (str | None)

  • direction (str | None)

__init__(value, lang=None, direction=None)
Parameters:

  • value (str)

  • lang (str | None)

  • direction (str | None)

Return type:

None

direction: str | None = None
lang: str | None = None
value: str
class triplemodel.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.

Parameters:
__init__(mapping=None, /, **langs)[source]
Parameters:
Return type:

None

classmethod from_mapping(mapping)[source]

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

Return type:

MultiLangString

Parameters:

mapping (Mapping[str, str | LangString | None])

get(lang, default=None)[source]
Return type:

LangString | None

Parameters:
values()[source]

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

Return type:

list[LangString]

by_lang: Mapping[str, LangString]
class triplemodel.LiteralRegistry[source]

Bases: object

Registry of Python type ↔ RDF literal converters.

__init__()[source]
Return type:

None

converter_for_datatype(datatype)[source]

Return import converter registered for an XSD datatype IRI.

Return type:

Callable[[Literal], str | int | float | bool | date | datetime | Decimal | UUID] | None

Parameters:

datatype (str)

converter_for_type(py_type)[source]
Return type:

tuple[Callable[..., Literal], Callable[[Literal], str | int | float | bool | date | datetime | Decimal | UUID]] | None

Parameters:

py_type (type)

literal_to_python(term, py_type)[source]
Return type:

Union[str, int, float, bool, date, datetime, Decimal, UUID, TypeVar(PyT), None]

Parameters:

  • term (Literal)

  • py_type (type[PyT] | type | None)

python_to_literal(value, py_type=None)[source]

Use registry for value when a converter is registered.

Return type:

Literal | None

Parameters:
register_literal_type(py_type, to_literal, from_literal, *, datatype=None)[source]

Register converters for a Python type.

Return type:

None

Parameters:

  • py_type (type[PyT])

  • to_literal (Callable[[PyT], Literal])

  • from_literal (Callable[[Literal], PyT])

  • datatype (str | None)

class triplemodel.OpaqueLiteral(value, datatype=None)[source]

Bases: object

A literal value preserved with its original datatype URI.

Parameters:

  • value (str)

  • datatype (str | None)

__init__(value, datatype=None)
Parameters:

  • value (str)

  • datatype (str | None)

Return type:

None

datatype: str | None = None
classmethod from_literal(term)[source]
Return type:

OpaqueLiteral

Parameters:

term (Literal)

to_literal()[source]
Return type:

Literal

value: str
class triplemodel.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.

Parameters:

  • value (str)

  • datatype (str | None)

__init__(value, datatype=None)[source]
Parameters:
Return type:

None

datatype: str | None = None
classmethod from_literal(term)[source]

Build from a pyoxigraph Literal (language tags are not preserved).

Return type:

TypedLiteral

Parameters:

term (Literal)

to_literal()[source]

Serialize to a pyoxigraph Literal.

Return type:

Literal

value: str
class triplemodel.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.

Parameters:
__init__(_graph=None, _parent_to_children=<factory>, _registered_inverse_forward=<factory>, _registered_inverse_reverse=<factory>, _graph_inverse_forward=<factory>, _graph_inverse_reverse=<factory>)
Parameters:
Return type:

None

classmethod from_graph(graph)[source]

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

Return type:

OntologyRegistry

Parameters:

graph (RdfGraph)

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

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

Return type:

OntologyRegistry

Parameters:
inverse_of(predicate_uri)[source]

Return the inverse property IRI for predicate_uri, if known.

Return type:

str | None

Parameters:

predicate_uri (str)

load_graph(graph)[source]

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

Return type:

None

Parameters:

graph (RdfGraph)

register_inverse(forward_predicate, inverse_predicate)[source]

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

Return type:

None

Parameters:

  • forward_predicate (str)

  • inverse_predicate (str)

register_subclasses(base_type_uri, subtype_uris)[source]

Register direct rdfs:subClassOf links without an ontology file.

Return type:

None

Parameters:
subtypes_of(type_uri)[source]

Return type_uri and all registered or inferred subclass IRIs.

Return type:

frozenset[str]

Parameters:

type_uri (str)

triplemodel.apply_hints_to_model(model_cls, registry, *, mutate=False, overwrite=False)[source]

Suggest or apply inverse= metadata from OntologyRegistry.inverse_of().

Returns {field_name: inverse_predicate_uri} for each mapped field that received a hint. When mutate is True, updates json_schema_extra on the class and calls model_rebuild() so inverse_for_field() sees the inverse IRI. Skips fields that already declare an inverse unless overwrite is True.

Return type:

dict[str, str]

Parameters:
class triplemodel.VocabularyRegistry(type_uri_to_class=<factory>, prefix_to_uri=<factory>)[source]

Bases: object

Maps RDF type IRIs and namespace prefixes to TripleModel classes.

Parameters:
__init__(type_uri_to_class=<factory>, prefix_to_uri=<factory>)
Parameters:
Return type:

None

bind_vocab(graph)[source]

Bind all registered prefixes on graph.

Return type:

None

Parameters:

graph (RdfGraph)

classmethod from_registered()[source]

Build a registry snapshot from process-wide registrations.

Return type:

VocabularyRegistry

model_for_subject(graph, subject, *, use_subclass=True)[source]
Return type:

type[BaseModel]

Parameters:

  • graph (RdfGraph)

  • subject (NamedNode | BlankNode | Literal)

  • use_subclass (bool)

model_for_type_uri(type_uri)[source]
Return type:

type[BaseModel] | None

Parameters:

type_uri (str)

register(model_cls)[source]

Register a model class and its Rdf metadata.

Return type:

None

Parameters:

model_cls (type[BaseModel])

type_uri_to_class: dict[str, type[BaseModel]]
prefix_to_uri: dict[str, str]
class triplemodel.ResourceRef(iri)[source]

Bases: object

Reference to an RDF resource identified by IRI.

Parameters:

iri (str)

__init__(iri)
Parameters:

iri (str)

Return type:

None

iri: str
class triplemodel.RdfResource(*args, **kwargs)[source]

Bases: Protocol

Marker protocol for Pydantic models that map to RDF resources.

__init__(*args, **kwargs)
class triplemodel.TripleModel(**data)[source]

Bases: BaseModel

Pydantic model that can be serialized to and from an RDF graph.

Subclasses declare RDF metadata on a nested Rdf class and map fields with rdf_field() or Annotated[..., Predicate(...)]. A nested Rdf on a subclass replaces the parent’s config entirely; do not declare an empty class Rdf: on a child if you intend to inherit the parent’s namespace, type_uri, or id_field.

Example:

class Person(TripleModel):
    class Rdf:
        namespace = "http://example.org/people/"
        type_uri = "http://xmlns.com/foaf/0.1/Person"
        id_field = "slug"

    slug: str
    name: str = rdf_field("http://xmlns.com/foaf/0.1/name")
classmethod all_from_dataset(dataset, *, graph_iri=None, type_uri=None, validate_type=True, on_duplicate='warn', resolver=None, registry=<triplemodel.terms.registry.LiteralRegistry object>, de_skolemize=None)[source]

Load every resource of this model’s RDF type from its named graph context.

Return type:

list[Self]

Parameters:
classmethod all_from_graph(graph, *, type_uri=None, validate_type=True, on_duplicate='warn', resolver=None, registry=<triplemodel.terms.registry.LiteralRegistry object>, de_skolemize=None)[source]

Load every resource of this model’s RDF type from graph.

Return type:

list[Self]

Parameters:
classmethod ask_sparql(graph, query, **kwargs)[source]

Execute an ASK query on graph.

Return type:

bool

Parameters:

  • graph (RdfGraph)

  • query (str)

  • kwargs (Any)

classmethod cbd(graph, uri, *, dispatch=False, validate_type=True, on_duplicate='warn', resolver=None, registry=<triplemodel.terms.registry.LiteralRegistry object>, de_skolemize=None, include_reifications=True)[source]

Load an instance from the concise bounded description around uri.

Return type:

Self

Parameters:
classmethod construct_from_sparql(graph, query, *, dispatch=False, graph_out=None, type_uri=None, validate_type=True, on_duplicate='warn', resolver=None, registry=<triplemodel.terms.registry.LiteralRegistry object>, de_skolemize=None, **kwargs)[source]

Run CONSTRUCT/DESCRIBE and load instances of this class.

Return type:

list[Self]

Parameters:
classmethod from_dataset(dataset, uri, *, graph_iri=None, validate_type=True, on_duplicate='warn', resolver=None, registry=<triplemodel.terms.registry.LiteralRegistry object>, de_skolemize=None)[source]

Construct an instance from triples in this model’s named graph context.

Return type:

Self

Parameters:
classmethod from_graph(graph, uri, *, validate_type=True, on_duplicate='warn', resolver=None, registry=<triplemodel.terms.registry.LiteralRegistry object>, de_skolemize=None)[source]

Construct an instance from triples about uri (IRI string or term).

Return type:

Self

Parameters:
classmethod load_sparql(endpoint, query, *, query_form=None, read_only=True, dispatch=False, type_uri=None, validate_type=True, on_duplicate='warn', resolver=None, registry=<triplemodel.terms.registry.LiteralRegistry object>, de_skolemize=None, **kwargs)[source]

Query a remote SPARQL endpoint and return instances.

Return type:

list[Self]

Parameters:
model_config: ClassVar[ConfigDict] = {'str_strip_whitespace': False, 'validate_assignment': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

classmethod parse(source=None, *, data=None, format=None, base=None, dispatch=False, type_uri=None, validate_type=True, on_duplicate='warn', resolver=None, registry=<triplemodel.terms.registry.LiteralRegistry object>, de_skolemize=None, lenient=False, without_named_graphs=False, rename_blank_nodes=False, **format_kwargs)[source]

Parse an RDF document and load model instances.

Return type:

list[Self]

Parameters:
classmethod parse_file(path, *, format=None, base=None, dispatch=False, type_uri=None, validate_type=True, on_duplicate='warn', resolver=None, registry=<triplemodel.terms.registry.LiteralRegistry object>, de_skolemize=None, **format_kwargs)[source]

Parse RDF from a local file path.

Return type:

list[Self]

Parameters:
classmethod parse_url(url, *, format=None, base=None, timeout=30.0, dispatch=False, type_uri=None, validate_type=True, on_duplicate='warn', resolver=None, registry=<triplemodel.terms.registry.LiteralRegistry object>, de_skolemize=None, **format_kwargs)[source]

Parse RDF from a URL.

Return type:

list[Self]

Parameters:
classmethod rdf_config()[source]

Return resolved RDF configuration for this model class.

Return type:

RdfConfig

classmethod select_from_sparql(graph, query, *, field_map=None, subject_var=None, hydrate=False, type_uri=None, validate_type=True, on_duplicate='warn', resolver=None, registry=<triplemodel.terms.registry.LiteralRegistry object>, de_skolemize=None, **kwargs)[source]

Run SELECT and build instances from bindings or hydration.

Return type:

list[Self]

Parameters:
serialize(*, format='turtle', destination=None, uri=None, mode=None, resolver=None, registry=<triplemodel.terms.registry.LiteralRegistry object>, skolemize=None, **format_kwargs)[source]

Serialize this instance to an RDF document string or file.

Return type:

str | bytes | None

Parameters:
subject_uri(*, uri=None)[source]

Return the RDF subject IRI for this instance.

Return type:

str

Parameters:

uri (str | None)

sync_to_dataset(dataset, *, uri=None, graph_iri=None, mode=None, resolver=None, registry=<triplemodel.terms.registry.LiteralRegistry object>, skolemize=None)[source]

Update the named graph for this instance within dataset.

Return type:

RdfDataset

Parameters:
sync_to_graph(graph, *, uri=None, mode=None, resolver=None, registry=<triplemodel.terms.registry.LiteralRegistry object>, skolemize=None)[source]

Update graph with owned triples for this instance (see mode).

When mode is omitted, uses Rdf.graph_mode if set to something other than "add"; otherwise defaults to "replace".

Return type:

RdfGraph

Parameters:
to_dataset(dataset=None, *, uri=None, graph_iri=None, mode=None, resolver=None, registry=<triplemodel.terms.registry.LiteralRegistry object>, skolemize=None)[source]

Serialize this instance into a named-graph dataset.

Return type:

RdfDataset

Parameters:
to_graph(graph=None, *, uri=None, mode=None, resolver=None, registry=<triplemodel.terms.registry.LiteralRegistry object>, skolemize=None)[source]

Serialize this instance into a Store graph.

When mode is omitted, uses Rdf.graph_mode (default "add").

Return type:

RdfGraph

Parameters:
to_triples(*, uri=None, resolver=None, registry=None)[source]

Export instance data as (subject, predicate, object) tuples.

Return type:

list[tuple[str | NamedNode | BlankNode | Literal, str, Union[str, int, float, bool, date, datetime, Decimal, UUID, Enum, NamedNode, BlankNode, Literal]]]

Parameters:
triplemodel.bind_namespaces(graph, prefixes, *, strategy='core')[source]

Bind prefixes on graph for serialization (recorded for Turtle output).

Return type:

None

Parameters:

  • graph (_NamespaceBindable)

  • prefixes (dict[str, str])

  • strategy (Literal['core', 'none'])

triplemodel.expand_curie(curie, prefixes)[source]

Expand prefix:local to a full IRI; pass through absolute IRIs.

Return type:

str

Parameters:
triplemodel.freeze_prefixes(raw)[source]
Return type:

Mapping[str, str]

Parameters:

raw (Mapping[str, str] | list[tuple[str, str]] | None)

triplemodel.graph_set(graph, subject, predicate, value, *, registry=<triplemodel.terms.registry.LiteralRegistry object>)[source]

Set objects for (subject, predicate) using remove-then-add semantics.

Return type:

None

Parameters:
triplemodel.graph_to_model(graph, model_cls, uri, *, config=None, validate_type=True, on_duplicate='warn', resolver=None, registry=<triplemodel.terms.registry.LiteralRegistry object>, de_skolemize=None, strict_import=None, warn_unmapped_fields=None)[source]

Hydrate a single model instance from triples about uri.

Return type:

TypeVar(T, bound= BaseModel)

Parameters:
triplemodel.graph_to_models(graph, model_cls, *, chunk_size=None, type_uri=None, config=None, validate_type=True, on_duplicate='warn', resolver=None, registry=<triplemodel.terms.registry.LiteralRegistry object>, de_skolemize=None, strict_import=None, warn_unmapped_fields=None)[source]

Load all resources of type_uri (or the model’s configured type) as models.

Return type:

list[TypeVar(T, bound= BaseModel)]

Parameters:
triplemodel.graph_store_session(graph)[source]

Yield graph (pyoxigraph handles persistence for on-disk stores).

Return type:

Iterator[RdfGraph]

Parameters:

graph (RdfGraph)

triplemodel.graph_value(graph, subject, predicate, model_cls, field_name, *, on_duplicate='first', resolver=None, registry=<triplemodel.terms.registry.LiteralRegistry object>)[source]

Return a single object for a functional-property field, if present.

Return type:

Union[str, int, float, bool, date, datetime, Decimal, UUID, None]

Parameters:
triplemodel.backup_store(target_directory, *, graph=None, store_path=None)[source]

Create a backup of an on-disk pyoxigraph store directory.

Return type:

None

Parameters:

  • target_directory (str | Path)

  • graph (RdfGraph | None)

  • store_path (str | Path | None)

triplemodel.bulk_load_into_graph(graph, source=None, *, data=None, format=None, base_iri=None, to_graph=None, lenient=False)[source]

Load RDF from a file or bytes into graph.store via pyoxigraph.Store.bulk_load().

Return type:

None

Parameters:

  • graph (RdfGraph)

  • source (str | Path | None)

  • data (str | bytes | None)

  • format (str | None)

  • base_iri (str | None)

  • to_graph (str | NamedNode | DefaultGraph | None)

  • lenient (bool)

triplemodel.clear_named_graph(graph, graph_iri)[source]

Remove all quads from a named graph.

Return type:

None

Parameters:

  • graph (RdfGraph)

  • graph_iri (str)

triplemodel.destroy_store(identifier, *, store='disk', **kwargs)[source]

Remove an on-disk store directory.

Return type:

None

Parameters:
triplemodel.dump_store(output, *, graph=None, store_path=None, format='nquads', from_graph=None, prefixes=None, base_iri=None)[source]

Dump an on-disk store to a file (default N-Quads).

Return type:

None

Parameters:

  • output (str | Path)

  • graph (RdfGraph | None)

  • store_path (str | Path | None)

  • format (str)

  • from_graph (str | NamedNode | DefaultGraph | None)

  • prefixes (dict[str, str] | None)

  • base_iri (str | None)

triplemodel.ensure_named_graph(graph, graph_iri)[source]

Ensure a named graph exists in the store.

Return type:

None

Parameters:

  • graph (RdfGraph)

  • graph_iri (str)

triplemodel.iter_quads_for_pattern(graph, subject=None, predicate=None, obj=None, *, graph_iri=None)[source]

Iterate quads matching an optional (s, p, o, graph) pattern on graph.store.

Return type:

Iterator[Quad]

Parameters:

  • graph (RdfGraph)

  • subject (Any)

  • predicate (Any)

  • obj (Any)

  • graph_iri (str | NamedNode | DefaultGraph | None)

triplemodel.list_named_graphs(graph)[source]

Return IRIs of all named graphs in graph.store.

Return type:

list[str]

Parameters:

graph (RdfGraph)

triplemodel.load_store(graph, source=None, *, data=None, format='nquads', base_iri=None, to_graph=None, lenient=False)[source]

Load a dump file into an on-disk store backing graph.

Return type:

None

Parameters:

  • graph (RdfGraph)

  • source (str | Path | None)

  • data (str | bytes | None)

  • format (str)

  • base_iri (str | None)

  • to_graph (str | NamedNode | DefaultGraph | None)

  • lenient (bool)

triplemodel.optimize_store(*, graph=None, store_path=None)[source]

Optimize an on-disk store after bulk import or heavy updates.

Return type:

None

Parameters:

  • graph (RdfGraph | None)

  • store_path (str | Path | None)

triplemodel.parse_query_results(source=None, *, data=None, format=None, path=None)[source]

Parse a SPARQL results file into SparqlResult.

Return type:

SparqlResult

Parameters:
triplemodel.remove_named_graph(graph, graph_iri)[source]

Remove a named graph and its quads from the store.

Return type:

None

Parameters:

  • graph (RdfGraph)

  • graph_iri (str)

triplemodel.iter_graph_to_models(graph, model_cls, *, chunk_size=500, type_uri=None, config=None, validate_type=True, on_duplicate='warn', resolver=None, registry=<triplemodel.terms.registry.LiteralRegistry object>, de_skolemize=None, strict_import=None, warn_unmapped_fields=None)[source]

Yield chunks of model instances loaded from graph.

Return type:

Iterator[list[TypeVar(T, bound= BaseModel)]]

Parameters:
triplemodel.id_from_subject_uri(namespace, uri)[source]

Extract the id segment from uri when it was built from namespace.

Return type:

str | None

Parameters:
triplemodel.merge_graphs(*graphs)[source]

Return a new graph containing the union of graphs.

Return type:

RdfGraph

Parameters:

graphs (RdfGraph)

triplemodel.model_to_graph(model, graph=None, *, uri=None, config=None, mode=None, bind=None, resolver=None, registry=<triplemodel.terms.registry.LiteralRegistry object>, skolemize=None)[source]

Add triples for model to graph (or a new graph) and return it.

Return type:

RdfGraph

Parameters:
triplemodel.model_to_triples(model, *, uri=None, config=None, resolver=None, registry=None)[source]

Return (subject, predicate, object) tuples for a model instance.

list[T] fields are omitted here; use model_to_graph() or to_graph() for full export including rdf:List.

registry is accepted for API symmetry with graph writers; literal conversion happens when triples are added to a graph (graph_set_many).

Return type:

list[tuple[str | NamedNode | BlankNode | Literal, str, Union[str, int, float, bool, date, datetime, Decimal, UUID, Enum, NamedNode, BlankNode, Literal]]]

Parameters:
triplemodel.models_to_graph(models, graph=None, *, mode='add', resolver=None, registry=<triplemodel.terms.registry.LiteralRegistry object>)[source]

Serialize multiple model instances into one graph.

Return type:

RdfGraph

Parameters:
triplemodel.objects_for_field(graph, uri, model_cls, field_name, *, resolver=None, registry=<triplemodel.terms.registry.LiteralRegistry object>)[source]

Return all RDF objects for a model field’s predicate.

Return type:

list[Union[str, int, float, bool, date, datetime, Decimal, UUID]]

Parameters:
triplemodel.rdf_field(predicate, *, inverse=None, back_populates=None, literal_datatype=None, transitive=False, default=Ellipsis, **field_kwargs)[source]

Create a Pydantic field bound to an RDF predicate.

Return type:

TypeVar(_T)

Parameters:

  • predicate (str)

  • inverse (str | None)

  • back_populates (Any | None)

  • literal_datatype (str | None)

  • transitive (bool)

  • default (_T | EllipsisType)

  • field_kwargs (Unpack[RdfFieldKwargs])

triplemodel.register_literal_type(py_type, to_literal, from_literal, *, datatype=None)[source]

Register converters on the package-default default_registry.

Return type:

None

Parameters:

  • py_type (type[PyT])

  • to_literal (Callable[[PyT], Literal])

  • from_literal (Callable[[Literal], PyT])

  • datatype (str | None)

triplemodel.register_rdf_resource(model_cls)[source]

Record a TripleModel subclass for nested-embed detection.

Return type:

None

Parameters:

model_cls (type)

triplemodel.store_commit(graph)[source]

Flush an on-disk store when supported.

Return type:

None

Parameters:

graph (RdfGraph)

triplemodel.store_flush(graph)[source]

Flush pending writes on an on-disk store (no-op if unsupported).

Return type:

None

Parameters:

graph (RdfGraph)

triplemodel.store_rollback(graph)[source]

No-op for pyoxigraph (transactions are not exposed on Store).

Return type:

None

Parameters:

graph (RdfGraph)

triplemodel.subject_base(namespace)[source]

Return the prefix used when appending an id to namespace.

Return type:

str

Parameters:

namespace (str)

triplemodel.sync_to_dataset(model, dataset, *, uri=None, graph_iri=None, mode=None, config=None, bind=True, resolver=None, registry=<triplemodel.terms.registry.LiteralRegistry object>, skolemize=None)[source]

Sync model into the resolved named graph within dataset.

Return type:

RdfDataset

Parameters:
triplemodel.sync_to_graph(model, graph=None, *, uri=None, mode=None, config=None, bind=True, resolver=None, registry=<triplemodel.terms.registry.LiteralRegistry object>, skolemize=None)[source]

Write model into graph using mode sync semantics.

Return type:

RdfGraph

Parameters:
class triplemodel.GraphDiff(only_in_first, only_in_second)[source]

Bases: object

Structural difference between two RDF graphs.

Parameters:
__init__(only_in_first, only_in_second)
Parameters:
Return type:

None

property equal: bool
only_in_first: frozenset[tuple[str, str, str]]
only_in_second: frozenset[tuple[str, str, str]]
triplemodel.apply_update(graph, update, *, model_cls=None, initNs=None, initBindings=None, use_store_provided=True, **kwargs)[source]

Apply a SPARQL UPDATE to graph (in-memory models may be stale afterward).

Return type:

None

Parameters:
triplemodel.ask(graph, query, *, model_cls=None, initNs=None, initBindings=None, use_store_provided=True, **kwargs)[source]

Execute an ASK query and return the boolean result.

Return type:

bool

Parameters:
triplemodel.cbd_graph(graph, subject, *, target_graph=None, include_reifications=True)[source]

Return the concise bounded description of subject in graph.

Return type:

RdfGraph

Parameters:

  • graph (RdfGraph)

  • subject (str | NamedNode | BlankNode | Literal)

  • target_graph (RdfGraph | None)

  • include_reifications (bool)

triplemodel.cbd_model(model_cls, graph, uri, *, dispatch=False, validate_type=True, on_duplicate='warn', resolver=None, registry=<triplemodel.terms.registry.LiteralRegistry object>, de_skolemize=None, include_reifications=True)[source]

Load a model instance from the CBD subgraph around uri.

Return type:

TypeVar(T, bound= BaseModel)

Parameters:
triplemodel.construct_models(model_cls, graph, query, *, dispatch=False, graph_out=None, type_uri=None, validate_type=True, on_duplicate='warn', resolver=None, registry=<triplemodel.terms.registry.LiteralRegistry object>, de_skolemize=None, initNs=None, initBindings=None, use_store_provided=True, **kwargs)[source]

Run CONSTRUCT/DESCRIBE and load models from the result graph.

Return type:

list[TypeVar(T, bound= BaseModel)]

Parameters:
triplemodel.detect_query_form(query)[source]

Return the first SPARQL query form keyword in query.

Return type:

Literal['select', 'construct', 'describe', 'ask', 'unknown']

Parameters:

query (str)

triplemodel.graph_diff(graph_a, graph_b, *, normalize_bnodes=True)[source]

Compare two graphs by normalized triple sets (after optional skolemization).

Return type:

GraphDiff

Parameters:

  • graph_a (RdfGraph)

  • graph_b (RdfGraph)

  • normalize_bnodes (bool)

triplemodel.graph_from_construct_result(result, graph_out=None)[source]

Merge a CONSTRUCT/DESCRIBE Result graph into graph_out or a new graph.

Return type:

RdfGraph

Parameters:

  • result (SparqlResult)

  • graph_out (RdfGraph | None)

triplemodel.graphs_equal(graph_a, graph_b, *, normalize_bnodes=True)[source]

Return whether graph_a and graph_b are isomorphic RDF graphs.

Return type:

bool

Parameters:

  • graph_a (RdfGraph)

  • graph_b (RdfGraph)

  • normalize_bnodes (bool)

triplemodel.hydrate_refs(instances, graph, *field_names, spec=None, validate_type=True, on_duplicate='warn', resolver=None, registry=<triplemodel.terms.registry.LiteralRegistry object>, de_skolemize=None)[source]

Batch-load reference fields from graph, reusing one model per object URI.

Return type:

list[TypeVar(T, bound= BaseModel)]

Parameters:
triplemodel.init_bindings_from_model(instance, mapping)[source]

Map SPARQL variable names to RDF terms from model field values.

Return type:

dict[Variable, NamedNode | BlankNode | Literal]

Parameters:
triplemodel.init_ns_from_model(model_cls)[source]

Build prefix map for SPARQL from model_cls Rdf.prefixes.

Return type:

dict[str, str]

Parameters:

model_cls (type[BaseModel])

triplemodel.load_sparql(model_cls, endpoint, query, *, query_form=None, read_only=True, dispatch=False, type_uri=None, validate_type=True, on_duplicate='warn', resolver=None, registry=<triplemodel.terms.registry.LiteralRegistry object>, de_skolemize=None, initNs=None, initBindings=None, use_store_provided=True, **kwargs)[source]

Query a remote SPARQL endpoint and return model instances.

Uses open_sparql_graph(), which raises NotImplementedError in 0.10.0. Query a local Store with construct_models() / select_models() after fetching data, or use SparqlModel for remote endpoints.

Return type:

list[TypeVar(T, bound= BaseModel)]

Parameters:
triplemodel.model_diff(model_a, model_b, graph=None, *, normalize_bnodes=True)[source]

Return field-level differences between two model instances.

When graph is provided, also includes graph_diff between model_a.to_graph() and model_b.to_graph() merged into the source graph.

Return type:

dict[str, Any]

Parameters:

  • model_a (T)

  • model_b (T)

  • graph (RdfGraph | None)

  • normalize_bnodes (bool)

triplemodel.model_join(instances, graph, spec, **kwargs)[source]

Hydrate reference fields named in spec (field name → model class).

Return type:

list[TypeVar(T, bound= BaseModel)]

Parameters:
triplemodel.open_sparql_graph(endpoint, *, read_only=True)[source]

Open a remote SPARQL graph (removed in 0.10.0).

Raises NotImplementedError. Load remote data into a local Store or use SparqlModel for session-level remote stores.

Return type:

RdfGraph

Parameters:
triplemodel.prepare_model_query(model_cls, query)[source]

Prepare query (prefixes applied at execution from model_cls).

Return type:

PreparedModelQuery

Parameters:
triplemodel.resolve_model_class_with_rdfs(graph, subject, *, use_subclass=True)[source]

Pick the most specific registered class for subject’s types.

Return type:

type[BaseModel]

Parameters:

  • graph (RdfGraph)

  • subject (NamedNode | BlankNode | Literal)

  • use_subclass (bool)

triplemodel.run_sparql(graph, query, *, model_cls=None, initNs=None, initBindings=None, use_store_provided=True, use_default_graph_as_union=False, default_graph=None, named_graphs=None, base_iri=None, **kwargs)[source]

Run SPARQL on graph.store with optional prefixes and bindings.

Return type:

SparqlResult

Parameters:

  • graph (RdfGraph)

  • query (str)

  • model_cls (type[BaseModel] | None)

  • initNs (Mapping[str, Any] | None)

  • initBindings (Mapping[Variable, NamedNode | BlankNode | Literal] | Mapping[str, NamedNode | BlankNode | Literal] | None)

  • use_store_provided (bool)

  • use_default_graph_as_union (bool)

  • default_graph (str | NamedNode | DefaultGraph | None)

  • named_graphs (list[str | NamedNode] | None)

  • base_iri (str | None)

  • kwargs (Any)

triplemodel.select_models(model_cls, graph, query, *, field_map=None, subject_var=None, hydrate=False, type_uri=None, validate_type=True, on_duplicate='warn', resolver=None, registry=<triplemodel.terms.registry.LiteralRegistry object>, de_skolemize=None, initNs=None, initBindings=None, use_store_provided=True, **kwargs)[source]

Run SELECT and return model instances from result bindings or hydration.

Return type:

list[TypeVar(T, bound= BaseModel)]

Parameters:
triplemodel.subject_type_closure(graph, subject)[source]

Return rdf:type IRIs for subject plus rdfs:subClassOf ancestors.

Return type:

frozenset[str]

Parameters:

  • graph (RdfGraph)

  • subject (NamedNode | BlankNode | Literal)

triplemodel.subclass_uris(graph, type_uri)[source]

Return type_uri and all superclasses via rdfs:subClassOf.

Return type:

frozenset[str]

Parameters:

  • graph (RdfGraph)

  • type_uri (str)

triplemodel.transitive_objects(graph, subject, predicate)[source]

Return object IRIs reachable from subject along predicate (transitive).

Return type:

list[str]

Parameters:

  • graph (RdfGraph)

  • subject (str | NamedNode | BlankNode | Literal)

  • predicate (str)

triplemodel.transitive_subjects(graph, predicate, obj)[source]

Return subject IRIs that reach obj along predicate (transitive).

Return type:

list[str]

Parameters:

  • graph (RdfGraph)

  • predicate (str)

  • obj (str | NamedNode | BlankNode | Literal)

Module reference