triplemodel.io

Graph I/O: export, import, sync, and helpers.

triplemodel.io.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.io.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.io.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:
class triplemodel.io.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.io.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.io.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.io.discover_subject_uris(graph, model_cls, cfg, *, resolver=None)[source]

URIRef subjects with at least one owned predicate triple (excluding rdf:type).

Return type:

list[str]

Parameters:
triplemodel.io.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.io.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.io.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.io.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.io.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.io.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.io.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.io.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.io.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.io.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)

triplemodel.io.discover_subjects_by_instance_of(graph, cfg)[source]

Subjects with (subject, instance_of, type) per RdfConfig.instance_of.

Return type:

list[str]

Parameters:
triplemodel.io.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.io.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.io.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.io.fetch_url(url, *, timeout=30.0)[source]

Download url and return the response body.

Return type:

bytes

Parameters:
triplemodel.io.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.io.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.io.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.io.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.io.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.io.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.io.is_quad_format(fmt)[source]

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

Return type:

bool

Parameters:

fmt (str)

triplemodel.io.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.io.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.io.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.io.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.io.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.io.clear_named_graph(graph, graph_iri)[source]

Remove all quads from a named graph.

Return type:

None

Parameters:

  • graph (RdfGraph)

  • graph_iri (str)

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

Remove an on-disk store directory.

Return type:

None

Parameters:
triplemodel.io.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.io.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.io.graph_store_session(graph)[source]

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

Return type:

Iterator[RdfGraph]

Parameters:

graph (RdfGraph)

triplemodel.io.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.io.list_named_graphs(graph)[source]

Return IRIs of all named graphs in graph.store.

Return type:

list[str]

Parameters:

graph (RdfGraph)

triplemodel.io.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.io.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.io.parse_query_results(source=None, *, data=None, format=None, path=None)[source]

Parse a SPARQL results file into SparqlResult.

Return type:

SparqlResult

Parameters:
triplemodel.io.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.io.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.io.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.io.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.io.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.io.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.io.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.io.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.io.store_commit(graph)[source]

Flush an on-disk store when supported.

Return type:

None

Parameters:

graph (RdfGraph)

triplemodel.io.store_flush(graph)[source]

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

Return type:

None

Parameters:

graph (RdfGraph)

triplemodel.io.store_rollback(graph)[source]

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

Return type:

None

Parameters:

graph (RdfGraph)

triplemodel.io.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.io.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.io.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.io.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.io.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.io.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.io.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.io.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.io.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.io.graph_set_many(graph, subject, predicate, values, *, registry=<triplemodel.terms.registry.LiteralRegistry object>)[source]

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

Return type:

None

Parameters:
triplemodel.io.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.io.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.io.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.io.import_field_value(graph, objects, field_info, field_name, predicate, uri, *, embed, on_duplicate, registry=<triplemodel.terms.registry.LiteralRegistry object>, de_skolemize=False)[source]

Hydrate a single model field from RDF objects (used by graph_to_model()).

Return type:

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

Parameters:

  • graph (RdfGraph)

  • objects (list[NamedNode | BlankNode | Literal | Triple])

  • field_info (FieldInfo)

  • field_name (str)

  • predicate (str)

  • uri (str)

  • embed (Literal['iri', 'bnode'])

  • on_duplicate (Literal['ignore', 'warn', 'error', 'first'])

  • registry (LiteralRegistry)

  • de_skolemize (bool)

triplemodel.io.merge_graphs(*graphs)[source]

Return a new graph containing the union of graphs.

Return type:

RdfGraph

Parameters:

graphs (RdfGraph)

triplemodel.io.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.io.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.io.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.io.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.io.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:
triplemodel.io.write_model_add(graph, model, *, uri=None, config=None, bind=False, resolver=None, registry=<triplemodel.terms.registry.LiteralRegistry object>, skolemize=None)[source]

Append triples for model to graph.

Return type:

RdfGraph

Parameters: