Mercurial > code > home > repos > rdfdb
annotate rdfdb/service.py @ 130:d195a5f50137
rename isEmpty to match js
| author | drewp@bigasterisk.com |
|---|---|
| date | Mon, 29 May 2023 16:14:17 -0700 |
| parents | a71e4272d808 |
| children | 8fa6a47521d7 |
| rev | line source |
|---|---|
| 105 | 1 import functools |
| 80 | 2 import itertools |
| 3 import logging | |
|
89
1120c6489888
refactor, redo the SyncedGraph class ordering, fix most typing errors
drewp@bigasterisk.com
parents:
87
diff
changeset
|
4 from pathlib import Path |
| 105 | 5 from typing import Dict, Optional, cast |
|
46
3b36b2c8ae65
fix bug that echoed a patch back to KC. https://bigasterisk.com/light9/work/2019/kc-patch-echo-bug.png
Drew Perttula <drewp@bigasterisk.com>
parents:
45
diff
changeset
|
6 |
| 105 | 7 from rdflib import URIRef |
| 8 from starlette.applications import Starlette | |
| 9 from starlette.endpoints import WebSocketEndpoint | |
| 10 from starlette.requests import Request | |
| 11 from starlette.responses import PlainTextResponse, Response | |
| 12 from starlette.routing import Route, WebSocketRoute | |
| 13 from starlette.websockets import WebSocket | |
| 14 from starlette_exporter import PrometheusMiddleware, handle_metrics | |
|
13
c9d1764d64ad
add web server. remove more traces of light9
Drew Perttula <drewp@bigasterisk.com>
parents:
diff
changeset
|
15 |
| 105 | 16 from rdfdb.file_vs_uri import DirUriMap |
| 17 from rdfdb.patch import Patch | |
| 107 | 18 from rdfdb.shared_graph import SharedGraph |
| 52 | 19 |
| 120 | 20 log = logging.getLogger('rdfdb.net') |
|
13
c9d1764d64ad
add web server. remove more traces of light9
Drew Perttula <drewp@bigasterisk.com>
parents:
diff
changeset
|
21 |
|
65
9bc9de580033
big rewrite of rdfdb and syncedgraph to use a single websocket connecton
Drew Perttula <drewp@bigasterisk.com>
parents:
59
diff
changeset
|
22 _wsClientSerial = itertools.count(0) |
|
9bc9de580033
big rewrite of rdfdb and syncedgraph to use a single websocket connecton
Drew Perttula <drewp@bigasterisk.com>
parents:
59
diff
changeset
|
23 |
|
9bc9de580033
big rewrite of rdfdb and syncedgraph to use a single websocket connecton
Drew Perttula <drewp@bigasterisk.com>
parents:
59
diff
changeset
|
24 |
| 105 | 25 class SyncedGraphSocket(WebSocketEndpoint): |
|
65
9bc9de580033
big rewrite of rdfdb and syncedgraph to use a single websocket connecton
Drew Perttula <drewp@bigasterisk.com>
parents:
59
diff
changeset
|
26 """ |
|
9bc9de580033
big rewrite of rdfdb and syncedgraph to use a single websocket connecton
Drew Perttula <drewp@bigasterisk.com>
parents:
59
diff
changeset
|
27 Send patches to the client (starting with a client who has 0 |
|
9bc9de580033
big rewrite of rdfdb and syncedgraph to use a single websocket connecton
Drew Perttula <drewp@bigasterisk.com>
parents:
59
diff
changeset
|
28 statements) to keep it in sync with the graph. |
|
9bc9de580033
big rewrite of rdfdb and syncedgraph to use a single websocket connecton
Drew Perttula <drewp@bigasterisk.com>
parents:
59
diff
changeset
|
29 |
|
9bc9de580033
big rewrite of rdfdb and syncedgraph to use a single websocket connecton
Drew Perttula <drewp@bigasterisk.com>
parents:
59
diff
changeset
|
30 Accept patches from the client, and assume that the client has |
|
9bc9de580033
big rewrite of rdfdb and syncedgraph to use a single websocket connecton
Drew Perttula <drewp@bigasterisk.com>
parents:
59
diff
changeset
|
31 already applied them to its local graph. |
|
9bc9de580033
big rewrite of rdfdb and syncedgraph to use a single websocket connecton
Drew Perttula <drewp@bigasterisk.com>
parents:
59
diff
changeset
|
32 |
|
9bc9de580033
big rewrite of rdfdb and syncedgraph to use a single websocket connecton
Drew Perttula <drewp@bigasterisk.com>
parents:
59
diff
changeset
|
33 Treat a disconnect as 'out of sync'. Either the client thinks it |
|
9bc9de580033
big rewrite of rdfdb and syncedgraph to use a single websocket connecton
Drew Perttula <drewp@bigasterisk.com>
parents:
59
diff
changeset
|
34 is out of sync and wants to start over, or we can't apply a patch |
|
9bc9de580033
big rewrite of rdfdb and syncedgraph to use a single websocket connecton
Drew Perttula <drewp@bigasterisk.com>
parents:
59
diff
changeset
|
35 correctly therefore we disconnect to make the client start over. |
|
9bc9de580033
big rewrite of rdfdb and syncedgraph to use a single websocket connecton
Drew Perttula <drewp@bigasterisk.com>
parents:
59
diff
changeset
|
36 |
|
9bc9de580033
big rewrite of rdfdb and syncedgraph to use a single websocket connecton
Drew Perttula <drewp@bigasterisk.com>
parents:
59
diff
changeset
|
37 This socket may also carry some special messages meant for the |
|
9bc9de580033
big rewrite of rdfdb and syncedgraph to use a single websocket connecton
Drew Perttula <drewp@bigasterisk.com>
parents:
59
diff
changeset
|
38 rdfdb web UI, e.g. about who is connected, etc. |
|
9bc9de580033
big rewrite of rdfdb and syncedgraph to use a single websocket connecton
Drew Perttula <drewp@bigasterisk.com>
parents:
59
diff
changeset
|
39 """ |
| 105 | 40 encoding = "text" |
|
65
9bc9de580033
big rewrite of rdfdb and syncedgraph to use a single websocket connecton
Drew Perttula <drewp@bigasterisk.com>
parents:
59
diff
changeset
|
41 |
| 107 | 42 def __init__(self, db: SharedGraph, *a, **kw): |
| 105 | 43 WebSocketEndpoint.__init__(self, *a, **kw) |
| 44 self.db = db | |
| 45 self.connectionId = f'WS{next(_wsClientSerial)}' # unneeded? | |
| 46 log.info(f'ws connection init {self.connectionId}') | |
|
65
9bc9de580033
big rewrite of rdfdb and syncedgraph to use a single websocket connecton
Drew Perttula <drewp@bigasterisk.com>
parents:
59
diff
changeset
|
47 |
|
9bc9de580033
big rewrite of rdfdb and syncedgraph to use a single websocket connecton
Drew Perttula <drewp@bigasterisk.com>
parents:
59
diff
changeset
|
48 def __repr__(self): |
| 105 | 49 return f'<WebSocket {self.connectionId}>' |
| 50 | |
| 51 async def on_connect(self, websocket: WebSocket): | |
| 52 await websocket.accept() | |
| 53 await websocket.send_json({'connectedAs': self.connectionId}) | |
| 54 log.info(f"new ws client {self.connectionId}") | |
| 55 | |
| 109 | 56 await self.db.addClient(self.connectionId, functools.partial(self._onPatch, websocket)) |
| 105 | 57 |
| 58 async def _onPatch(self, websocket: WebSocket, p: Patch): | |
| 126 | 59 try: |
| 60 await websocket.send_text(p.makeJsonRepr()) | |
| 61 except RuntimeError: | |
| 62 # likely WS disconnect | |
| 63 log.warning("onPatch failed- hope the client calls back") | |
|
65
9bc9de580033
big rewrite of rdfdb and syncedgraph to use a single websocket connecton
Drew Perttula <drewp@bigasterisk.com>
parents:
59
diff
changeset
|
64 |
| 105 | 65 async def on_receive(self, websocket: WebSocket, data: str): |
| 66 if data == 'PING': | |
| 67 await websocket.send_text('PONG') | |
| 68 return | |
| 69 log.debug("%r sends patch to us: %s", self, data[:64]) | |
| 70 p = Patch(jsonRepr=data) | |
|
127
a71e4272d808
cut way down on some no-op patches that were causing file rewrites/rereads
drewp@bigasterisk.com
parents:
126
diff
changeset
|
71 |
|
a71e4272d808
cut way down on some no-op patches that were causing file rewrites/rereads
drewp@bigasterisk.com
parents:
126
diff
changeset
|
72 # this is very important- I caught clients piling up dozens of retries, causing |
|
a71e4272d808
cut way down on some no-op patches that were causing file rewrites/rereads
drewp@bigasterisk.com
parents:
126
diff
changeset
|
73 # us to rewrite files then notice them, etc. Problem may not be fully solved. |
| 130 | 74 if p.isEmpty(): |
|
127
a71e4272d808
cut way down on some no-op patches that were causing file rewrites/rereads
drewp@bigasterisk.com
parents:
126
diff
changeset
|
75 return |
|
a71e4272d808
cut way down on some no-op patches that were causing file rewrites/rereads
drewp@bigasterisk.com
parents:
126
diff
changeset
|
76 |
| 105 | 77 try: |
| 78 await self.db.patch(p, sender=self.connectionId) | |
| 79 except ValueError as e: | |
| 80 log.warning(f'patch from {self!r} did not apply: {e!r}') | |
| 81 # here we should disconnect that client and make them reset | |
| 82 await websocket.close() | |
| 80 | 83 |
| 105 | 84 async def on_disconnect(self, websocket, close_code): |
| 123 | 85 log.info("bye ws client %r: close_code=%s", self.connectionId, close_code) |
| 105 | 86 self.db.clientDisconnected(self.connectionId) |
|
13
c9d1764d64ad
add web server. remove more traces of light9
Drew Perttula <drewp@bigasterisk.com>
parents:
diff
changeset
|
87 |
|
c9d1764d64ad
add web server. remove more traces of light9
Drew Perttula <drewp@bigasterisk.com>
parents:
diff
changeset
|
88 |
| 107 | 89 def get_graph(db: SharedGraph, request: Request) -> Response: |
| 105 | 90 accept = request.headers.get('accept', '') |
| 91 if accept == 'text/plain': | |
| 92 format = 'nt' | |
| 93 elif accept == 'application/n-quads': | |
| 94 format = 'nquads' | |
| 95 else: | |
|
13
c9d1764d64ad
add web server. remove more traces of light9
Drew Perttula <drewp@bigasterisk.com>
parents:
diff
changeset
|
96 format = 'n3' |
| 105 | 97 return PlainTextResponse(db.graph.serialize(format=format)) |
|
13
c9d1764d64ad
add web server. remove more traces of light9
Drew Perttula <drewp@bigasterisk.com>
parents:
diff
changeset
|
98 |
| 45 | 99 |
| 107 | 100 async def post_prefixes(db: SharedGraph, request: Request) -> Response: |
| 105 | 101 suggestion = await request.json() |
| 102 db.addPrefixes( | |
| 103 ctx=URIRef(suggestion['ctx']), # | |
| 104 prefixes=dict((cast(str, k), URIRef(v)) for k, v in suggestion['prefixes'].items())) | |
| 105 return Response(status_code=204) | |
| 45 | 106 |
| 107 | |
| 105 | 108 def makeApp(dirUriMap: Optional[DirUriMap] = None, prefixes: Optional[Dict[str, URIRef]] = None) -> Starlette: |
| 109 log.info('makeApp start') | |
|
13
c9d1764d64ad
add web server. remove more traces of light9
Drew Perttula <drewp@bigasterisk.com>
parents:
diff
changeset
|
110 |
|
c9d1764d64ad
add web server. remove more traces of light9
Drew Perttula <drewp@bigasterisk.com>
parents:
diff
changeset
|
111 if dirUriMap is None: |
|
89
1120c6489888
refactor, redo the SyncedGraph class ordering, fix most typing errors
drewp@bigasterisk.com
parents:
87
diff
changeset
|
112 dirUriMap = {Path('data/'): URIRef('http://example.com/data/')} |
|
13
c9d1764d64ad
add web server. remove more traces of light9
Drew Perttula <drewp@bigasterisk.com>
parents:
diff
changeset
|
113 if prefixes is None: |
|
c9d1764d64ad
add web server. remove more traces of light9
Drew Perttula <drewp@bigasterisk.com>
parents:
diff
changeset
|
114 prefixes = { |
| 42 | 115 'rdf': URIRef('http://www.w3.org/1999/02/22-rdf-syntax-ns#'), |
| 116 'rdfs': URIRef('http://www.w3.org/2000/01/rdf-schema#'), | |
| 117 'xsd': URIRef('http://www.w3.org/2001/XMLSchema#'), | |
|
13
c9d1764d64ad
add web server. remove more traces of light9
Drew Perttula <drewp@bigasterisk.com>
parents:
diff
changeset
|
118 } |
| 45 | 119 |
| 107 | 120 db = SharedGraph(dirUriMap=dirUriMap, prefixes=prefixes) |
|
13
c9d1764d64ad
add web server. remove more traces of light9
Drew Perttula <drewp@bigasterisk.com>
parents:
diff
changeset
|
121 |
| 105 | 122 app = Starlette( |
| 123 debug=True, | |
| 124 routes=[ | |
| 125 Route('/graph', functools.partial(get_graph, db)), | |
| 126 WebSocketRoute('/syncedGraph', functools.partial(SyncedGraphSocket, db)), | |
| 127 Route('/prefixes', functools.partial(post_prefixes, db)), | |
| 128 ], | |
| 129 ) | |
| 86 | 130 |
| 105 | 131 app.add_middleware(PrometheusMiddleware) |
| 132 app.add_route("/metrics", handle_metrics) | |
| 86 | 133 |
| 105 | 134 return app |