Mercurial > code > home > repos > patchablegraph
annotate patchablegraph.py @ 20:8ec07d997cd5
declare labelnames on metrics
| author | drewp@bigasterisk.com |
|---|---|
| date | Wed, 24 Nov 2021 20:29:32 -0800 |
| parents | 388a5e15d249 |
| children | 8d6ba6d372c8 |
| rev | line source |
|---|---|
| 0 | 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 | |
| 14 | |
| 15 See also: | |
| 16 * http://iswc2007.semanticweb.org/papers/533.pdf RDFSync: efficient remote synchronization of RDF | |
| 17 models | |
| 18 * https://www.w3.org/2009/12/rdf-ws/papers/ws07 Supporting Change Propagation in RDF | |
| 19 * https://www.w3.org/DesignIssues/lncs04/Diff.pdf Delta: an ontology for the distribution of | |
| 20 differences between RDF graphs | |
| 21 | |
| 22 """ | |
| 4 | 23 import html |
| 24 import itertools | |
| 25 import json | |
| 26 import logging | |
| 27 from typing import Callable, List, Optional, cast | |
| 0 | 28 |
| 4 | 29 import cyclone.sse |
| 30 import cyclone.web | |
| 31 from cycloneerr import PrettyErrorHandler | |
| 3 | 32 from prometheus_client import Counter, Gauge, Summary |
| 0 | 33 from rdfdb.grapheditapi import GraphEditApi |
| 4 | 34 from rdfdb.patch import Patch |
| 35 from rdfdb.rdflibpatch import inGraph, patchQuads | |
| 0 | 36 from rdflib import ConjunctiveGraph |
| 37 from rdflib.namespace import NamespaceManager | |
| 38 from rdflib.parser import StringInputSource | |
| 39 from rdflib.plugins.serializers.jsonld import from_rdf | |
| 17 | 40 |
| 41 from . import patch_cyclone | |
| 0 | 42 |
|
9
1b6718a54c00
add a workaround for a str/bytes bug in cyclone SSE
drewp@bigasterisk.com
parents:
4
diff
changeset
|
43 patch_cyclone.patch_sse() |
| 0 | 44 log = logging.getLogger('patchablegraph') |
| 45 | |
| 20 | 46 SERIALIZE_CALLS = Summary('serialize_calls', 'PatchableGraph.serialize calls', labelnames=['graph']) |
| 47 PATCH_CALLS = Summary('patch_calls', 'PatchableGraph.patch calls', labelnames=['graph']) | |
| 48 STATEMENT_COUNT = Gauge('statement_count', 'current PatchableGraph graph size', labelnames=['graph']) | |
| 49 OBSERVERS_CURRENT = Gauge('observers_current', 'current observer count', labelnames=['graph']) | |
| 50 OBSERVERS_ADDED = Counter('observers_added', 'observers added', labelnames=['graph']) | |
| 3 | 51 |
| 52 | |
| 0 | 53 # forked from /my/proj/light9/light9/rdfdb/rdflibpatch.py |
| 54 def _graphFromQuads2(q): | |
| 55 g = ConjunctiveGraph() | |
| 56 #g.addN(q) # no effect on nquad output | |
| 4 | 57 for s, p, o, c in q: |
| 58 g.get_context(c).add((s, p, o)) # kind of works with broken rdflib nquad serializer code | |
| 0 | 59 #g.store.add((s,p,o), c) # no effect on nquad output |
| 60 return g | |
| 61 | |
| 4 | 62 |
| 63 def jsonFromPatch(p: Patch) -> str: | |
| 64 return json.dumps( | |
| 65 {'patch': { | |
| 66 'adds': from_rdf(_graphFromQuads2(p.addQuads)), | |
| 67 'deletes': from_rdf(_graphFromQuads2(p.delQuads)), | |
| 68 }}) | |
| 0 | 69 |
| 70 | |
| 4 | 71 patchAsJson = jsonFromPatch # deprecated name |
| 72 | |
| 73 | |
| 74 def patchFromJson(j: str) -> Patch: | |
| 0 | 75 body = json.loads(j)['patch'] |
| 76 a = ConjunctiveGraph() | |
| 77 a.parse(StringInputSource(json.dumps(body['adds']).encode('utf8')), format='json-ld') | |
| 78 d = ConjunctiveGraph() | |
| 79 d.parse(StringInputSource(json.dumps(body['deletes']).encode('utf8')), format='json-ld') | |
| 80 return Patch(addGraph=a, delGraph=d) | |
| 81 | |
| 4 | 82 |
| 83 def graphAsJson(g: ConjunctiveGraph) -> str: | |
| 0 | 84 # This is not the same as g.serialize(format='json-ld')! That |
| 85 # version omits literal datatypes. | |
| 86 return json.dumps(from_rdf(g)) | |
| 87 | |
| 4 | 88 |
| 0 | 89 _graphsInProcess = itertools.count() |
| 4 | 90 |
| 91 | |
| 0 | 92 class PatchableGraph(GraphEditApi): |
| 93 """ | |
| 94 Master graph that you modify with self.patch, and we get the | |
| 95 updates to all current listeners. | |
| 96 """ | |
| 97 | |
| 4 | 98 def __init__(self, label: Optional[str] = None): |
| 99 self._graph = ConjunctiveGraph() | |
| 100 self._observers: List[Callable[[str], None]] = [] | |
| 101 if label is None: | |
| 102 label = f'patchableGraph{next(_graphsInProcess)}' | |
| 103 self.label = label | |
| 0 | 104 |
| 4 | 105 def serialize(self, *arg, **kw) -> bytes: |
| 106 with SERIALIZE_CALLS.labels(graph=self.label).time(): | |
| 107 return cast(bytes, self._graph.serialize(*arg, **kw)) | |
| 108 | |
| 109 def patch(self, p: Patch): | |
| 3 | 110 with PATCH_CALLS.labels(graph=self.label).time(): |
| 0 | 111 # assuming no stmt is both in p.addQuads and p.delQuads. |
| 112 dels = set([q for q in p.delQuads if inGraph(q, self._graph)]) | |
| 113 adds = set([q for q in p.addQuads if not inGraph(q, self._graph)]) | |
| 114 minimizedP = Patch(addQuads=adds, delQuads=dels) | |
| 115 if minimizedP.isNoop(): | |
| 116 return | |
| 4 | 117 patchQuads(self._graph, deleteQuads=dels, addQuads=adds, perfect=False) # true? |
| 0 | 118 for ob in self._observers: |
| 119 ob(patchAsJson(p)) | |
| 3 | 120 STATEMENT_COUNT.labels(graph=self.label).set(len(self._graph)) |
| 0 | 121 |
| 4 | 122 def asJsonLd(self) -> str: |
| 0 | 123 return graphAsJson(self._graph) |
| 124 | |
| 4 | 125 def addObserver(self, onPatch: Callable[[str], None]): |
| 0 | 126 self._observers.append(onPatch) |
| 3 | 127 OBSERVERS_CURRENT.labels(graph=self.label).set(len(self._observers)) |
| 128 OBSERVERS_ADDED.labels(graph=self.label).inc() | |
| 0 | 129 |
| 4 | 130 def removeObserver(self, onPatch: Callable[[str], None]): |
| 0 | 131 try: |
| 132 self._observers.remove(onPatch) | |
| 133 except ValueError: | |
| 134 pass | |
| 135 self._currentObservers = len(self._observers) | |
| 136 | |
| 4 | 137 def setToGraph(self, newGraph: ConjunctiveGraph): |
| 0 | 138 self.patch(Patch.fromDiff(self._graph, newGraph)) |
| 139 | |
| 3 | 140 |
| 141 SEND_SIMPLE_GRAPH = Summary('send_simple_graph', 'calls to _writeGraphResponse') | |
| 0 | 142 |
| 143 | |
| 144 class CycloneGraphHandler(PrettyErrorHandler, cyclone.web.RequestHandler): | |
| 4 | 145 |
| 0 | 146 def initialize(self, masterGraph: PatchableGraph): |
| 147 self.masterGraph = masterGraph | |
| 148 | |
| 149 def get(self): | |
| 3 | 150 with SEND_SIMPLE_GRAPH.time(): |
| 0 | 151 self._writeGraphResponse() |
| 152 | |
| 153 def _writeGraphResponse(self): | |
| 154 acceptHeader = self.request.headers.get( | |
| 155 'Accept', | |
| 156 # see https://github.com/fiorix/cyclone/issues/20 | |
| 157 self.request.headers.get('accept', '')) | |
| 158 | |
| 159 if acceptHeader == 'application/nquads': | |
| 160 self.set_header('Content-type', 'application/nquads') | |
| 161 self.masterGraph.serialize(self, format='nquads') | |
| 162 elif acceptHeader == 'application/ld+json': | |
| 163 self.set_header('Content-type', 'application/ld+json') | |
| 164 self.masterGraph.serialize(self, format='json-ld', indent=2) | |
| 165 else: | |
| 166 if acceptHeader.startswith('text/html'): | |
| 167 self._writeGraphForBrowser() | |
| 168 return | |
| 169 self.set_header('Content-type', 'application/x-trig') | |
| 170 self.masterGraph.serialize(self, format='trig') | |
| 171 | |
| 172 def _writeGraphForBrowser(self): | |
| 173 # We think this is a browser, so respond with a live graph view | |
| 174 # (todo) | |
| 175 self.set_header('Content-type', 'text/html') | |
| 176 | |
| 177 self.write(b''' | |
| 178 <html><body><pre>''') | |
| 179 | |
| 180 ns = NamespaceManager(self.masterGraph._graph) | |
| 181 # maybe these could be on the PatchableGraph instance | |
| 182 ns.bind('ex', 'http://example.com/') | |
| 183 ns.bind('', 'http://projects.bigasterisk.com/room/') | |
| 184 ns.bind("rdf", "http://www.w3.org/1999/02/22-rdf-syntax-ns#") | |
| 185 ns.bind("xsd", "http://www.w3.org/2001/XMLSchema#") | |
| 186 | |
| 187 for s, p, o, g in sorted(self.masterGraph._graph.quads()): | |
| 188 g = g.identifier | |
| 189 nquadLine = f'{s.n3(ns)} {p.n3(ns)} {o.n3(ns)} {g.n3(ns)} .\n' | |
| 190 self.write(html.escape(nquadLine).encode('utf8')) | |
| 191 | |
| 192 self.write(b''' | |
| 193 </pre> | |
| 194 <p> | |
| 195 <a href="#">[refresh]</a> | |
| 196 <label><input type="checkbox"> Auto-refresh</label> | |
| 197 </p> | |
| 198 <script> | |
| 199 | |
| 200 if (new URL(window.location).searchParams.get('autorefresh') == 'on') { | |
| 201 document.querySelector("input").checked = true; | |
| 202 setTimeout(() => { | |
| 203 requestAnimationFrame(() => { | |
| 204 window.location.replace(window.location.href); | |
| 205 }); | |
| 206 }, 2000); | |
| 207 } | |
| 208 | |
| 209 document.querySelector("a").addEventListener("click", (ev) => { | |
| 210 ev.preventDefault(); | |
| 211 window.location.replace(window.location.href); | |
| 212 | |
| 213 }); | |
| 214 document.querySelector("input").addEventListener("change", (ev) => { | |
| 215 if (document.querySelector("input").checked) { | |
| 216 const u = new URL(window.location); | |
| 217 u.searchParams.set('autorefresh', 'on'); | |
| 218 window.location.replace(u.href); | |
| 219 } else { | |
| 220 const u = new URL(window.location); | |
| 221 u.searchParams.delete('autorefresh'); | |
| 222 window.location.replace(u.href); | |
| 223 } | |
| 224 }); | |
| 225 | |
| 226 </script> | |
| 227 </body></html> | |
| 228 ''') | |
| 229 | |
| 230 | |
| 3 | 231 SEND_FULL_GRAPH = Summary('send_full_graph', 'fullGraph SSE events') |
| 232 SEND_PATCH = Summary('send_patch', 'patch SSE events') | |
| 233 | |
| 234 | |
| 0 | 235 class CycloneGraphEventsHandler(cyclone.sse.SSEHandler): |
| 236 """ | |
| 237 One session with one client. | |
| 238 | |
| 239 returns current graph plus future patches to keep remote version | |
| 240 in sync with ours. | |
| 241 | |
| 242 intsead of turning off buffering all over, it may work for this | |
| 243 response to send 'x-accel-buffering: no', per | |
| 244 http://nginx.org/en/docs/http/ngx_http_proxy_module.html#proxy_buffering | |
| 245 """ | |
| 4 | 246 |
| 0 | 247 def __init__(self, application, request, masterGraph): |
| 248 cyclone.sse.SSEHandler.__init__(self, application, request) | |
| 249 self.masterGraph = masterGraph | |
| 250 | |
| 251 def bind(self): | |
| 3 | 252 with SEND_FULL_GRAPH.time(): |
| 0 | 253 graphJson = self.masterGraph.asJsonLd() |
| 254 log.debug("send fullGraph event: %s", graphJson) | |
| 255 self.sendEvent(message=graphJson, event=b'fullGraph') | |
| 256 self.masterGraph.addObserver(self.onPatch) | |
| 257 | |
| 258 def onPatch(self, patchJson): | |
| 3 | 259 with SEND_PATCH.time(): |
| 0 | 260 # throttle and combine patches here- ideally we could see how |
| 261 # long the latency to the client is to make a better rate choice | |
| 262 self.sendEvent(message=patchJson, event=b'patch') | |
| 263 | |
| 264 def unbind(self): | |
| 265 self.masterGraph.removeObserver(self.onPatch) |