Mercurial > code > home > repos > light9
changeset 1691:a18eb09762b7
move rdfdb to a new project
Ignore-this: 8ce04a43895d1516cd7c14421840eeaf
| author | Drew Perttula <drewp@bigasterisk.com> |
|---|---|
| date | Sun, 18 Feb 2018 08:05:04 +0000 |
| parents | fa2cc1f0e509 |
| children | 6fa4288da8a6 |
| files | light9/rdfdb/__init__.py light9/rdfdb/autodepgraphapi.py light9/rdfdb/clientsession.py light9/rdfdb/currentstategraphapi.py light9/rdfdb/file_vs_uri.py light9/rdfdb/grapheditapi.py light9/rdfdb/graphfile.py light9/rdfdb/graphfile_test.py light9/rdfdb/localsyncedgraph.py light9/rdfdb/mock_syncedgraph.py light9/rdfdb/patch.py light9/rdfdb/patchreceiver.py light9/rdfdb/patchsender.py light9/rdfdb/rdflibpatch.py light9/rdfdb/rdflibpatch_literal.py light9/rdfdb/syncedgraph.py light9/rdfdb/web/graphView.html light9/rdfdb/web/index.html light9/rdfdb/web/style.css light9/rdfdb/web/syncedgraph.js |
| diffstat | 19 files changed, 0 insertions(+), 1850 deletions(-) [+] |
line wrap: on
line diff
--- a/light9/rdfdb/autodepgraphapi.py Fri Jun 16 04:28:26 2017 +0000 +++ /dev/null Thu Jan 01 00:00:00 1970 +0000 @@ -1,221 +0,0 @@ -import logging -from rdflib import RDF, RDFS -from light9.rdfdb.currentstategraphapi import contextsForStatementNoWildcards -log = logging.getLogger('autodepgraphapi') - -class AutoDepGraphApi(object): - """ - knockoutjs-inspired API for automatically building a dependency - tree while reading the graph. See addHandler(). - - This design is more aggressive than knockoutjs, since the obvious - query methods like value() all error if you're not in a watched - section of code. This is supposed to make it easier to notice - dependency mistakes, especially when porting old code to use - SyncedGraph. - - This class is a mixin for SyncedGraph, separated here because - these methods work together - """ - - def __init__(self): - self._watchers = _GraphWatchers() - self.currentFuncs = [] # stack of addHandler callers - - def addHandler(self, func): - """ - run this (idempotent) func, noting what graph values it - uses. Run it again in the future if there are changes to those - graph values. The func might use different values during that - future call, and those will be what we watch for next. - """ - - # if we saw this func before, we need to forget the old - # callbacks it wanted and replace with the new ones we see - # now. - - # if one handler func calls another, does that break anything? - # maybe not? - - # no plan for sparql queries yet. Hook into a lower layer that - # reveals all their statement fetches? Just make them always - # new? Cache their results, so if i make the query again and - # it gives the same result, I don't call the handler? - - self.currentFuncs.append(func) - log.debug('graph.currentFuncs push %s', func) - try: - try: - func() - except: - import traceback - traceback.print_exc() - raise - finally: - self.currentFuncs.pop() - log.debug('graph.currentFuncs pop %s. stack now has %s', func, len(self.currentFuncs)) - - def runDepsOnNewPatch(self, p): - """ - patch p just happened to the graph; call everyone back who - might care, and then notice what data they depend on now - """ - for func in self._watchers.whoCares(p): - # todo: forget the old handlers for this func - log.debug('runDepsOnNewPatch calling watcher %s', p) - self.addHandler(func) - - def _getCurrentFunc(self): - if not self.currentFuncs: - # this may become a warning later - raise ValueError("asked for graph data outside of a handler") - - # we add the watcher to the deepest function, since that - # should be the cheapest way to update when this part of the - # data changes - return self.currentFuncs[-1] - - # these just call through to triples() so it might be possible to - # watch just that one. - - # if you get a bnode in your response, maybe the answer to - # dependency tracking is to say that you depended on the triple - # that got you that bnode, since it is likely to change to another - # bnode later. This won't work if the receiver stores bnodes - # between calls, but probably most of them don't do that (they - # work from a starting uri) - - def value(self, subject=None, predicate=RDF.value, object=None, - default=None, any=True): - if object is not None: - raise NotImplementedError() - func = self._getCurrentFunc() - self._watchers.addSubjPredWatcher(func, subject, predicate) - return self._graph.value(subject, predicate, object=object, - default=default, any=any) - - def objects(self, subject=None, predicate=None): - func = self._getCurrentFunc() - self._watchers.addSubjPredWatcher(func, subject, predicate) - return self._graph.objects(subject, predicate) - - def label(self, uri): - return self.value(uri, RDFS.label) - - def subjects(self, predicate=None, object=None): - func = self._getCurrentFunc() - self._watchers.addPredObjWatcher(func, predicate, object) - return self._graph.subjects(predicate, object) - - def predicate_objects(self, subject): - func = self._getCurrentFunc() - self._watchers.addSubjectWatcher(func, subject) - return self._graph.predicate_objects(subject) - - def items(self, listUri): - """generator. Having a chain of watchers on the results is not - well-tested yet""" - chain = set([listUri]) - while listUri: - item = self.value(listUri, RDF.first) - if item: - yield item - listUri = self.value(listUri, RDF.rest) - if listUri in chain: - raise ValueError("List contains a recursive rdf:rest reference") - chain.add(listUri) - - - def contains(self, triple): - func = self._getCurrentFunc() - self._watchers.addTripleWatcher(func, triple) - return triple in self._graph - - def contextsForStatement(self, triple): - """currently this needs to be in an addHandler section, but it - sets no watchers so it won't actually update if the statement - was added or dropped from contexts""" - func = self._getCurrentFunc() - return contextsForStatementNoWildcards(self._graph, triple) - - # i find myself wanting 'patch' (aka enter/leave) versions of these calls that tell - # you only what results have just appeared or disappeared. I think - # I'm going to be repeating that logic a lot. Maybe just for the - # subjects(RDF.type, t) call - - -class _GraphWatchers(object): - """ - store the current handlers that care about graph changes - """ - def __init__(self): - self._handlersSp = {} # (s,p): set(handlers) - self._handlersPo = {} # (p,o): set(handlers) - self._handlersSpo = {} # (s,p,o): set(handlers) - self._handlersS = {} # s: set(handlers) - - def addSubjPredWatcher(self, func, s, p): - if func is None: - return - key = s, p - try: - self._handlersSp.setdefault(key, set()).add(func) - except Exception: - log.error("with key %r and func %r" % (key, func)) - raise - - def addPredObjWatcher(self, func, p, o): - self._handlersPo.setdefault((p, o), set()).add(func) - - def addTripleWatcher(self, func, triple): - self._handlersSpo.setdefault(triple, set()).add(func) - - def addSubjectWatcher(self, func, s): - self._handlersS.setdefault(s, set()).add(func) - - def whoCares(self, patch): - """what handler functions would care about the changes in this patch? - - this removes the handlers that it gives you - """ - #self.dependencies() - ret = set() - affectedSubjPreds = set([(s, p) for s, p, o, c in patch.addQuads]+ - [(s, p) for s, p, o, c in patch.delQuads]) - for (s, p), funcs in self._handlersSp.iteritems(): - if (s, p) in affectedSubjPreds: - ret.update(funcs) - funcs.clear() - - affectedPredObjs = set([(p, o) for s, p, o, c in patch.addQuads]+ - [(p, o) for s, p, o, c in patch.delQuads]) - for (p, o), funcs in self._handlersPo.iteritems(): - if (p, o) in affectedPredObjs: - ret.update(funcs) - funcs.clear() - - affectedTriples = set([(s, p, o) for s, p, o, c in patch.addQuads]+ - [(s, p, o) for s, p, o, c in patch.delQuads]) - for triple, funcs in self._handlersSpo.iteritems(): - if triple in affectedTriples: - ret.update(funcs) - funcs.clear() - - affectedSubjs = set([s for s, p, o, c in patch.addQuads]+ - [s for s, p, o, c in patch.delQuads]) - for subj, funcs in self._handlersS.iteritems(): - if subj in affectedSubjs: - ret.update(funcs) - funcs.clear() - - return ret - - def dependencies(self): - """ - for debugging, make a list of all the active handlers and what - data they depend on. This is meant for showing on the web ui - for browsing. - """ - log.info("whocares:") - from pprint import pprint - pprint(self._handlersSp)
--- a/light9/rdfdb/clientsession.py Fri Jun 16 04:28:26 2017 +0000 +++ /dev/null Thu Jan 01 00:00:00 1970 +0000 @@ -1,17 +0,0 @@ -""" -some clients will support the concept of a named session that keeps -multiple instances of that client separate -""" -from rdflib import URIRef -from urllib import quote -from light9 import showconfig - -def add_option(parser): - parser.add_option( - '-s', '--session', - help="name of session used for levels and window position", - default='default') - -def getUri(appName, opts): - return URIRef("%s/sessions/%s/%s" % (showconfig.showUri(), appName, - quote(opts.session, safe='')))
--- a/light9/rdfdb/currentstategraphapi.py Fri Jun 16 04:28:26 2017 +0000 +++ /dev/null Thu Jan 01 00:00:00 1970 +0000 @@ -1,87 +0,0 @@ -import logging, traceback, time, itertools -from rdflib import ConjunctiveGraph -from light9.rdfdb.rdflibpatch import contextsForStatement as rp_contextsForStatement -log = logging.getLogger("currentstate") - -class ReadOnlyConjunctiveGraph(object): - """similar to rdflib's ReadOnlyGraphAggregate but takes one CJ in, instead - of a bunch of Graphs""" - def __init__(self, graph): - self.graph = graph - - def __getattr__(self, attr): - if attr in ['subjects', 'value', 'objects', 'triples', 'label']: # not complete - return getattr(self.graph, attr) - raise TypeError("can't access %r of read-only graph" % attr) - - def __len__(self): - return len(self.graph) - - -class CurrentStateGraphApi(object): - """ - mixin for SyncedGraph, separated here because these methods work together - """ - - def currentState(self, context=None, tripleFilter=(None, None, None)): - """ - a graph you can read without being in an addHandler - - you can save some time by passing a triple filter, and we'll only give you the matching triples - """ - if context is not None: - raise NotImplementedError("currentState with context arg") - - class Mgr(object): - def __enter__(self2): - # this should be a readonly view of the existing - # graph, maybe with something to guard against - # writes/patches happening while reads are being - # done. Typical usage will do some reads on this graph - # before moving on to writes. - - if 1: - g = ReadOnlyConjunctiveGraph(self._graph) - else: - t1 = time.time() - g = ConjunctiveGraph() - for s,p,o,c in self._graph.quads(tripleFilter): - g.store.add((s,p,o), c) - - if tripleFilter == (None, None, None): - self2.logThisCopy(g, time.time() - t1) - - g.contextsForStatement = lambda t: contextsForStatementNoWildcards(g, t) - return g - - def logThisCopy(self, g, sec): - log.info("copied graph %s statements (%.1f ms) " - "because of this:" % (len(g), sec * 1000)) - for frame in traceback.format_stack(limit=4)[:-2]: - for line in frame.splitlines(): - log.info(" "+line) - - def __exit__(self, type, val, tb): - return - - return Mgr() - - def sequentialUri(self, prefix): - """ - Prefix URIRef like http://example.com/r- will return - http://example.com/r-1 if that uri is not a subject in the graph, - or else http://example.com/r-2, etc - """ - for i in itertools.count(1): - newUri = prefix + str(i) - if not list(self._graph.triples((newUri, None, None))) and newUri not in getattr(self, '_reservedSequentials', []): - if not hasattr(self, '_reservedSequentials'): - self._reservedSequentials = set() - self._reservedSequentials.add(newUri) - return newUri - - -def contextsForStatementNoWildcards(g, triple): - if None in triple: - raise NotImplementedError("no wildcards") - return rp_contextsForStatement(g, triple)
--- a/light9/rdfdb/file_vs_uri.py Fri Jun 16 04:28:26 2017 +0000 +++ /dev/null Thu Jan 01 00:00:00 1970 +0000 @@ -1,33 +0,0 @@ -""" -note that if there are ambiguities in dirUriMap, you'll get -undefined results that depend on python's items() order. -""" -import os -from rdflib import URIRef - -def uriFromFile(dirUriMap, filename): - assert filename.endswith('.n3'), filename - for d, prefix in dirUriMap.items(): - if filename.startswith(d): - return URIRef(prefix + filename[len(d):-len('.n3')]) - raise ValueError("filename %s doesn't start with any of %s" % - (filename, dirUriMap.keys())) - -def fileForUri(dirUriMap, ctx): - assert isinstance(ctx, URIRef), ctx - for d, prefix in dirUriMap.items(): - if ctx.startswith(prefix): - return d + ctx[len(prefix):] + '.n3' - raise ValueError("don't know what filename to use for %s" % ctx) - -def correctToTopdirPrefix(dirUriMap, inFile): - if not any(inFile.startswith(prefix) for prefix in dirUriMap): - for prefix in dirUriMap: - prefixAbs = os.path.abspath(prefix) - if inFile.startswith(prefixAbs): - inFile = prefix + inFile[len(prefixAbs):] - break - else: - raise ValueError("can't correct %s to start with one of %s" % - (inFile, dirUriMap.keys())) - return inFile
--- a/light9/rdfdb/grapheditapi.py Fri Jun 16 04:28:26 2017 +0000 +++ /dev/null Thu Jan 01 00:00:00 1970 +0000 @@ -1,118 +0,0 @@ -import random, logging -from itertools import chain -from rdflib import URIRef, RDF -from light9.rdfdb.patch import Patch, quadsWithContextUris -log = logging.getLogger('graphedit') - -class GraphEditApi(object): - """ - fancier graph edits - - mixin for SyncedGraph, separated here because these methods work together - """ - - def getObjectPatch(self, context, subject, predicate, newObject): - """send a patch which removes existing values for (s,p,*,c) - and adds (s,p,newObject,c). Values in other graphs are not affected. - - newObject can be None, which will remove all (subj,pred,*) statements. - """ - - existing = [] - for spo in self._graph.triples((subject, predicate, None), - context=context): - existing.append(spo+(context,)) - # what layer is supposed to cull out no-op changes? - return Patch( - delQuads=existing, - addQuads=([(subject, predicate, newObject, context)] - if newObject is not None else [])) - - def patchObject(self, context, subject, predicate, newObject): - p = self.getObjectPatch(context, subject, predicate, newObject) - log.info("patchObject %r" % p.jsonRepr) - self.patch(p) - - def patchSubgraph(self, context, newGraph): - """ - replace all statements in 'context' with the quads in newGraph. - This is not cooperating with currentState. - """ - old = set(quadsWithContextUris(self._graph.quads((None, None, None, context)))) - new = set(quadsWithContextUris(newGraph)) - p = Patch(delQuads=old - new, addQuads=new - old) - self.patch(p) - - def patchMapping(self, context, subject, predicate, nodeClass, keyPred, valuePred, newKey, newValue): - """ - creates/updates a structure like this: - - ?subject ?predicate [ - a ?nodeClass; - ?keyPred ?newKey; - ?valuePred ?newValue ] . - - There should be a complementary readMapping that gets you a - value since that's tricky too - """ - - # as long as currentState is expensive and has the - # tripleFilter optimization, this looks like a mess. If - # currentState became cheap, a lot of code here could go away. - - with self.currentState(tripleFilter=(subject, predicate, None)) as current: - adds = set([]) - for setting in current.objects(subject, predicate): - with self.currentState(tripleFilter=(setting, keyPred, None)) as current2: - - match = current2.value(setting, keyPred) == newKey - if match: - break - else: - setting = URIRef(subject + "/map/%s" % - random.randrange(999999999)) - adds.update([ - (subject, predicate, setting, context), - (setting, RDF.type, nodeClass, context), - (setting, keyPred, newKey, context), - ]) - - with self.currentState(tripleFilter=(setting, valuePred, None)) as current: - dels = set([]) - for prev in current.objects(setting, valuePred): - dels.add((setting, valuePred, prev, context)) - adds.add((setting, valuePred, newValue, context)) - - if adds != dels: - self.patch(Patch(delQuads=dels, addQuads=adds)) - - def removeMappingNode(self, context, node): - """ - removes the statements with this node as subject or object, which - is the right amount of statements to remove a node that - patchMapping made. - """ - p = Patch(delQuads=[spo+(context,) for spo in - chain(self._graph.triples((None, None, node), - context=context), - self._graph.triples((node, None, None), - context=context))]) - self.patch(p) - -import unittest -from rdflib import ConjunctiveGraph -class TestPatchSubgraph(unittest.TestCase): - def testCollapsesIdenticalQuads(self): - appliedPatches = [] - class Obj(GraphEditApi): - def patch(self, p): - appliedPatches.append(p) - obj = Obj() - obj._graph = ConjunctiveGraph() - stmt1 = (URIRef('s'), URIRef('p'), URIRef('o'), URIRef('g')) - obj._graph.addN([stmt1]) - obj.patchSubgraph(URIRef('g'), [stmt1]) - self.assertEqual(len(appliedPatches), 1) - p = appliedPatches[0] - self.assert_(p.isNoop()) - self.assertEqual(p.jsonRepr, '{"patch": {"adds": "", "deletes": ""}}')
--- a/light9/rdfdb/graphfile.py Fri Jun 16 04:28:26 2017 +0000 +++ /dev/null Thu Jan 01 00:00:00 1970 +0000 @@ -1,252 +0,0 @@ -import logging, traceback, os, time -from twisted.python.filepath import FilePath -from twisted.internet import reactor -from twisted.internet.inotify import humanReadableMask -from rdflib import Graph, RDF -from light9.rdfdb.patch import Patch -from light9.rdfdb.rdflibpatch import inContext - -log = logging.getLogger('graphfile') -iolog = logging.getLogger('io') - -def patchN3SerializerToUseLessWhitespace(cutColumn=65): - # todo: make a n3serializer subclass with whitespace settings - from rdflib.plugins.serializers.turtle import TurtleSerializer, OBJECT - originalWrite = TurtleSerializer.write - def write(self, s): - lines = s.split('\n') - if len(lines) > 1: - self._column = len(lines[-1]) - else: - self._column += len(lines[0]) - return originalWrite(self, s) - TurtleSerializer.write = write - def predicateList(self, subject, newline=False): - properties = self.buildPredicateHash(subject) - propList = self.sortProperties(properties) - if len(propList) == 0: - return - self.verb(propList[0], newline=newline) - self.objectList(properties[propList[0]]) - for predicate in propList[1:]: - self.write(';') - # can't do proper wrapping since we don't know how much is coming - if self._column > cutColumn: - self.write('\n' + self.indent(1)) - self.verb(predicate, newline=False) - self.objectList(properties[predicate]) - def objectList(self, objects): - count = len(objects) - if count == 0: - return - depthmod = (count == 1) and 0 or 1 - self.depth += depthmod - self.path(objects[0], OBJECT) - for obj in objects[1:]: - self.write(', ') - self.path(obj, OBJECT, newline=True) - self.depth -= depthmod - - originalStatement = TurtleSerializer.statement - def statement(self, subject): - if list(self.store.triples((subject, RDF.type, None))): - self.write('\n') - originalStatement(self, subject) - return False # suppress blank line for 'minor' statements - TurtleSerializer.statement = statement - TurtleSerializer.predicateList = predicateList - TurtleSerializer.objectList = objectList - -patchN3SerializerToUseLessWhitespace() - -class GraphFile(object): - """ - one rdf file that we read from, write to, and notice external changes to - """ - def __init__(self, notifier, path, uri, patch, getSubgraph, globalPrefixes, ctxPrefixes): - """ - uri is the context for the triples in this file. We assume - sometimes that we're the only ones with triples in this - context. - - this does not include an initial reread() call - - Prefixes are mutable dicts. The caller may add to them later. - """ - self.path, self.uri = path, uri - self.patch, self.getSubgraph = patch, getSubgraph - - self.lastWriteTimestamp = 0 # mtime from the last time _we_ wrote - - self.globalPrefixes = globalPrefixes - self.ctxPrefixes = ctxPrefixes - self.readPrefixes = {} - - if not os.path.exists(path): - # can't start notify until file exists - try: - os.makedirs(os.path.dirname(path)) - except OSError: - pass - f = open(path, "w") - f.write("#new\n") - f.close() - iolog.info("%s created", path) - # this was supposed to cut out some extra reads but it - # didn't work: - self.lastWriteTimestamp = os.path.getmtime(path) - - - self.flushDelay = 2 # seconds until we have to call flush() when dirty - self.writeCall = None # or DelayedCall - - self.notifier = notifier - self.addWatch() - - def addWatch(self): - - # emacs save comes in as IN_MOVE_SELF, maybe - - # I was hoping not to watch IN_CHANGED and get lots of - # half-written files, but emacs doesn't close its files after - # a write, so there's no other event. I could try to sleep - # until after all the writes are done, but I think the only - # bug left is that we'll retry too agressively on a file - # that's being written - - from twisted.internet.inotify import IN_CLOSE_WRITE, IN_MOVED_FROM, IN_MODIFY, IN_DELETE, IN_DELETE_SELF, IN_CHANGED - - log.info("add watch on %s", self.path) - self.notifier.watch(FilePath(self.path), callbacks=[self.notify]) - - def notify(self, notifier, filepath, mask): - try: - maskNames = humanReadableMask(mask) - if maskNames[0] == 'delete_self': - if not filepath.exists(): - log.info("%s delete_self", filepath) - self.fileGone() - return - else: - log.warn("%s delete_self event but file is here. " - "probably a new version moved in", - filepath) - - # we could filter these out in the watch() call, but I want - # the debugging - if maskNames[0] in ['open', 'access', 'close_nowrite', 'attrib']: - log.debug("%s %s event, ignoring" % (filepath, maskNames)) - return - - try: - if filepath.getModificationTime() == self.lastWriteTimestamp: - log.debug("%s changed, but we did this write", filepath) - return - except OSError as e: - log.error("%s: %r" % (filepath, e)) - # getting OSError no such file, followed by no future reads - reactor.callLater(.5, self.addWatch) # ? - - return - - log.info("reread %s because of %s event", filepath, maskNames) - - self.reread() - except Exception: - traceback.print_exc() - - def fileGone(self): - """ - our file is gone; remove the statements from that context - """ - myQuads = [(s,p,o,self.uri) for s,p,o in self.getSubgraph(self.uri)] - log.debug("dropping all statements from context %s", self.uri) - if myQuads: - self.patch(Patch(delQuads=myQuads), dueToFileChange=True) - - def reread(self): - """update the graph with any diffs from this file - - n3 parser fails on "1.e+0" even though rdflib was emitting that itself - """ - old = self.getSubgraph(self.uri) - new = Graph() - try: - contents = open(self.path).read() - if contents.startswith("#new"): - log.debug("%s ignoring empty contents of my new file", self.path) - # this is a new file we're starting, and we should not - # patch our graph as if it had just been cleared. We - # shouldn't even be here reading this, but - # lastWriteTimestamp didn't work. - return - - new.parse(location=self.path, format='n3') - self.readPrefixes = dict(new.namespaces()) - except SyntaxError as e: - print e - traceback.print_exc() - log.error("%s syntax error", self.path) - # todo: likely bug- if a file has this error upon first - # read, I think we don't retry it right. - return - except IOError as e: - log.error("%s rereading %s: %r", self.path, self.uri, e) - return - - old = inContext(old, self.uri) - new = inContext(new, self.uri) - - p = Patch.fromDiff(old, new) - if p: - log.debug("%s applying patch for changes in file", self.path) - self.patch(p, dueToFileChange=True) - else: - log.debug("old == new after reread of %s", self.path) - - def dirty(self, graph): - """ - there are new contents to write to our file - - graph is the rdflib.Graph that contains the contents of the - file. It is allowed to change. Note that dirty() will probably - do the save later when the graph might be different. - - after a timer has passed, write it out. Any scheduling issues - between files? i don't think so. the timer might be kind of - huge, and then we might want to take a hint from a client that - it's a good time to save the files that it was editing, like - when the mouse moves out of the client's window and might be - going towards a text file editor - - """ - log.info("%s dirty, needs write", self.path) - - self.graphToWrite = graph - if self.writeCall: - self.writeCall.reset(self.flushDelay) - else: - self.writeCall = reactor.callLater(self.flushDelay, self.flush) - - def flush(self): - self.writeCall = None - - tmpOut = self.path + ".rdfdb-temp" - f = open(tmpOut, 'w') - t1 = time.time() - for p, n in (self.globalPrefixes.items() + - self.readPrefixes.items() + - self.ctxPrefixes.items()): - self.graphToWrite.bind(p, n) - self.graphToWrite.serialize(destination=f, format='n3') - serializeTime = time.time() - t1 - f.close() - self.lastWriteTimestamp = os.path.getmtime(tmpOut) - os.rename(tmpOut, self.path) - iolog.info("%s rewrote in %.1f ms", - self.path, serializeTime * 1000) - - def __repr__(self): - return "%s(path=%r, uri=%r, ...)" % ( - self.__class__.__name__, self.path, self.uri) -
--- a/light9/rdfdb/graphfile_test.py Fri Jun 16 04:28:26 2017 +0000 +++ /dev/null Thu Jan 01 00:00:00 1970 +0000 @@ -1,37 +0,0 @@ -import unittest -import mock -import tempfile -from rdflib import URIRef, Graph -from light9.rdfdb.graphfile import GraphFile - -class TestGraphFileOutput(unittest.TestCase): - def testMaintainsN3PrefixesFromInput(self): - tf = tempfile.NamedTemporaryFile(suffix='_test.n3') - tf.write(''' - @prefix : <http://example.com/> . - @prefix n: <http://example.com/n/> . - :foo n:bar :baz . - ''') - tf.flush() - - def getSubgraph(uri): - return Graph() - gf = GraphFile(mock.Mock(), tf.name, URIRef('uri'), mock.Mock(), getSubgraph, {}, {}) - gf.reread() - - newGraph = Graph() - newGraph.add((URIRef('http://example.com/boo'), - URIRef('http://example.com/n/two'), - URIRef('http://example.com/other/ns'))) - gf.dirty(newGraph) - gf.flush() - wroteContent = open(tf.name).read() - self.assertEqual('''@prefix : <http://example.com/> . -@prefix n: <http://example.com/n/> . -@prefix rdf: <http://www.w3.org/1999/02/22-rdf-syntax-ns#> . -@prefix rdfs: <http://www.w3.org/2000/01/rdf-schema#> . -@prefix xml: <http://www.w3.org/XML/1998/namespace> . -@prefix xsd: <http://www.w3.org/2001/XMLSchema#> . - -:boo n:two <http://example.com/other/ns> . -''', wroteContent)
--- a/light9/rdfdb/localsyncedgraph.py Fri Jun 16 04:28:26 2017 +0000 +++ /dev/null Thu Jan 01 00:00:00 1970 +0000 @@ -1,21 +0,0 @@ -from rdflib import ConjunctiveGraph - -from light9.rdfdb.currentstategraphapi import CurrentStateGraphApi -from light9.rdfdb.autodepgraphapi import AutoDepGraphApi -from light9.rdfdb.grapheditapi import GraphEditApi -from light9.rdfdb.rdflibpatch import patchQuads - -class LocalSyncedGraph(CurrentStateGraphApi, AutoDepGraphApi, GraphEditApi): - """for tests""" - def __init__(self, files=None): - self._graph = ConjunctiveGraph() - for f in files or []: - self._graph.parse(f, format='n3') - - - def patch(self, p): - patchQuads(self._graph, - deleteQuads=p.delQuads, - addQuads=p.addQuads, - perfect=True) - # no deps
--- a/light9/rdfdb/mock_syncedgraph.py Fri Jun 16 04:28:26 2017 +0000 +++ /dev/null Thu Jan 01 00:00:00 1970 +0000 @@ -1,50 +0,0 @@ - -from rdflib import Graph, RDF, RDFS -from rdflib.parser import StringInputSource - -class MockSyncedGraph(object): - """ - Lets users of SyncedGraph mostly work. Doesn't yet help with any - testing of the rerun-upon-graph-change behavior. - """ - def __init__(self, n3Content): - self._graph = Graph() - self._graph.parse(StringInputSource(n3Content), format='n3') - - def addHandler(self, func): - func() - - def value(self, subject=None, predicate=RDF.value, object=None, - default=None, any=True): - if object is not None: - raise NotImplementedError() - return self._graph.value(subject, predicate, object=object, - default=default, any=any) - - def objects(self, subject=None, predicate=None): - return self._graph.objects(subject, predicate) - - def label(self, uri): - return self.value(uri, RDFS.label) - - def subjects(self, predicate=None, object=None): - return self._graph.subjects(predicate, object) - - def predicate_objects(self, subject): - return self._graph.predicate_objects(subject) - - def items(self, listUri): - """generator. Having a chain of watchers on the results is not - well-tested yet""" - chain = set([listUri]) - while listUri: - item = self.value(listUri, RDF.first) - if item: - yield item - listUri = self.value(listUri, RDF.rest) - if listUri in chain: - raise ValueError("List contains a recursive rdf:rest reference") - chain.add(listUri) - - def contains(self, triple): - return triple in self._graph
--- a/light9/rdfdb/patch.py Fri Jun 16 04:28:26 2017 +0000 +++ /dev/null Thu Jan 01 00:00:00 1970 +0000 @@ -1,222 +0,0 @@ -import json, unittest -from rdflib import ConjunctiveGraph, Graph, URIRef, URIRef as U, Literal -from light9.namespaces import XSD -from light9.rdfdb.rdflibpatch import graphFromNQuad, graphFromQuads, serializeQuad - -ALLSTMTS = (None, None, None) - -def quadsWithContextUris(quads): - """ - yield the given quads, correcting any context values that are - Graphs into URIRefs - """ - if isinstance(quads, ConjunctiveGraph): - quads = quads.quads(ALLSTMTS) - for s,p,o,c in quads: - if isinstance(c, Graph): - c = c.identifier - if not isinstance(c, URIRef): - raise TypeError("bad quad context type in %r" % ((s,p,o,c),)) - yield s,p,o,c - -class Patch(object): - """ - immutable - - the json representation includes the {"patch":...} wrapper - """ - def __init__(self, jsonRepr=None, addQuads=None, delQuads=None, - addGraph=None, delGraph=None): - """ - addQuads/delQuads can be lists or sets, but if we make them internally, - they'll be lists - - 4th element of a quad must be a URIRef - """ - self._jsonRepr = jsonRepr - self._addQuads, self._delQuads = addQuads, delQuads - self._addGraph, self._delGraph = addGraph, delGraph - - if self._jsonRepr is not None: - body = json.loads(self._jsonRepr) - self._delGraph = graphFromNQuad(body['patch']['deletes']) - self._addGraph = graphFromNQuad(body['patch']['adds']) - if 'senderUpdateUri' in body: - self.senderUpdateUri = body['senderUpdateUri'] - - def __str__(self): - def shorten(n): - if isinstance(n, Literal): - if n.datatype == XSD['double']: - return str(n.toPython()) - if isinstance(n, URIRef): - for long, short in [ - ("http://light9.bigasterisk.com/", "l9"), - - ]: - if n.startswith(long): - return short+":"+n[len(long):] - return n.n3() - def formatQuad(quad): - return " ".join(shorten(n) for n in quad) - delLines = [" -%s" % formatQuad(q) for q in self.delQuads] - addLines = [" +%s" % formatQuad(q) for q in self.addQuads] - return "\nPatch:\n" + "\n".join(delLines) + "\n" + "\n".join(addLines) - - def shortSummary(self): - return "[-%s +%s]" % (len(self.delQuads), len(self.addQuads)) - - @classmethod - def fromDiff(cls, oldGraph, newGraph): - """ - make a patch that changes oldGraph to newGraph - """ - old = set(quadsWithContextUris(oldGraph)) - new = set(quadsWithContextUris(newGraph)) - return cls(addQuads=list(new - old), delQuads=list(old - new)) - - def __nonzero__(self): - """ - does this patch do anything to a graph? - """ - if self._jsonRepr and self._jsonRepr.strip(): - raise NotImplementedError() - return bool(self._addQuads or self._delQuads or - self._addGraph or self._delGraph) - - @property - def addQuads(self): - if self._addQuads is None: - if self._addGraph is None: - return [] - self._addQuads = list(quadsWithContextUris( - self._addGraph.quads(ALLSTMTS))) - return self._addQuads - - @property - def delQuads(self): - if self._delQuads is None: - if self._delGraph is None: - return [] - self._delQuads = list(quadsWithContextUris( - self._delGraph.quads(ALLSTMTS))) - return self._delQuads - - @property - def addGraph(self): - if self._addGraph is None: - self._addGraph = graphFromQuads(self.addQuads) - return self._addGraph - - @property - def delGraph(self): - if self._delGraph is None: - self._delGraph = graphFromQuads(self.delQuads) - return self._delGraph - - @property - def jsonRepr(self): - if self._jsonRepr is None: - self._jsonRepr = self.makeJsonRepr() - return self._jsonRepr - - def makeJsonRepr(self, extraAttrs={}): - d = {"patch" : { - 'adds' : serializeQuad(self.addGraph), - 'deletes' : serializeQuad(self.delGraph), - }} - if len(self.addGraph) > 0 and d['patch']['adds'].strip() == "": - # this is the bug that graphFromNQuad works around - raise ValueError("nquads serialization failure") - if '[<' in d['patch']['adds']: - raise ValueError("[< found in %s" % d['patch']['adds']) - d.update(extraAttrs) - return json.dumps(d) - - def concat(self, more): - """ - new Patch with the result of applying this patch and the - sequence of other Patches - """ - # not working yet - adds = set(self.addQuads) - dels = set(self.delQuads) - for p2 in more: - for q in p2.delQuads: - if q in adds: - adds.remove(q) - else: - dels.add(q) - for q in p2.addQuads: - if q in dels: - dels.remove(q) - else: - adds.add(q) - return Patch(delQuads=dels, addQuads=adds) - - def getContext(self): - """assumes that all the edits are on the same context""" - ctx = None - for q in self.addQuads + self.delQuads: - if ctx is None: - ctx = q[3] - - if ctx != q[3]: - raise ValueError("patch applies to multiple contexts, at least %r and %r" % (ctx, q[3])) - if ctx is None: - raise ValueError("patch affects no contexts") - assert isinstance(ctx, URIRef), ctx - return ctx - - def isNoop(self): - return set(self.addQuads) == set(self.delQuads) - - -stmt1 = U('http://a'), U('http://b'), U('http://c'), U('http://ctx1') - -class TestPatchFromDiff(unittest.TestCase): - def testEmpty(self): - g = ConjunctiveGraph() - p = Patch.fromDiff(g, g) - self.assert_(not p) - - def testNonEmpty(self): - g1 = ConjunctiveGraph() - g2 = graphFromQuads([stmt1]) - p = Patch.fromDiff(g1, g2) - self.assert_(p) - - def testNoticesAdds(self): - g1 = ConjunctiveGraph() - g2 = graphFromQuads([stmt1]) - p = Patch.fromDiff(g1, g2) - self.assertEqual(p.addQuads, [stmt1]) - self.assertEqual(p.delQuads, []) - - def testNoticesDels(self): - g1 = graphFromQuads([stmt1]) - g2 = ConjunctiveGraph() - p = Patch.fromDiff(g1, g2) - self.assertEqual(p.addQuads, []) - self.assertEqual(p.delQuads, [stmt1]) - - def testQuadSequenceOkInsteadOfGraph(self): - p = Patch.fromDiff([stmt1], ConjunctiveGraph()) - self.assertEqual(p.delQuads, [stmt1]) - p = Patch.fromDiff(ConjunctiveGraph(), [stmt1]) - self.assertEqual(p.addQuads, [stmt1]) - -class TestPatchGetContext(unittest.TestCase): - def testEmptyPatchCantGiveContext(self): - p = Patch() - self.assertRaises(ValueError, p.getContext) - - def testSimplePatchReturnsContext(self): - p = Patch(addQuads=[stmt1]) - self.assertEqual(p.getContext(), U('http://ctx1')) - - def testMultiContextPatchFailsToReturnContext(self): - p = Patch(addQuads=[stmt1[:3] + (U('http://ctx1'),), - stmt1[:3] + (U('http://ctx2'),)]) - self.assertRaises(ValueError, p.getContext) -
--- a/light9/rdfdb/patchreceiver.py Fri Jun 16 04:28:26 2017 +0000 +++ /dev/null Thu Jan 01 00:00:00 1970 +0000 @@ -1,65 +0,0 @@ -import logging, cyclone.httpclient, traceback, urllib -from twisted.internet import reactor -from light9 import networking -from light9.rdfdb.patch import Patch -log = logging.getLogger('syncedgraph') - -class PatchReceiver(object): - """ - runs a web server in this process and registers it with the rdfdb - master. See onPatch for what happens when the rdfdb master sends - us a patch - """ - def __init__(self, rdfdbRoot, label, onPatch): - """ - label is what we'll call ourselves to the rdfdb server - - onPatch is what we call back when the server sends a patch - """ - self.rdfdbRoot = rdfdbRoot - listen = reactor.listenTCP(0, cyclone.web.Application(handlers=[ - (r'/update', makePatchEndpoint(onPatch)), - ])) - port = listen._realPortNumber # what's the right call for this? - - self.updateResource = 'http://%s:%s/update' % ( - networking.patchReceiverUpdateHost.value, port) - log.info("listening on %s" % port) - self._register(label) - - def _register(self, label): - url = self.rdfdbRoot + 'graphClients' - body = urllib.urlencode([('clientUpdate', self.updateResource), - ('label', label)]) - cyclone.httpclient.fetch( - url=url, - method='POST', - headers={'Content-Type': ['application/x-www-form-urlencoded']}, - postdata=body, - ).addCallbacks(self._done, - lambda err: self._registerError(err, url, body)) - log.info("registering with rdfdb at %s", url) - - def _registerError(self, err, url, body): - log.error('registering to url=%r body=%r', url, body) - log.error(err) - - def _done(self, x): - log.debug("registered with rdfdb") - - -def makePatchEndpointPutMethod(cb): - def put(self): - try: - p = Patch(jsonRepr=self.request.body) - log.debug("received patch -%d +%d" % (len(p.delGraph), len(p.addGraph))) - cb(p) - except: - traceback.print_exc() - raise - return put - -def makePatchEndpoint(cb): - class Update(cyclone.web.RequestHandler): - put = makePatchEndpointPutMethod(cb) - return Update
--- a/light9/rdfdb/patchsender.py Fri Jun 16 04:28:26 2017 +0000 +++ /dev/null Thu Jan 01 00:00:00 1970 +0000 @@ -1,120 +0,0 @@ -import logging, time -import cyclone.httpclient -from twisted.internet import defer -log = logging.getLogger('syncedgraph') - -class PatchSender(object): - """ - SyncedGraph may generate patches faster than we can send - them. This object buffers and may even collapse patches before - they go the server - """ - def __init__(self, target, myUpdateResource): - """ - target is the URI we'll send patches to - - myUpdateResource is the URI for this sender of patches, which - maybe needs to be the actual requestable update URI for - sending updates back to us - """ - self.target = target - self.myUpdateResource = myUpdateResource - self._patchesToSend = [] - self._currentSendPatchRequest = None - - def sendPatch(self, p): - sendResult = defer.Deferred() - self._patchesToSend.append((p, sendResult)) - self._continueSending() - return sendResult - - def cancelAll(self): - self._patchesToSend[:] = [] - # we might be in the middle of a post; ideally that would be - # aborted, too. Since that's not coded yet, or it might be too late to - # abort, what should happen? - # 1. this could return deferred until we think our posts have stopped - # 2. or, other code could deal for the fact that cancelAll - # isn't perfect - - def _continueSending(self): - if not self._patchesToSend or self._currentSendPatchRequest: - return - if len(self._patchesToSend) > 1: - log.debug("%s patches left to send", len(self._patchesToSend)) - # this is where we could concatenate little patches into a - # bigger one. Often, many statements will cancel each - # other out. not working yet: - if 0: - p = self._patchesToSend[0].concat(self._patchesToSend[1:]) - print "concat down to" - print 'dels' - for q in p.delQuads: print q - print 'adds' - for q in p.addQuads: print q - print "----" - else: - p, sendResult = self._patchesToSend.pop(0) - else: - p, sendResult = self._patchesToSend.pop(0) - - self._currentSendPatchRequest = sendPatch( - self.target, p, senderUpdateUri=self.myUpdateResource) - self._currentSendPatchRequest.addCallbacks(self._sendPatchDone, - self._sendPatchErr) - self._currentSendPatchRequest.chainDeferred(sendResult) - - def _sendPatchDone(self, result): - self._currentSendPatchRequest = None - self._continueSending() - - def _sendPatchErr(self, e): - self._currentSendPatchRequest = None - # we're probably out of sync with the master now, since - # SyncedGraph.patch optimistically applied the patch to our - # local graph already. What happens to this patch? What - # happens to further pending patches? Some of the further - # patches, especially, may be commutable with the bad one and - # might still make sense to apply to the master graph. - - # if someday we are folding pending patches together, this - # would be the time to UNDO that and attempt the original - # separate patches again - - # this should screen for 409 conflict responses and raise a - # special exception for that, so SyncedGraph.sendFailed can - # screen for only that type - - # this code is going away; we're going to raise an exception that contains all the pending patches - log.error("_sendPatchErr") - log.error(e) - self._continueSending() - -def sendPatch(putUri, patch, **kw): - """ - PUT a patch as json to an http server. Returns deferred. - - kwargs will become extra attributes in the toplevel json object - """ - t1 = time.time() - body = patch.makeJsonRepr(kw) - jsonTime = time.time() - t1 - intro = body[:200] - if len(body) > 200: - intro = intro + "..." - log.debug("send body (rendered %.1fkB in %.1fms): %s", len(body) / 1024, jsonTime * 1000, intro) - sendTime = time.time() - def putDone(done): - if not str(done.code).startswith('2'): - raise ValueError("sendPatch request failed %s: %s" % - (done.code, done.body)) - dt = 1000 * (time.time() - sendTime) - log.debug("sendPatch to %s took %.1fms" % (putUri, dt)) - return done - - return cyclone.httpclient.fetch( - url=putUri, - method='PUT', - headers={'Content-Type': ['application/json']}, - postdata=body, - ).addCallback(putDone)
--- a/light9/rdfdb/rdflibpatch.py Fri Jun 16 04:28:26 2017 +0000 +++ /dev/null Thu Jan 01 00:00:00 1970 +0000 @@ -1,201 +0,0 @@ -""" -this is a proposal for a ConjunctiveGraph method in rdflib -""" -import sys -if sys.path[0] == '/usr/lib/python2.7/dist-packages': - # nosetests puts this in - sys.path = sys.path[1:] - -import unittest -from rdflib import ConjunctiveGraph, Graph, URIRef as U, Literal - -def patchQuads(graph, deleteQuads, addQuads, perfect=False): - """ - Delete the sequence of given quads. Then add the given quads just - like addN would. If perfect is True, we'll error before the - deletes or before the adds (not a real transaction) if any of the - deletes isn't in the graph or if any of the adds was already in - the graph. - - These input quads use URIRef for the context, but - Graph(identifier=) is also allowed (which is what you'll get - sometimes from rdflib APIs). - """ - toDelete = [] - for spoc in deleteQuads: - spoc = fixContextToUri(spoc) - - if perfect: - if inGraph(spoc, graph): - toDelete.append(spoc) - else: - raise ValueError("%r not in %r" % (spoc[:3], spoc[3])) - else: - graph.remove(spoc) - for spoc in toDelete: - graph.remove(spoc) - - if perfect: - addQuads = list(addQuads) - for spoc in addQuads: - spoc = fixContextToUri(spoc) - if inGraph(spoc, graph): - raise ValueError("%r already in %r" % (spoc[:3], spoc[3])) - graph.addN(addQuads) - -def fixContextToUri(spoc): - if not isinstance(spoc[3], U): - return spoc[:3] + (spoc[3].identifier,) - return spoc - -def inGraph(spoc, graph): - """ - c is just a URIRef. - Workaround for https://github.com/RDFLib/rdflib/issues/398 - """ - spoi = spoc[:3] + (Graph(identifier=spoc[3]),) - if spoi not in graph: - # this is a huge speedup, avoid many whole-graph scans - return False - return spoi in graph.quads() - -# some of the following workarounds may be fixed in https://github.com/RDFLib/rdflib/issues/299 -def graphFromQuads(q): - g = ConjunctiveGraph() - #g.addN(q) # no effect on nquad output - for s,p,o,c in q: - #g.get_context(c).add((s,p,o)) # kind of works with broken rdflib nquad serializer code; you need this for json_ld serialize to work :( - g.store.add((s,p,o), c) # no effect on nquad output - return g - -def graphFromNQuad(text): - g1 = ConjunctiveGraph() - # text might omit ctx on some lines. rdflib just puts in a bnode, which shows up later. - g1.parse(data=text, format='nquads') - return g1 - -from rdflib.plugins.serializers.nt import _quoteLiteral -def serializeQuad(g): - """ - replacement for graph.serialize(format='nquads') - - Still broken in rdflib 4.2.2: graph.serialize(format='nquads') - returns empty string for my graph in - TestGraphFromQuads.testSerializes. - """ - out = [] - for s,p,o,c in g.quads((None,None,None)): - if isinstance(c, Graph): - # still not sure why this is Graph sometimes, - # already URIRef other times - c = c.identifier - if '[' in c.n3(): - import ipdb;ipdb.set_trace() - ntObject = _quoteLiteral(o) if isinstance(o, Literal) else o.n3() - out.append(u"%s %s %s %s .\n" % (s.n3(), - p.n3(), - ntObject, - c.n3())) - return ''.join(out) - -def inContext(graph, newContext): - """ - make a ConjunctiveGraph where all the triples in the given graph - (or collection) are now in newContext (a uri) - """ - return graphFromQuads((s,p,o,newContext) for s,p,o in graph) - -def contextsForStatement(graph, triple): - return [q[3] for q in graph.quads(triple)] - - -A = U("http://a"); B = U("http://b") -class TestInContext(unittest.TestCase): - def testResultHasQuads(self): - g = inContext([(A,A,A)], B) - self.assertEqual(list(g.quads())[0], (A,A,A,B)) - -class TestContextsForStatement(unittest.TestCase): - def testNotFound(self): - g = graphFromQuads([(A,A,A,A)]) - self.assertEqual(contextsForStatement(g, (B,B,B)), []) - def testOneContext(self): - g = graphFromQuads([(A,A,A,A), (A,A,B,B)]) - self.assertEqual(contextsForStatement(g, (A,A,A)), [A]) - def testTwoContexts(self): - g = graphFromQuads([(A,A,A,A), (A,A,A,B)]) - self.assertEqual(sorted(contextsForStatement(g, (A,A,A))), sorted([A,B])) - # There's a case where contextsForStatement was returning a Graph - # with identifier, which I've fixed without a test - - -class TestGraphFromQuads(unittest.TestCase): - nqOut = '<http://example.com/> <http://example.com/> <http://example.com/> <http://example.com/> .\n' - def testSerializes(self): - n = U("http://example.com/") - g = graphFromQuads([(n,n,n,n)]) - out = serializeQuad(g) - self.assertEqual(out.strip(), self.nqOut.strip()) - - def testNquadParserSerializes(self): - g = graphFromNQuad(self.nqOut) - self.assertEqual(len(g), 1) - out = serializeQuad(g) - self.assertEqual(out.strip(), self.nqOut.strip()) - - -A = U("http://a"); B = U("http://b"); C = U("http://c") -CTX1 = U('http://ctx1'); CTX2 = U('http://ctx2') -stmt1 = A, B, C, CTX1 -stmt2 = A, B, C, CTX2 -class TestPatchQuads(unittest.TestCase): - def testAddsToNewContext(self): - g = ConjunctiveGraph() - patchQuads(g, [], [stmt1]) - self.assert_(len(g), 1) - quads = list(g.quads((None,None,None))) - self.assertEqual(quads, [(A, B, C, Graph(identifier=CTX1))]) - - def testDeletes(self): - g = ConjunctiveGraph() - patchQuads(g, [], [stmt1]) - patchQuads(g, [stmt1], []) - quads = list(g.quads((None,None,None))) - self.assertEqual(quads, []) - - def testDeleteRunsBeforeAdd(self): - g = ConjunctiveGraph() - patchQuads(g, [stmt1], [stmt1]) - quads = list(g.quads((None,None,None))) - self.assertEqual(quads, [(A, B, C, Graph(identifier=CTX1))]) - - def testPerfectAddRejectsExistingStmt(self): - g = ConjunctiveGraph() - patchQuads(g, [], [stmt1]) - self.assertRaises(ValueError, patchQuads, g, [], [stmt1], perfect=True) - - def testPerfectAddAllowsExistingStmtInNewContext(self): - g = ConjunctiveGraph() - patchQuads(g, [], [stmt1]) - patchQuads(g, [], [stmt2], perfect=True) - self.assertEqual(len(list(g.quads((None,None,None)))), 2) - - def testPerfectDeleteRejectsAbsentStmt(self): - g = ConjunctiveGraph() - self.assertRaises(ValueError, patchQuads, g, [stmt1], [], perfect=True) - - def testPerfectDeleteRejectsStmtFromOtherGraph(self): - g = ConjunctiveGraph() - patchQuads(g, [], [stmt2]) - self.assertRaises(ValueError, patchQuads, g, [stmt1], [], perfect=True) - - def testPerfectDeleteAllowsRemovalOfStmtInMultipleContexts(self): - g = ConjunctiveGraph() - patchQuads(g, [], [stmt1, stmt2]) - patchQuads(g, [stmt1], [], perfect=True) - - def testRedundantStmtOkForAddOrDelete(self): - g = ConjunctiveGraph() - patchQuads(g, [], [stmt1, stmt1], perfect=True) - patchQuads(g, [stmt1, stmt1], [], perfect=True) -
--- a/light9/rdfdb/rdflibpatch_literal.py Fri Jun 16 04:28:26 2017 +0000 +++ /dev/null Thu Jan 01 00:00:00 1970 +0000 @@ -1,56 +0,0 @@ -import sys -if sys.path[0] == '/usr/lib/python2.7/dist-packages': - # nosetests puts this in - sys.path = sys.path[1:] - -from rdflib.term import _PLAIN_LITERAL_TYPES, _XSD_DOUBLE, _XSD_DECIMAL, Literal -from re import sub - -def _literal_n3(self, use_plain=False, qname_callback=None): - if use_plain and self.datatype in _PLAIN_LITERAL_TYPES: - try: - self.toPython() # check validity - # this is a bit of a mess - - # in py >=2.6 the string.format function makes this easier - # we try to produce "pretty" output - if self.datatype == _XSD_DOUBLE: - # this is the drewp fix - return sub(r"\.?0*e","e", u'%e' % float(self)) - elif self.datatype == _XSD_DECIMAL: - return sub("0*$","0",u'%f' % float(self)) - else: - return u'%s' % self - except ValueError: - pass # if it's in, we let it out? - - encoded = self._quote_encode() - - datatype = self.datatype - quoted_dt = None - if datatype: - if qname_callback: - quoted_dt = qname_callback(datatype) - if not quoted_dt: - quoted_dt = "<%s>" % datatype - - language = self.language - if language: - if datatype: - # TODO: this isn't valid RDF (it's datatype XOR language) - return '%s@%s^^%s' % (encoded, language, quoted_dt) - return '%s@%s' % (encoded, language) - elif datatype: - return '%s^^%s' % (encoded, quoted_dt) - else: - return '%s' % encoded - -def patch(): - Literal._literal_n3 = _literal_n3 - -import unittest -class TestDoubleOutput(unittest.TestCase): - def testNoDanglingPoint(self): - vv = Literal("0.88", datatype=_XSD_DOUBLE) - out = _literal_n3(vv, use_plain=True) - print out - self.assert_(out in ["8.8e-01", "0.88"], out)
--- a/light9/rdfdb/syncedgraph.py Fri Jun 16 04:28:26 2017 +0000 +++ /dev/null Thu Jan 01 00:00:00 1970 +0000 @@ -1,167 +0,0 @@ -""" -client code uses a SyncedGraph, which has a few things: - -AutoDepGraphApi - knockoutjs-inspired API for querying the graph in a -way that lets me call you again when there were changes to the things -you queried - -CurrentStateGraphApi - a way to query the graph that doesn't gather -your dependencies like AutoDepGraphApi does - -GraphEditApi - methods to write patches to the graph for common -operations, e.g. replacing a value, or editing a mapping - -PatchReceiver - our web server that listens to edits from the master graph - -PatchSender - collects and transmits your graph edits -""" - -from rdflib import ConjunctiveGraph -import logging, cyclone.httpclient, traceback -from twisted.internet import defer -import treq, json -log = logging.getLogger('syncedgraph') -from light9.rdfdb.rdflibpatch import patchQuads - -from light9.rdfdb.patchsender import PatchSender -from light9.rdfdb.patchreceiver import PatchReceiver -from light9.rdfdb.currentstategraphapi import CurrentStateGraphApi -from light9.rdfdb.autodepgraphapi import AutoDepGraphApi -from light9.rdfdb.grapheditapi import GraphEditApi - -# everybody who writes literals needs to get this -from rdflibpatch_literal import patch -patch() - - -class SyncedGraph(CurrentStateGraphApi, AutoDepGraphApi, GraphEditApi): - """ - graph for clients to use. Changes are synced with the master graph - in the rdfdb process. - - This api is like rdflib.Graph but it can also call you back when - there are graph changes to the parts you previously read. - - You may want to attach to self.initiallySynced deferred so you - don't attempt patches before we've heard the initial contents of - the graph. It would be ok to accumulate some patches of new - material, but usually you won't correctly remove the existing - statements unless we have the correct graph. - - If we get out of sync, we abandon our local graph (even any - pending local changes) and get the data again from the - server. - """ - def __init__(self, rdfdbRoot, label): - """ - label is a string that the server will display in association - with your connection - """ - self.rdfdbRoot = rdfdbRoot - self.initiallySynced = defer.Deferred() - self._graph = ConjunctiveGraph() - - self._receiver = PatchReceiver(self.rdfdbRoot, label, self._onPatch) - - self._sender = PatchSender(self.rdfdbRoot + 'patches', - self._receiver.updateResource) - AutoDepGraphApi.__init__(self) - # this needs more state to track if we're doing a resync (and - # everything has to error or wait) or if we're live - - def resync(self): - """ - get the whole graph again from the server (e.g. we had a - conflict while applying a patch and want to return to the - truth). - - To avoid too much churn, we remember our old graph and diff it - against the replacement. This way, our callers only see the - corrections. - - Edits you make during a resync will surely be lost, so I - should just fail them. There should be a notification back to - UIs who want to show that we're doing a resync. - """ - log.info('resync') - self._sender.cancelAll() - # this should be locked so only one resync goes on at once - return cyclone.httpclient.fetch( - url=self.rdfdbRoot + "graph", - method="GET", - headers={'Accept':['x-trig']}, - ).addCallback(self._resyncGraph) - - def _resyncGraph(self, response): - log.warn("new graph in") - - #diff against old entire graph - #broadcast that change - - def patch(self, p): - """send this patch to the server and apply it to our local - graph and run handlers""" - - if p.isNoop(): - log.info("skipping no-op patch") - return - - # these could fail if we're out of sync. One approach: - # Rerequest the full state from the server, try the patch - # again after that, then give up. - debugKey = '[id=%s]' % (id(p) % 1000) - log.debug("\napply local patch %s %s", debugKey, p) - try: - patchQuads(self._graph, - deleteQuads=p.delQuads, - addQuads=p.addQuads, - perfect=True) - except ValueError as e: - log.error(e) - self.sendFailed(None) - return - log.debug('runDepsOnNewPatch') - self.runDepsOnNewPatch(p) - log.debug('sendPatch') - self._sender.sendPatch(p).addErrback(self.sendFailed) - log.debug('patch is done %s', debugKey) - - def suggestPrefixes(self, ctx, prefixes): - """ - when writing files for this ctx, try to use these n3 - prefixes. async, not guaranteed to finish before any - particular file flush - """ - treq.post(self.rdfdbRoot + 'prefixes', json.dumps({'ctx': ctx, 'prefixes': prefixes})) - - def sendFailed(self, result): - """ - we asked for a patch to be queued and sent to the master, and - that ultimately failed because of a conflict - """ - log.warn("sendFailed") - self.resync() - - #i think we should receive back all the pending patches, - #do a resync here, - #then requeue all the pending patches (minus the failing one?) after that's done. - - def _onPatch(self, p): - """ - central server has sent us a patch - """ - log.debug('_onPatch server has sent us %s', p) - patchQuads(self._graph, p.delQuads, p.addQuads, perfect=True) - log.debug("graph now has %s statements" % len(self._graph)) - try: - self.runDepsOnNewPatch(p) - except Exception: - # don't reflect this error back to the server; we did - # receive its patch correctly. However, we're in a bad - # state since some dependencies may not have rerun - traceback.print_exc() - log.warn("some graph dependencies may not have completely run") - - if self.initiallySynced: - self.initiallySynced.callback(None) - self.initiallySynced = None
--- a/light9/rdfdb/web/graphView.html Fri Jun 16 04:28:26 2017 +0000 +++ /dev/null Thu Jan 01 00:00:00 1970 +0000 @@ -1,42 +0,0 @@ -<!DOCTYPE html> -<html> - <head> - <title>graphview</title> - <link rel="stylesheet" type="text/css" href="/style.css"> - - </head> - <body> - <!-- - this page is another graph client who maintains the whole - graph all the time - --> - - <div id="status">starting...</div> - - <fieldset> - <legend>Messages</legend> - <div id="out"></div> - </fieldset> - - <p>URI substring: <input type="text" id="uriSubstring"></p> - - <table> - <thead> - <tr> - <th>subject</th> <th>predicate</th> <th>object</th> <th>context</th> - </tr> - </thead> - <tbody> - </tbody> - </table> - - <script type="text/javascript" src="/lib/jquery/dist/jquery.min.js"></script> - <script type="text/javascript" src="/websocket.js"></script> - <script type="text/javascript" src="syncedgraph.js"></script> - <script type="text/javascript"> - $(function(){ - - }); - </script> - </body> -</html>
--- a/light9/rdfdb/web/index.html Fri Jun 16 04:28:26 2017 +0000 +++ /dev/null Thu Jan 01 00:00:00 1970 +0000 @@ -1,58 +0,0 @@ -<!DOCTYPE html> -<html> - <head> - <title>rdfdb</title> - <link rel="stylesheet" type="text/css" href="/style.css"> - <link rel="stylesheet" type="text/css" href="style.css"> - </head> - <body id="rdfdb"> - <h1>rdfdb</h1> - <div id="status">starting...</div> - - <section id="edits"> - <h2>Edits</h2> - <div id="patches"></div> - </section> - - <p>Clients: <span id="clients"></span></p> - - <section id="messages"> - <h2>Messages</h2> - <div id="out"></div> - </fieldset> - - <script type="text/javascript" src="/lib/jquery/dist/jquery.min.js"></script> - <script type="text/javascript" src="/websocket.js"></script> - <script type="text/javascript"> - $(function(){ - - function collapseCuries(s) { - // this is temporary. The correct thing is to parse the quad (or receive it as a tree) and then make links to the full URIs and display curies of them - - return s - .replace(/<http:\/\/www.w3.org\/2001\/XMLSchema#(.*?)>/g, function (match, short) { return "xsd:"+short; }) - .replace(/<http:\/\/light9.bigasterisk.com\/(.*?)>/g, function (match, short) { return "light9:"+short; }) - .replace(/<http:\/\/light9.bigasterisk.com\/show\/dance2012\/sessions\/(.*?)>/g, function (match, short) { return "kcsession:"+short }); - } - - function onMessage(d) { - if (d.clients !== undefined) { - $("#clients").empty().text(JSON.stringify(d.clients)); - } - if (d.patch !== undefined) { - $("#patches").prepend( - $("<fieldset>").addClass("patch") - .append($("<legend>").text("Patch")) - .append($("<div>").addClass("deletes").text(collapseCuries(d.patch.deletes))) - .append($("<div>").addClass("adds").text(collapseCuries(d.patch.adds))) - ); - } - - $('#out').append($('<div>').text(JSON.stringify(d))); - } - reconnectingWebSocket("live", onMessage); - }); - </script> - - </body> -</html>
--- a/light9/rdfdb/web/style.css Fri Jun 16 04:28:26 2017 +0000 +++ /dev/null Thu Jan 01 00:00:00 1970 +0000 @@ -1,54 +0,0 @@ -body { - display: flex; - flex-direction: column; - position: absolute; - width: 100%; - height: 100%; - padding: 0px; - margin: 0; -} - -#edits { - flex-grow: 3; - display: flex; - flex-direction: column; -} -#messages { - flex-grow: 1; - display: flex; - flex-direction: column; - margin-bottom: 3px; -} - -/* rdfdb */ -#patches { - overflow-y: scroll; - flex: 1 1 0; -} -.patch { - border: 1px solid gray; - padding: 2px; - margin: 4px; -} -.patch > div { - font-family: monospace; - font-size: 90%; - white-space: pre-wrap; -} -.patch .adds { - color: #3AEA38; -} -.patch .deletes { - color: #FF2828; -} -#out { - white-space: pre-wrap; - overflow-y: auto; - /* height: 50px; */ - flex: 1 0 0; -} -.patch fieldset { - color: gray; - font-family: arial; - font-size: 75%; -}
--- a/light9/rdfdb/web/syncedgraph.js Fri Jun 16 04:28:26 2017 +0000 +++ /dev/null Thu Jan 01 00:00:00 1970 +0000 @@ -1,29 +0,0 @@ -function SyncedGraph(label) { - /* - like python SyncedGraph but talks over a websocket to - rdfdb. This one has an API more conducive to reading and - querying. - - light9/web/graph.coffee is the newer attempt - */ - var self = this; - - - - self.patch = function (p) { - throw; - } - self.nodesWithSubstring = function (subString) { - - } - self.quads = function (s, p, o, c) { - // any args can be null for wildcard - } - - - function onMessage(d) { - $('#out').append($('<div>').text(JSON.stringify(d))); - } - - reconnectingWebSocket("liveSyncedGraph", onMessage); -}