Coverage for src/metador_core/widget/server/__init__.py: 32%

1"""The Metador widget server.""" 

2import io 

3from typing import Dict, List, Literal, Optional, Union 

4 

5import numpy as np 

6from bokeh.application import Application 

7from bokeh.application.handlers.function import FunctionHandler 

8from bokeh.document import Document 

9from bokeh.embed import server_document 

10from bokeh.server.server import Server 

11from flask import Blueprint, request, send_file 

12from panel.viewable import Viewable 

13from tornado.ioloop import IOLoop 

14from werkzeug.exceptions import BadRequest, NotFound 

15 

16from metador_core.container import ContainerProxy, MetadorContainer, MetadorNode 

17 

18 

19class WidgetServer: 

20 """Server backing the instances of Metador widgets (and dashboard). 

21 

22 Metador widgets depend on a `WidgetServer` to: 

23 * get data from Metador containers (via special flask API, provided as a mountable blueprint) 

24 * wire up the information flow with a bokeh server instance (requirement for interactive bokeh widgets) 

25 

26 For information on running a bokeh server see: 

27 https://docs.bokeh.org/en/latest/docs/user_guide/server.html#embedding-bokeh-server-as-a-library 

28 """ 

29 

30 @classmethod 

31 def _get_widget_arg(cls, args: Dict[str, List[bytes]], name: str) -> Optional[str]: 

32 """Extract argument from bokeh server request argument dict.""" 

33 return args[name][0].decode("utf-8") if name in args and args[name] else None 

34 

35 @classmethod 

36 def _get_widget_args(cls, doc: Document): 

37 """Extract arguments from bokeh server request parameters.""" 

38 args = doc.session_context.request.arguments 

39 return dict( 

40 container_id=cls._get_widget_arg(args, "id"), 

41 container_path=cls._get_widget_arg(args, "path"), 

42 ) 

43 

44 @classmethod 

45 def _make_widget_args( 

46 cls, container_id: str, container_path: Optional[str] 

47 ) -> Dict[str, str]: 

48 """Construct dict to be passed through bokeh request into widget.""" 

49 req_args = {"id": container_id} 

50 if container_path: 

51 req_args["path"] = container_path 

52 return req_args 

53 

54 def _get_bokeh_widget_name( 

55 self, 

56 viewable_type: Literal["widget", "dashboard"], 

57 name: str, 

58 ) -> str: 

59 """Return mapped name of a registered widget or dashboard (bokeh server endpoint). 

60 

61 Raises NotFound exception if widget has not been found. 

62 """ 

63 if viewable_type not in {"widget", "dashboard"}: 

64 msg = f"Invalid type: {viewable_type}. Must be widget or dashboard!" 

65 raise NotFound(msg) 

66 known = self._reg_widgets if viewable_type == "widget" else self._reg_dashboards 

67 if name not in known: 

68 raise NotFound(f"Bokeh {viewable_type} not found: '{name}'") 

69 return known[name] 

70 

71 def _get_container_node( 

72 self, container_id: str, container_path: Optional[str] = None 

73 ) -> Optional[Union[MetadorContainer, MetadorNode]]: 

74 """Retrieve desired container (and target path, if provided). 

75 

76 If `path` is provided in the query parameters, 

77 will return the container node, otherwise returns the full container. 

78 

79 Raises NotFound exception if container or path in container do not exist. 

80 """ 

81 try: 

82 container = self._containers.get(container_id) 

83 except KeyError as e: 

84 raise NotFound(f"Container not found: '{container_id}'") from e 

85 

86 if container_path is None: 

87 return container 

88 if node := container.get(container_path): 

89 return node.restrict(read_only=True, local_only=True) 

90 

91 raise NotFound(f"Path not found in container: {container_path}") 

92 

93 # ---- 

94 

95 def __init__( 

96 self, 

97 containers: ContainerProxy[str], 

98 *, 

99 bokeh_endpoint: Optional[str] = None, 

100 flask_endpoint: Optional[str] = None, 

101 populate: bool = True, 

102 ): 

