Source code for rdflib.plugins.serializers.trix
from typing import IO, Optional
from rdflib.graph import ConjunctiveGraph, Graph
from rdflib.namespace import Namespace
from rdflib.plugins.serializers.xmlwriter import XMLWriter
from rdflib.serializer import Serializer
from rdflib.term import BNode, Literal, URIRef
__all__ = ["TriXSerializer"]
# TODO: Move this somewhere central
TRIXNS = Namespace("http://www.w3.org/2004/03/trix/trix-1/")
XMLNS = Namespace("http://www.w3.org/XML/1998/namespace")
[docs]class TriXSerializer(Serializer):
[docs] def __init__(self, store: Graph):
super(TriXSerializer, self).__init__(store)
if not store.context_aware:
raise Exception(
"TriX serialization only makes sense for context-aware stores"
)
[docs] def serialize(
self,
stream: IO[bytes],
base: Optional[str] = None,
encoding: Optional[str] = None,
**args,
):
nm = self.store.namespace_manager
self.writer = XMLWriter(stream, nm, encoding, extra_ns={"": TRIXNS})
self.writer.push(TRIXNS["TriX"])
# if base is given here, use that, if not and a base is set for the graph use that
if base is None and self.store.base is not None:
base = self.store.base
if base is not None:
self.writer.attribute("http://www.w3.org/XML/1998/namespacebase", base)
self.writer.namespaces()
if isinstance(self.store, ConjunctiveGraph):
for subgraph in self.store.contexts():
self._writeGraph(subgraph)
elif isinstance(self.store, Graph):
self._writeGraph(self.store)
else:
raise Exception(f"Unknown graph type: {type(self.store)}")
self.writer.pop()
stream.write("\n".encode("latin-1"))
def _writeGraph(self, graph):
self.writer.push(TRIXNS["graph"])
if graph.base:
self.writer.attribute(
"http://www.w3.org/XML/1998/namespacebase", graph.base
)
if isinstance(graph.identifier, URIRef):
self.writer.element(TRIXNS["uri"], content=str(graph.identifier))
for triple in graph.triples((None, None, None)):
self._writeTriple(triple)
self.writer.pop()
def _writeTriple(self, triple):
self.writer.push(TRIXNS["triple"])
for component in triple:
if isinstance(component, URIRef):
self.writer.element(TRIXNS["uri"], content=str(component))
elif isinstance(component, BNode):
self.writer.element(TRIXNS["id"], content=str(component))
elif isinstance(component, Literal):
if component.datatype:
self.writer.element(
TRIXNS["typedLiteral"],
content=str(component),
attributes={TRIXNS["datatype"]: str(component.datatype)},
)
elif component.language:
self.writer.element(
TRIXNS["plainLiteral"],
content=str(component),
attributes={XMLNS["lang"]: str(component.language)},
)
else:
self.writer.element(TRIXNS["plainLiteral"], content=str(component))
self.writer.pop()