Store helpers

pyoxigraph store factories and lifecycle helpers.

triplemodel.io.stores.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.stores.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.stores.cleanup_ephemeral_store_path(path)[source]

Remove a temporary on-disk store directory (non-throwing on failure).

Return type:

None

Parameters:

path (str)

triplemodel.io.stores.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.stores.coerce_store_name(store, *, stacklevel=2)[source]

Normalize store; map legacy backend alias names to disk.

Return type:

str

Parameters:

  • store (str)

  • stacklevel (int)

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

Remove an on-disk store directory.

Return type:

None

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

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

Return type:

Iterator[RdfGraph]

Parameters:

graph (RdfGraph)

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

Return IRIs of all named graphs in graph.store.

Return type:

list[str]

Parameters:

graph (RdfGraph)

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

Flush an on-disk store when supported.

Return type:

None

Parameters:

graph (RdfGraph)

triplemodel.io.stores.store_flush(graph)[source]

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

Return type:

None

Parameters:

graph (RdfGraph)

triplemodel.io.stores.store_rollback(graph)[source]

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

Return type:

None

Parameters:

graph (RdfGraph)

Store operations

pyoxigraph Store operations (bulk load, backup, named graphs, patterns).

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

Remove all quads from a named graph.

Return type:

None

Parameters:

  • graph (RdfGraph)

  • graph_iri (str)

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

Return IRIs of all named graphs in graph.store.

Return type:

list[str]

Parameters:

graph (RdfGraph)

triplemodel.store.ops.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.store.ops.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.store.ops.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.store.ops.store_flush(graph)[source]

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

Return type:

None

Parameters:

graph (RdfGraph)

SPARQL results I/O

Parse and serialize SPARQL result documents.

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

Parse a SPARQL results file into SparqlResult.

Return type:

SparqlResult

Parameters:

Canonicalize (in-memory)

Canonicalize quads in memory via pyoxigraph.Dataset.

triplemodel.store.canonicalize.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)