103 """Widget server to serve widget- and dashboard-like bokeh entities. 

104 

105 Args: 

106 containers: `ContainerProxy` to retrieve containers by some container id string. 

107 bokeh_endpoint: Endpoint where the bokeh server will run (`WidgetServer.run()`) 

108 flask_endpoint: Endpoint where Widget API is mounted (`WidgetServer.get_flask_blueprint()`) 

109 populate: If true (default), load and serve all installed widgets and generic dashboard 

110 """ 

111 self._containers = containers 

112 self._bokeh_apps: Dict[str, Application] = {} 

113 self._reg_widgets: Dict[str, str] = {} 

114 self._reg_dashboards: Dict[str, str] = {} 

115 

116 # these can be set after launching the server threads 

117 # (e.g. in case of dynamic port selection) 

118 self._flask_endpoint = flask_endpoint or "" 

119 self._bokeh_endpoint = bokeh_endpoint or "" 

120 

121 if populate: 

122 self.register_installed() 

123 

124 def register_installed(self) -> None: 

125 """Register installed widgets and the generic dashboard.""" 

126 # NOTE: do imports here, otherwise circular imports. 

127 from metador_core.plugins import widgets 

128 

129 from ..dashboard import Dashboard 

130 

131 self.register_dashboard("generic", self.make_bokeh_app(Dashboard)) 

132 for wclass in widgets.values(): 

133 self.register_widget( 

134 wclass.Plugin.plugin_string(), self.make_bokeh_app(wclass) 

135 ) 

136 

137 def register_widget(self, name: str, bokeh_app: Application) -> None: 

138 """Register a new widget application.""" 

139 mapped_name = f"w-{name}" 

140 self._bokeh_apps[f"/{mapped_name}"] = bokeh_app 

141 self._reg_widgets[name] = mapped_name 

142 

143 def register_dashboard(self, name: str, bokeh_app: Application) -> None: 

144 """Register a new dashboard application.""" 

145 mapped_name = f"d-{name}" 

146 self._bokeh_apps[f"/{mapped_name}"] = bokeh_app 

147 self._reg_dashboards[name] = mapped_name 

148 

149 def make_bokeh_app(self, viewable_class: Viewable) -> Application: 

150 def handler(doc: Document) -> None: 

151 """Return bokeh app for Metador widget. 

152 

153 In this context, a suitable class must satisfy the interface 

154 of being initialized with a metador node or container, 

155 and having a `show()` method returning a panel `Viewable`. 

156 

157 The app will understand take `id` and optionally a `path` as query params. 

158 These are parsed and used to look up the correct container (node). 

159 """ 

160 w_args = self._get_widget_args(doc) 

161 if c_obj := self._get_container_node(**w_args): 

162 # if we retrieved container / node, instantiate a widget and show it 

163 widget = viewable_class( 

164 c_obj, server=self, container_id=w_args["container_id"] 

165 ).show() 

166 doc.add_root(widget.get_root(doc)) 

167 

168 return Application(FunctionHandler(handler, trap_exceptions=True)) 

169 

170 @property 

171 def flask_endpoint(self) -> str: 

172 """Get configured endpoint where WidgetServer API is mounted.""" 

173 return self._flask_endpoint 

174 

175 @flask_endpoint.setter 

176 def flask_endpoint(self, uri: str): 

177 """Set URI where the blueprint from `get_flask_blueprint` is mounted.""" 

178 self._flask_endpoint = uri.rstrip("/") 

179 

180 @property 

181 def bokeh_endpoint(self) -> str: 

182 """Get URI where the bokeh server is running.""" 

183 return self._bokeh_endpoint 

184 

185 @bokeh_endpoint.setter 

186 def bokeh_endpoint(self, uri: str): 

187 """Set URI where the bokeh server is running.""" 

188 self._bokeh_endpoint = uri.rstrip("/") 

189 

190 def run(self, **kwargs): 

191 """Run bokeh server with the registered apps (will block the current process).""" 

192 # kwargs["io_loop"] = kwargs.get("io_loop") or IOLoop() 

193 # server = pn.io.server.get_server(self._bokeh_apps, **kwargs) 

194 

195 kwargs["loop"] = kwargs.get("io_loop") or IOLoop() 

196 server = Server(self._bokeh_apps, **kwargs) 

197 

198 server.start() 

199 server.io_loop.start() 

200 

201 # ---- 

202 # Helper functions exposed to widgets 

