Mercurial > code > home > repos > homeauto

comparison lib/patchablegraph.py @ 224:596c645a1fc5

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
refactor /graph and /graph/events handlers to lib/ Ignore-this: fdd8f3d753f76b32929a6a318314d2b5
author drewp@bigasterisk.com
date Sun, 24 Jan 2016 22:53:29 -0800
parents 9236b736bc34
children c1b98006f56e
comparison
equal deleted inserted replaced
223:9236b736bc34 224:596c645a1fc5
1 """
2 Design:
3
4 1. Services each have (named) graphs, which they patch as things
5 change. PatchableGraph is an object for holding this graph.
6 2. You can http GET that graph, or ...
7 3. You can http GET/SSE that graph and hear about modifications to it
8 4. The client that got the graph holds and maintains a copy. The
9 client may merge together multiple graphs.
10 5. Client queries its graph with low-level APIs or client-side sparql.
11 6. When the graph changes, the client knows and can update itself at
12 low or high granularity.
13 """
1 import sys, json 14 import sys, json
2 import cyclone.sse 15 import cyclone.sse
3 sys.path.append("/my/proj/light9") 16 sys.path.append("/my/proj/light9")
4 from light9.rdfdb.grapheditapi import GraphEditApi 17 from light9.rdfdb.grapheditapi import GraphEditApi
5 from rdflib import ConjunctiveGraph 18 from rdflib import ConjunctiveGraph
6 from light9.rdfdb.rdflibpatch import patchQuads 19 from light9.rdfdb.rdflibpatch import patchQuads
7 from rdflib_jsonld.serializer import from_rdf 20 from rdflib_jsonld.serializer import from_rdf
21 from cycloneerr import PrettyErrorHandler
8 22
9 def writeGraphResponse(req, graph, acceptHeader): 23 def writeGraphResponse(req, graph, acceptHeader):
10 if acceptHeader == 'application/nquads': 24 if acceptHeader == 'application/nquads':
11 req.set_header('Content-type', 'application/nquads') 25 req.set_header('Content-type', 'application/nquads')
12 graph.serialize(req, format='nquads') 26 graph.serialize(req, format='nquads')
16 else: 30 else:
17 req.set_header('Content-type', 'application/x-trig') 31 req.set_header('Content-type', 'application/x-trig')
18 graph.serialize(req, format='trig') 32 graph.serialize(req, format='trig')
19 33
20 # forked from /my/proj/light9/light9/rdfdb/rdflibpatch.py 34 # forked from /my/proj/light9/light9/rdfdb/rdflibpatch.py
21 def graphFromQuads2(q): 35 def _graphFromQuads2(q):
22 g = ConjunctiveGraph() 36 g = ConjunctiveGraph()
23 #g.addN(q) # no effect on nquad output 37 #g.addN(q) # no effect on nquad output
24 for s,p,o,c in q: 38 for s,p,o,c in q:
25 g.get_context(c).add((s,p,o)) # kind of works with broken rdflib nquad serializer code 39 g.get_context(c).add((s,p,o)) # kind of works with broken rdflib nquad serializer code
26 #g.store.add((s,p,o), c) # no effect on nquad output 40 #g.store.add((s,p,o), c) # no effect on nquad output
27 return g 41 return g
28 42
29 def patchAsJson(p): 43 def patchAsJson(p):
30 return json.dumps({'patch': { 44 return json.dumps({'patch': {
31 'adds': from_rdf(graphFromQuads2(p.addQuads)), 45 'adds': from_rdf(_graphFromQuads2(p.addQuads)),
32 'deletes': from_rdf(graphFromQuads2(p.delQuads)), 46 'deletes': from_rdf(_graphFromQuads2(p.delQuads)),
33 }}) 47 }})
34 48
35 class PatchableGraph(GraphEditApi): 49 class PatchableGraph(GraphEditApi):
36 """ 50 """
37 Master graph that you modify with self.patch, and we get the 51 Master graph that you modify with self.patch, and we get the
62 self._observers.remove(onPatch) 76 self._observers.remove(onPatch)
63 except ValueError: 77 except ValueError:
64 pass 78 pass
65 79
66 80
81 class CycloneGraphHandler(PrettyErrorHandler, cyclone.web.RequestHandler):
82 def initialize(self, masterGraph):
83 self.masterGraph = masterGraph
67 84
68 class GraphEventsHandler(cyclone.sse.SSEHandler): 85 def get(self):
86 writeGraphResponse(self, self.masterGraph,
87 self.request.headers.get('accept'))
88
89 class CycloneGraphEventsHandler(cyclone.sse.SSEHandler):
69 """ 90 """
70 One session with one client. 91 One session with one client.
71 92
72 returns current graph plus future patches to keep remote version 93 returns current graph plus future patches to keep remote version
73 in sync with ours. 94 in sync with ours.
74 95
75 intsead of turning off buffering all over, it may work for this 96 intsead of turning off buffering all over, it may work for this
76 response to send 'x-accel-buffering: no', per 97 response to send 'x-accel-buffering: no', per
77 http://nginx.org/en/docs/http/ngx_http_proxy_module.html#proxy_buffering 98 http://nginx.org/en/docs/http/ngx_http_proxy_module.html#proxy_buffering
78 """ 99 """
100 def __init__(self, application, request, masterGraph):
101 cyclone.sse.SSEHandler.__init__(self, application, request)
102 self.masterGraph = masterGraph
103
79 def bind(self): 104 def bind(self):
80 mg = self.settings.masterGraph 105 self.sendEvent(
81 # todo: needs to be on one line, or else fix cyclone to stripe headers 106 message=self.masterGraph.serialize(None, format='json-ld',
82 self.sendEvent(message=mg.serialize(None, format='json-ld', indent=None), event='fullGraph') 107 indent=None),
83 mg.addObserver(self.onPatch) 108 event='fullGraph')
109 self.masterGraph.addObserver(self.onPatch)
84 110
85 def onPatch(self, patchJson): 111 def onPatch(self, patchJson):
86 self.sendEvent(message=patchJson, event='patch') 112 self.sendEvent(message=patchJson, event='patch')
87 113
88 def unbind(self): 114 def unbind(self):