from __future__ import annotations
from typing import TYPE_CHECKING, Iterable, Iterator, List, Optional
from rdflib.namespace import RDF
from rdflib.term import BNode, Node
if TYPE_CHECKING:
from rdflib.graph import Graph
__all__ = ["Collection"]
[docs]class Collection:
__doc__ = """
See "Emulating container types":
https://docs.python.org/reference/datamodel.html#emulating-container-types
>>> from rdflib.term import Literal
>>> from rdflib.graph import Graph
>>> from pprint import pprint
>>> listname = BNode()
>>> g = Graph('Memory')
>>> listItem1 = BNode()
>>> listItem2 = BNode()
>>> g.add((listname, RDF.first, Literal(1))) # doctest: +ELLIPSIS
<Graph identifier=... (<class 'rdflib.graph.Graph'>)>
>>> g.add((listname, RDF.rest, listItem1)) # doctest: +ELLIPSIS
<Graph identifier=... (<class 'rdflib.graph.Graph'>)>
>>> g.add((listItem1, RDF.first, Literal(2))) # doctest: +ELLIPSIS
<Graph identifier=... (<class 'rdflib.graph.Graph'>)>
>>> g.add((listItem1, RDF.rest, listItem2)) # doctest: +ELLIPSIS
<Graph identifier=... (<class 'rdflib.graph.Graph'>)>
>>> g.add((listItem2, RDF.rest, RDF.nil)) # doctest: +ELLIPSIS
<Graph identifier=... (<class 'rdflib.graph.Graph'>)>
>>> g.add((listItem2, RDF.first, Literal(3))) # doctest: +ELLIPSIS
<Graph identifier=... (<class 'rdflib.graph.Graph'>)>
>>> c = Collection(g,listname)
>>> pprint([term.n3() for term in c])
[u'"1"^^<http://www.w3.org/2001/XMLSchema#integer>',
u'"2"^^<http://www.w3.org/2001/XMLSchema#integer>',
u'"3"^^<http://www.w3.org/2001/XMLSchema#integer>']
>>> Literal(1) in c
True
>>> len(c)
3
>>> c._get_container(1) == listItem1
True
>>> c.index(Literal(2)) == 1
True
"""
[docs] def __init__(self, graph: Graph, uri: Node, seq: List[Node] = []):
self.graph = graph
self.uri = uri or BNode()
self += seq
[docs] def n3(self) -> str:
"""
>>> from rdflib.term import Literal
>>> from rdflib.graph import Graph
>>> listname = BNode()
>>> g = Graph('Memory')
>>> listItem1 = BNode()
>>> listItem2 = BNode()
>>> g.add((listname, RDF.first, Literal(1))) # doctest: +ELLIPSIS
<Graph identifier=... (<class 'rdflib.graph.Graph'>)>
>>> g.add((listname, RDF.rest, listItem1)) # doctest: +ELLIPSIS
<Graph identifier=... (<class 'rdflib.graph.Graph'>)>
>>> g.add((listItem1, RDF.first, Literal(2))) # doctest: +ELLIPSIS
<Graph identifier=... (<class 'rdflib.graph.Graph'>)>
>>> g.add((listItem1, RDF.rest, listItem2)) # doctest: +ELLIPSIS
<Graph identifier=... (<class 'rdflib.graph.Graph'>)>
>>> g.add((listItem2, RDF.rest, RDF.nil)) # doctest: +ELLIPSIS
<Graph identifier=... (<class 'rdflib.graph.Graph'>)>
>>> g.add((listItem2, RDF.first, Literal(3))) # doctest: +ELLIPSIS
<Graph identifier=... (<class 'rdflib.graph.Graph'>)>
>>> c = Collection(g, listname)
>>> print(c.n3()) #doctest: +NORMALIZE_WHITESPACE
( "1"^^<http://www.w3.org/2001/XMLSchema#integer>
"2"^^<http://www.w3.org/2001/XMLSchema#integer>
"3"^^<http://www.w3.org/2001/XMLSchema#integer> )
"""
# type error: "Node" has no attribute "n3"
return "( %s )" % (" ".join([i.n3() for i in self])) # type: ignore[attr-defined]
def _get_container(self, index: int) -> Optional[Node]:
"""Gets the first, rest holding node at index."""
assert isinstance(index, int)
graph = self.graph
container: Optional[Node] = self.uri
i = 0
while i < index:
i += 1
container = graph.value(container, RDF.rest)
if container is None:
break
return container
[docs] def __len__(self) -> int:
"""length of items in collection."""
return len(list(self.graph.items(self.uri)))
[docs] def index(self, item: Node) -> int:
"""
Returns the 0-based numerical index of the item in the list
"""
listname = self.uri
index = 0
while True:
if (listname, RDF.first, item) in self.graph:
return index
else:
newlink = list(self.graph.objects(listname, RDF.rest))
index += 1
if newlink == [RDF.nil]:
raise ValueError("%s is not in %s" % (item, self.uri))
elif not newlink:
raise Exception("Malformed RDF Collection: %s" % self.uri)
else:
assert len(newlink) == 1, "Malformed RDF Collection: %s" % self.uri
listname = newlink[0]
[docs] def __getitem__(self, key: int) -> Node:
"""TODO"""
c = self._get_container(key)
if c:
v = self.graph.value(c, RDF.first)
if v:
return v
else:
raise KeyError(key)
else:
raise IndexError(key)
[docs] def __setitem__(self, key: int, value: Node) -> None:
"""TODO"""
c = self._get_container(key)
if c:
self.graph.set((c, RDF.first, value))
else:
raise IndexError(key)
[docs] def __delitem__(self, key: int) -> None:
"""
>>> from rdflib.namespace import RDF, RDFS
>>> from rdflib import Graph
>>> from pprint import pformat
>>> g = Graph()
>>> a = BNode('foo')
>>> b = BNode('bar')
>>> c = BNode('baz')
>>> g.add((a, RDF.first, RDF.type)) # doctest: +ELLIPSIS
<Graph identifier=... (<class 'rdflib.graph.Graph'>)>
>>> g.add((a, RDF.rest, b)) # doctest: +ELLIPSIS
<Graph identifier=... (<class 'rdflib.graph.Graph'>)>
>>> g.add((b, RDF.first, RDFS.label)) # doctest: +ELLIPSIS
<Graph identifier=... (<class 'rdflib.graph.Graph'>)>
>>> g.add((b, RDF.rest, c)) # doctest: +ELLIPSIS
<Graph identifier=... (<class 'rdflib.graph.Graph'>)>
>>> g.add((c, RDF.first, RDFS.comment)) # doctest: +ELLIPSIS
<Graph identifier=... (<class 'rdflib.graph.Graph'>)>
>>> g.add((c, RDF.rest, RDF.nil)) # doctest: +ELLIPSIS
<Graph identifier=... (<class 'rdflib.graph.Graph'>)>
>>> len(g)
6
>>> def listAncestry(node, graph):
... for i in graph.subjects(RDF.rest, node):
... yield i
>>> [str(node.n3())
... for node in g.transitiveClosure(listAncestry, RDF.nil)]
['_:baz', '_:bar', '_:foo']
>>> lst = Collection(g, a)
>>> len(lst)
3
>>> b == lst._get_container(1)
True
>>> c == lst._get_container(2)
True
>>> del lst[1]
>>> len(lst)
2
>>> len(g)
4
"""
self[key] # to raise any potential key exceptions
graph = self.graph
current = self._get_container(key)
assert current
if len(self) == 1 and key > 0:
pass
elif key == len(self) - 1:
# the tail
priorlink = self._get_container(key - 1)
# type error: Argument 1 to "set" of "Graph" has incompatible type "Tuple[Optional[Node], URIRef, URIRef]"; expected "Tuple[Node, Node, Any]"
self.graph.set((priorlink, RDF.rest, RDF.nil)) # type: ignore[arg-type]
graph.remove((current, None, None))
else:
next = self._get_container(key + 1)
prior = self._get_container(key - 1)
assert next and prior
graph.remove((current, None, None))
graph.set((prior, RDF.rest, next))
[docs] def __iter__(self) -> Iterator[Node]:
"""Iterator over items in Collections"""
return self.graph.items(self.uri)
def _end(self) -> Node:
# find end of list
container = self.uri
while True:
rest = self.graph.value(container, RDF.rest)
if rest is None or rest == RDF.nil:
return container
else:
container = rest
[docs] def append(self, item: Node) -> Collection:
"""
>>> from rdflib.term import Literal
>>> from rdflib.graph import Graph
>>> listname = BNode()
>>> g = Graph()
>>> c = Collection(g,listname,[Literal(1),Literal(2)])
>>> links = [
... list(g.subjects(object=i, predicate=RDF.first))[0] for i in c]
>>> len([i for i in links if (i, RDF.rest, RDF.nil) in g])
1
"""
end = self._end()
if (end, RDF.first, None) in self.graph:
# append new node to the end of the linked list
node = BNode()
self.graph.set((end, RDF.rest, node))
end = node
self.graph.add((end, RDF.first, item))
self.graph.add((end, RDF.rest, RDF.nil))
return self
[docs] def __iadd__(self, other: Iterable[Node]):
end = self._end()
self.graph.remove((end, RDF.rest, None))
for item in other:
if (end, RDF.first, None) in self.graph:
nxt = BNode()
self.graph.add((end, RDF.rest, nxt))
end = nxt
self.graph.add((end, RDF.first, item))
self.graph.add((end, RDF.rest, RDF.nil))
return self
[docs] def clear(self):
container: Optional[Node] = self.uri
graph = self.graph
while container:
rest = graph.value(container, RDF.rest)
graph.remove((container, RDF.first, None))
graph.remove((container, RDF.rest, None))
container = rest
return self
