from random import randint
from rdflib.namespace import RDF
from rdflib.term import BNode, URIRef
__all__ = ["Container", "Bag", "Seq", "Alt", "NoElementException"]
[docs]class Container(object):
"""A class for constructing RDF containers, as per https://www.w3.org/TR/rdf11-mt/#rdf-containers
Basic usage, creating a ``Bag`` and adding to it::
>>> from rdflib import Graph, BNode, Literal, Bag
>>> g = Graph()
>>> b = Bag(g, BNode(), [Literal("One"), Literal("Two"), Literal("Three")])
>>> print(g.serialize(format="turtle"))
@prefix rdf: <http://www.w3.org/1999/02/22-rdf-syntax-ns#> .
<BLANKLINE>
[] a rdf:Bag ;
rdf:_1 "One" ;
rdf:_2 "Two" ;
rdf:_3 "Three" .
<BLANKLINE>
<BLANKLINE>
>>> # print out an item using an index reference
>>> print(b[2])
Two
>>> # add a new item
>>> b.append(Literal("Hello")) # doctest: +ELLIPSIS
<rdflib.container.Bag object at ...>
>>> print(g.serialize(format="turtle"))
@prefix rdf: <http://www.w3.org/1999/02/22-rdf-syntax-ns#> .
<BLANKLINE>
[] a rdf:Bag ;
rdf:_1 "One" ;
rdf:_2 "Two" ;
rdf:_3 "Three" ;
rdf:_4 "Hello" .
<BLANKLINE>
<BLANKLINE>
"""
[docs] def __init__(self, graph, uri, seq=[], rtype="Bag"):
"""Creates a Container
:param graph: a Graph instance
:param uri: URI or Blank Node of the Container
:param seq: the elements of the Container
:param rtype: the type of Container, one of "Bag", "Seq" or "Alt"
"""
self.graph = graph
self.uri = uri or BNode()
self._len = 0
self._rtype = rtype # rdf:Bag or rdf:Seq or rdf:Alt
self.append_multiple(seq)
# adding triple corresponding to container type
self.graph.add((self.uri, RDF.type, RDF[self._rtype]))
[docs] def n3(self):
items = []
for i in range(len(self)):
v = self[i + 1]
items.append(v)
return "( %s )" % " ".join([a.n3() for a in items])
def _get_container(self):
"""Returns the URI of the container"""
return self.uri
[docs] def __len__(self):
"""Number of items in container"""
return self._len
[docs] def type_of_conatiner(self):
return self._rtype
[docs] def index(self, item):
"""Returns the 1-based numerical index of the item in the container"""
pred = self.graph.predicates(self.uri, item)
if not pred:
raise ValueError("%s is not in %s" % (item, "container"))
LI_INDEX = URIRef(str(RDF) + "_")
i = None
for p in pred:
i = int(p.replace(LI_INDEX, ""))
return i
[docs] def __getitem__(self, key):
"""Returns item of the container at index key"""
c = self._get_container()
assert isinstance(key, int)
elem_uri = str(RDF) + "_" + str(key)
if key <= 0 or key > len(self):
raise KeyError(key)
v = self.graph.value(c, URIRef(elem_uri))
if v:
return v
else:
raise KeyError(key)
[docs] def __setitem__(self, key, value):
"""Sets the item at index key or predicate rdf:_key of the container to value"""
assert isinstance(key, int)
c = self._get_container()
elem_uri = str(RDF) + "_" + str(key)
if key <= 0 or key > len(self):
raise KeyError(key)
self.graph.set((c, URIRef(elem_uri), value))
[docs] def __delitem__(self, key):
"""Removing the item with index key or predicate rdf:_key"""
assert isinstance(key, int)
if key <= 0 or key > len(self):
raise KeyError(key)
graph = self.graph
container = self.uri
elem_uri = str(RDF) + "_" + str(key)
graph.remove((container, URIRef(elem_uri), None))
for j in range(key + 1, len(self) + 1):
elem_uri = str(RDF) + "_" + str(j)
v = graph.value(container, URIRef(elem_uri))
graph.remove((container, URIRef(elem_uri), v))
elem_uri = str(RDF) + "_" + str(j - 1)
graph.add((container, URIRef(elem_uri), v))
self._len -= 1
[docs] def items(self):
"""Returns a list of all items in the container"""
l_ = []
container = self.uri
i = 1
while True:
elem_uri = str(RDF) + "_" + str(i)
if (container, URIRef(elem_uri), None) in self.graph:
i += 1
l_.append(self.graph.value(container, URIRef(elem_uri)))
else:
break
return l_
[docs] def end(self): #
# find end index (1-based) of container
container = self.uri
i = 1
while True:
elem_uri = str(RDF) + "_" + str(i)
if (container, URIRef(elem_uri), None) in self.graph:
i += 1
else:
return i - 1
[docs] def append(self, item):
"""Adding item to the end of the container"""
end = self.end()
elem_uri = str(RDF) + "_" + str(end + 1)
container = self.uri
self.graph.add((container, URIRef(elem_uri), item))
self._len += 1
return self
[docs] def append_multiple(self, other):
"""Adding multiple elements to the container to the end which are in python list other"""
end = self.end() # it should return the last index
container = self.uri
for item in other:
end += 1
self._len += 1
elem_uri = str(RDF) + "_" + str(end)
self.graph.add((container, URIRef(elem_uri), item))
return self
[docs] def clear(self):
"""Removing all elements from the container"""
container = self.uri
graph = self.graph
i = 1
while True:
elem_uri = str(RDF) + "_" + str(i)
if (container, URIRef(elem_uri), None) in self.graph:
graph.remove((container, URIRef(elem_uri), None))
i += 1
else:
break
self._len = 0
return self
[docs]class Bag(Container):
"""Unordered container (no preference order of elements)"""
[docs] def __init__(self, graph, uri, seq=[]):
Container.__init__(self, graph, uri, seq, "Bag")
[docs]class Alt(Container):
[docs] def __init__(self, graph, uri, seq=[]):
Container.__init__(self, graph, uri, seq, "Alt")
[docs] def anyone(self):
if len(self) == 0:
raise NoElementException()
else:
p = randint(1, len(self))
item = self.__getitem__(p)
return item
[docs]class Seq(Container):
[docs] def __init__(self, graph, uri, seq=[]):
Container.__init__(self, graph, uri, seq, "Seq")
[docs] def add_at_position(self, pos, item):
assert isinstance(pos, int)
if pos <= 0 or pos > len(self) + 1:
raise ValueError("Invalid Position for inserting element in rdf:Seq")
if pos == len(self) + 1:
self.append(item)
else:
for j in range(len(self), pos - 1, -1):
container = self._get_container()
elem_uri = str(RDF) + "_" + str(j)
v = self.graph.value(container, URIRef(elem_uri))
self.graph.remove((container, URIRef(elem_uri), v))
elem_uri = str(RDF) + "_" + str(j + 1)
self.graph.add((container, URIRef(elem_uri), v))
elem_uri_pos = str(RDF) + "_" + str(pos)
self.graph.add((container, URIRef(elem_uri_pos), item))
self._len += 1
return self
[docs]class NoElementException(Exception):
[docs] def __init__(self, message="rdf:Alt Container is empty"):
self.message = message
[docs] def __str__(self):
return self.message