flpq_data.graphs.readwrite.rdf

Read (and write) a graph from (and to) RDF file.

Functions

graph_from_rdf(path)	Loads a graph from RDF file.
graph_to_rdf(graph, path)	Saves the graph to the RDF file by path.
iter_edges_from_rdf(path)	Yields the edges of an RDF file as a stream.
rdf_from_edges(edges, path)	Writes an edge stream to an RDF file.

graph_from_rdf(path: Path | str) → MultiDiGraph

Loads a graph from RDF file.

Parameters:

pathUnion[Path, str]

The path to the RDF file with which the graph will be created.

Returns:

gMultiDiGraph

Loaded graph.

Examples

>>> from flpq_data import *
>>> import pathlib, tempfile
>>> d = pathlib.Path(tempfile.mkdtemp())
>>> g = graph_from_text(["0 a 1", "1 b 2"])
>>> path = graph_to_rdf(g, d / "g.ttl")
>>> generations = graph_from_rdf(path)
>>> generations.number_of_nodes()
3
>>> generations.number_of_edges()
2

graph_to_rdf(graph: MultiDiGraph, path: Path | str) → Path

Saves the graph to the RDF file by path.

The file holds valid RDF 1.1 Turtle — one triple per edge with IRI nodes and predicates (see rdf_from_edges() and the “RDF encoding” section of docs/graphs/index.rst).

Parameters:

graphMultiDiGraph

Graph to save.

pathUnion[Path, str]

The path to the file where the graph will be saved.

Returns:

pathPath

Path to the file where the graph will be saved.

Examples

>>> from flpq_data import *
>>> import pathlib, tempfile
>>> d = pathlib.Path(tempfile.mkdtemp())
>>> g = graph_from_text(["0 a 1", "1 b 2"])
>>> path = graph_to_rdf(g, d / "g.ttl")

iter_edges_from_rdf(path: Path | str) → Iterator[Tuple[str, str, str]]

Yields the edges of an RDF file as a stream.

Reads both the current encoding (IRI nodes urn:flpq:node:<id> and IRI predicates urn:flpq:label:<label>) and the legacy form written by older versions (blank-node endpoints and Literal predicates). rdflib has no streaming parser, so the file is materialized in memory.

Parameters:

pathUnion[Path, str]

The path to the RDF file with which the edges will be created.

Returns:

edgesIterator[Tuple[str, str, str]]

The (u, label, v) edge tuples (the order is not guaranteed — rdflib iterates in store order).

Examples

>>> import pathlib, tempfile
>>> d = pathlib.Path(tempfile.mkdtemp())
>>> p = d / "g.ttl"
>>> _ = p.write_text(
...     "<urn:flpq:node:0> <urn:flpq:label:a> <urn:flpq:node:1> .\n"
...     "<urn:flpq:node:1> <urn:flpq:label:b_5> <urn:flpq:node:2> .\n"
... )
>>> sorted(iter_edges_from_rdf(p))
[('0', 'a', '1'), ('1', 'b_5', '2')]

rdf_from_edges(edges: Iterable[Tuple[Any, str, Any]], path: Path | str) → Path

Writes an edge stream to an RDF file.

Emits valid RDF 1.1 Turtle — one triple per line with IRI nodes urn:flpq:node:<id> and IRI predicates urn:flpq:label:<label>, the variable components percent-encoded (the “RDF encoding” section of docs/graphs/index.rst). The file is written line by line without an in-memory RDF store, so the memory stays O(1) in the number of edges.

Parameters:

edgesIterable[Tuple[Any, str, Any]]

The (u, label, v) edge tuples to write.

pathUnion[Path, str]

The path to the RDF file where the edges will be saved.

Returns:

pathPath

Path to the RDF file where the edges will be saved.

Examples

>>> import pathlib, tempfile
>>> p = pathlib.Path(tempfile.mkdtemp()) / "g.ttl"
>>> _ = rdf_from_edges([(0, "a b", 1)], p)
>>> p.read_text().strip()
'<urn:flpq:node:0> <urn:flpq:label:a%20b> <urn:flpq:node:1> .'