Source code for rdflib.collection

from rdflib.namespace import RDF
from rdflib.term import BNode
from rdflib.term import Literal


__all__ = ["Collection"]


[docs]class Collection(object): __doc__ = """ See "Emulating container types": https://docs.python.org/reference/datamodel.html#emulating-container-types >>> 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, uri, seq=[]): self.graph = graph self.uri = uri or BNode() self += seq
[docs] def n3(self): """ >>> 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> ) """ return "( %s )" % (" ".join([i.n3() for i in self]))
def _get_container(self, index): """Gets the first, rest holding node at index.""" assert isinstance(index, int) graph = self.graph container = 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): """length of items in collection.""" return len(list(self.graph.items(self.uri)))
[docs] def index(self, item): """ 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): """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, value): """TODO""" c = self._get_container(key) if c: self.graph.set((c, RDF.first, value)) else: raise IndexError(key)
[docs] def __delitem__(self, key): """ >>> 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) self.graph.set((priorLink, RDF.rest, RDF.nil)) 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 over items in Collections""" return self.graph.items(self.uri)
def _end(self): # 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): """ >>> 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): 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 = 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
def test(): import doctest doctest.testmod() if __name__ == "__main__": test() from rdflib import Graph g = Graph() c = Collection(g, BNode()) assert len(c) == 0 c = Collection(g, BNode(), [Literal("1"), Literal("2"), Literal("3"), Literal("4")]) assert len(c) == 4 assert c[1] == Literal("2"), c[1] del c[1] assert list(c) == [Literal("1"), Literal("3"), Literal("4")], list(c) try: del c[500] except IndexError: pass c.append(Literal("5")) print(list(c)) for i in c: print(i) del c[3] c.clear() assert len(c) == 0