203 

204 def file_url_for(self, container_id: str, node: MetadorNode) -> str: 

205 """Return URL for given container ID and file at Metador Container node. 

206 

207 To be used by widgets that need direct access to files in the container. 

208 """ 

209 if not self._flask_endpoint: 

210 raise RuntimeError("missing flask endpoint!") 

211 return f"{self._flask_endpoint}/file/{container_id}{node.name}" 

212 

213 # ---- 

214 # Functions making up the WidgetServer API 

215 

216 def index(self): 

217 """Return information about current Metador environment. 

218 

219 Response includes an overview of metador-related Python packages, 

220 Metador plugins, and the known widgets (nodes) and dashboards (containers). 

221 """ 

222 from metador_core.plugin.types import to_ep_name 

223 from metador_core.plugins import plugingroups 

224 

225 # build dict with all available metador plugins 

226 pgs = {to_ep_name(x.name, x.version): x.dict() for x in plugingroups.keys()} 

227 groups = {plugingroups.Plugin.name: pgs} 

228 for pg in plugingroups.values(): 

229 groups[pg.Plugin.name] = { 

230 to_ep_name(x.name, x.version): x.dict() for x in pg.keys() 

231 } 

232 

233 return { 

234 "widgets": list(self._reg_widgets), 

235 "dashboards": list(self._reg_dashboards), 

236 "plugins": groups, 

237 } 

238 

239 def download(self, container_id: str, container_path: str): 

240 """Return file download stream of a file embedded in the container.""" 

241 node = self._get_container_node(container_id, container_path) 

242 # get data out of container 

243 obj = node[()] 

244 bs = obj.tolist() if isinstance(obj, np.void) else obj 

245 if not isinstance(bs, bytes): 

246 raise BadRequest(f"Path not a bytes object: /{container_path}") 

247 

248 # construct a default file name based on path in container 

249 def_name = f"{container_id}_{container_path.replace('/', '__')}" 

250 # if object has attached file metadata, use it to serve data: 

251 filemeta = node.meta.get("core.file") 

252 name = filemeta.id_ if filemeta else def_name 

253 mime = filemeta.encodingFormat if filemeta else None 

254 

255 # requested as explicit file download? 

256 dl = bool(request.args.get("download", False)) 

257 # return file download stream with download metadata 

258 return send_file( 

259 io.BytesIO(bs), download_name=name, mimetype=mime, as_attachment=dl 

260 ) 

261 

262 def get_script( 

263 self, 

264 viewable_type: Literal["widget", "dashboard"], 

265 name: str, 

266 container_id: str, 

267 container_path: Optional[str] = None, 

268 ) -> str: 

269 """Return a script tag that will auto-load the desired widget for selected container.""" 

270 if not self._bokeh_endpoint: 

271 raise RuntimeError("missing bokeh endpoint!") 

272 if viewable_type == "dashboard" and container_path: 

273 raise BadRequest("Dashboards do not accept a container path!") 

274 

275 return server_document( 

276 f"{self._bokeh_endpoint}/{self._get_bokeh_widget_name(viewable_type, name)}", 

277 arguments=self._make_widget_args(container_id, container_path), 

278 ) 

279 

280 def get_flask_blueprint(self, *args): 

281 """Return a Flask blueprint with the Metador container and widget API.""" 

282 api = Blueprint(*args) 

283 

284 api.route("/")(self.index) 

285 api.route("/file/<container_id>/<path:container_path>")(self.download) 

286 api.route("/<viewable_type>/<name>/<container_id>/")( 

287 api.route("/<viewable_type>/<name>/<container_id>/<path:container_path>")( 

288 self.get_script 

289 ) 

290 ) 

291 

292 return api 

293 

294 

295# NOTE: snippet to make script tag not evaluate by default 

296# it can be used to prevent the auto-loading during DOM injection, if needed for some reason 

297# # disable self-evaluation (save call in variable, call when requested) 

298# sc_id = re.match(r"\s*<script id=\"(.*)\">", script).group(1) 

299# script = script.replace("(function", f"w{sc_id} = (function").replace( 

300# "})();", "});" 

301# ) 

302# return f'{script}<button type="button" onclick="w{sc_id}()">Load widget</button>'