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

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

2from __future__ import annotations 

3 

4import io 

5from typing import TYPE_CHECKING, Dict, List, Literal, Optional, Union 

6 

7import numpy as np 

8import panel as pn 

9from bokeh.application import Application 

10from bokeh.application.handlers.function import FunctionHandler 

11from bokeh.document import Document 

12from bokeh.embed import server_document 

13from bokeh.server.server import Server 

14from flask import Blueprint, request, send_file 

15from tornado.ioloop import IOLoop 

16from werkzeug.exceptions import BadRequest, NotFound 

17 

18from metador_core.container import ContainerProxy, MetadorContainer, MetadorNode 

19 

20if TYPE_CHECKING: 

21 from panel.viewable import Viewable 

22 

23 

24class WidgetServer: 

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

26 

27 Metador widgets depend on a `WidgetServer` to: 

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

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

30 

31 For information on running a bokeh server see: 

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

33 """ 

34 

35 @classmethod 

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

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

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

39 

40 @classmethod 

41 def _get_widget_args(cls, doc: Document): 

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

43 args = doc.session_context.request.arguments 

44 return dict( 

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

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

47 ) 

48 

49 @classmethod 

50 def _make_widget_args( 

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

52 ) -> Dict[str, str]: 

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

54 req_args = {"id": container_id} 

55 if container_path: 

56 req_args["path"] = container_path 

57 return req_args 

58 

59 def _get_bokeh_widget_name( 

60 self, 

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

62 name: str, 

63 ) -> str: 

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

65 

66 Raises NotFound exception if widget has not been found. 

67 """ 

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

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

70 raise NotFound(msg) 

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

72 if name not in known: 

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

74 return known[name] 

75 

76 def _get_container_node( 

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

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

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

80 

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

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

83 

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

85 """ 

86 try: 

87 container = self._containers.get(container_id) 

88 except KeyError as e: 

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

90 

91 if container_path is None: 

92 return container 

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

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

95 

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

97 

98 # ---- 

99 

100 def __init__( 

101 self, 

102 containers: ContainerProxy[str], 

103 *, 

104 bokeh_endpoint: Optional[str] = None, 

105 flask_endpoint: Optional[str] = None, 

106 populate: bool = True, 

107 ): 

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

109 

110 Args: 

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

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

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

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

115 """ 

116 self._containers = containers 

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

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

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

120 

121 # these can be set after launching the server threads 

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

123 self._flask_endpoint = flask_endpoint or "" 

124 self._bokeh_endpoint = bokeh_endpoint or "" 

125 

126 if populate: 

127 self.register_installed() 

128 

129 def register_installed(self) -> None: 

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

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

132 from metador_core.plugins import widgets 

133 

134 from ..dashboard import Dashboard 

135 

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

137 for wclass in widgets.values(): 

138 self.register_widget( 

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

140 ) 

141 

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

143 """Register a new widget application.""" 

144 mapped_name = f"w-{name}" 

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

146 self._reg_widgets[name] = mapped_name 

147 

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

149 """Register a new dashboard application.""" 

150 mapped_name = f"d-{name}" 

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

152 self._reg_dashboards[name] = mapped_name 

153 

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

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

156 """Return bokeh app for Metador widget. 

157 

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

159 of being initialized with a metador node or container, 

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

161 

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

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

164 """ 

165 w_args = self._get_widget_args(doc) 

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

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

168 widget = viewable_class( 

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

170 ).show() 

171 doc.add_root(widget.get_root(doc)) 

172 

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

174 

175 @property 

176 def flask_endpoint(self) -> str: 

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

178 return self._flask_endpoint 

179 

180 @flask_endpoint.setter 

181 def flask_endpoint(self, uri: str): 

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

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

184 

185 @property 

186 def bokeh_endpoint(self) -> str: 

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

188 return self._bokeh_endpoint 

189 

190 @bokeh_endpoint.setter 

191 def bokeh_endpoint(self, uri: str): 

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

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

194 

195 def run(self, **kwargs): 

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

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

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

199 

200 # NOTE: this loads unused extensions (e.g. ace) that are not even listed?! 

201 # pn.extension(inline=True) 

202 # This seems to work ok: 

203 pn.config.inline = True 

204 

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

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

207 

208 server.start() 

209 server.io_loop.start() 

210 

211 # ---- 

212 # Helper functions exposed to widgets 

213 

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

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

216 

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

218 """ 

219 if not self._flask_endpoint: 

220 raise RuntimeError("missing flask endpoint!") 

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

222 

223 # ---- 

224 # Functions making up the WidgetServer API 

225 

226 def index(self): 

227 """Return information about current Metador environment. 

228 

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

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

231 """ 

232 from metador_core.plugin.types import to_ep_name 

233 from metador_core.plugins import plugingroups 

234 

235 # build dict with all available metador plugins 

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

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

238 for pg in plugingroups.values(): 

239 groups[pg.Plugin.name] = { 

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

241 } 

242 

243 return { 

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

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

246 "plugins": groups, 

247 } 

248 

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

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

251 node = self._get_container_node(container_id, container_path) 

252 # get data out of container 

253 obj = node[()] 

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

255 if not isinstance(bs, bytes): 

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

257 

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

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

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

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

262 name = filemeta.id_ if filemeta else def_name 

263 mime = filemeta.encodingFormat if filemeta else None 

264 

265 # requested as explicit file download? 

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

267 # return file download stream with download metadata 

268 return send_file( 

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

270 ) 

271 

272 def get_script( 

273 self, 

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

275 name: str, 

276 container_id: str, 

277 container_path: Optional[str] = None, 

278 ) -> str: 

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

280 if not self._bokeh_endpoint: 

281 raise RuntimeError("missing bokeh endpoint!") 

282 if viewable_type == "dashboard" and container_path: 

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

284 

285 return server_document( 

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

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

288 ) 

289 

290 def get_flask_blueprint(self, *args): 

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

292 api = Blueprint(*args) 

293 

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

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

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

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

298 self.get_script 

299 ) 

300 ) 

301 

302 return api 

303 

304 

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

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

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

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

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

310# "})();", "});" 

311# ) 

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