Viewing file: Redland.py (3.5 KB) -rw-r--r-- Select action/file-type: (+) | (+) | (+) | Code (+) | Session (+) | (+) | SDB (+) | (+) | (+) | (+) | (+) | (+) |
import rdflib from rdflib.Graph import Graph from rdflib.URIRef import URIRef from rdflib.Node import Node from rdflib.BNode import BNode from rdflib.Literal import Literal try: import RDF except ImportError: import warnings warnings.warn("Redlands not installed") __test__=False
from rdflib.store import Store
def _t(i): if isinstance(i, rdflib.URIRef): return RDF.Node(RDF.Uri(unicode(i))) if isinstance(i, rdflib.BNode): return RDF.Node(blank=str(i)) if isinstance(i, rdflib.Literal): return RDF.Node(literal=str(i)) if isinstance(i, Graph): return _t(i.identifier) if i is None: return None raise TypeError, 'Cannot convert %s' % `i`
def _c(i): return _t(i)
def _f(i): if isinstance(i, RDF.Uri): return rdflib.URIRef(i) if isinstance(i, RDF.Node): if i.is_blank(): return rdflib.BNode(i.blank_identifier) elif i.is_literal(): return rdflib.Literal(i) else: return URIRef(i.uri) if i is None: return None raise TypeError, 'Cannot convert %s' % `i`
class Redland(Store): context_aware = True def __init__(self, model=None): super(Redland, self).__init__() if model is None: model = RDF.Model(RDF.MemoryStorage(options_string="contexts='yes'")) self.model = model
def __len__(self, context=None): """ Return number of triples (statements in librdf). """
count = 0 for triple, cg in self.triples((None, None, None), context): count += 1 return count
def add(self, (subject, predicate, object), context=None, quoted=False): """\ Add a triple to the store of triples. """ if context is not None: self.model.append(RDF.Statement(_t(subject), _t(predicate), _t(object)), _c(context)) else: self.model.append(RDF.Statement(_t(subject), _t(predicate), _t(object)))
def remove(self, (subject, predicate, object), context, quoted=False): if context is None: contexts = self.contexts() else: contexts = [context] for context in contexts: if subject is None and predicate is None and object is None: self.model.remove_statements_with_context(_c(context)) else: del self.model[RDF.Statement(_t(subject), _t(predicate), _t(object)), _c(context)]
def triples(self, (subject, predicate, object), context=None): """A generator over all the triples matching """ cgraph = RDF.Model() triple = RDF.Statement(_t(subject), _t(predicate), _t(object)) for statement, c in self.model.find_statements_context(triple): if context is None or _f(c) == context.identifier: cgraph.append(statement) for statement in cgraph.find_statements(triple): ret = [] for c in self.model.get_contexts(): if self.model.contains_statement_context(statement, _c(context)): ret.append(c) yield (_f(statement.subject), _f(statement.predicate), _f(statement.object)), iter(ret)
def contexts(self, triple=None): # TODO: have Graph support triple? for context in self.model.get_contexts(): yield Graph(self, _f(context))
def bind(self, prefix, namespace): pass
def namespace(self, prefix): pass
def prefix(self, namespace): pass
def namespaces(self): pass
|