Coverage for /home/ubuntu/shekels/python/shekels/server/app.py: 100%

1from typing import Any, Dict, List, Tuple, Union # noqa: F401 

2 

3from copy import copy 

4from copy import deepcopy 

5from pathlib import Path 

6import os 

7 

8from dash.dependencies import Input, Output, State 

9from flask_caching import Cache 

10import dash 

11from dash import dash_table 

12from dash import dcc 

13from dash import html 

14from flask_healthz import healthz, HealthError 

15import flasgger as swg 

16import flask 

17import flask_monitoringdashboard as fmdb 

18import jsoncomment as jsonc 

19 

20from shekels.server.api import API 

21import shekels.core.config as cfg 

22import shekels.server.components as svc 

23import shekels.server.event_listener as sev 

24import shekels.server.server_tools as svt 

25# ------------------------------------------------------------------------------ 

26 

27 

28''' 

29Shekels app used for displaying and interacting with database. 

30''' 

31 

32 

33def liveness(): 

34 # type: () -> None 

35 '''Liveness probe for kubernetes.''' 

36 pass 

37 

38 

39def readiness(): 

40 # type: () -> None 

41 ''' 

42 Readiness probe for kubernetes. 

43 

44 Raises: 

45 HealthError: If api is not availiable. 

46 ''' 

47 if not hasattr(APP, 'api'): 

48 raise HealthError('App is missing API.') 

49 

50 

51def get_app(): 

52 # type: () -> dash.Dash 

53 ''' 

54 Creates a Shekels app. 

55 

56 Returns: 

57 Dash: Dash app. 

58 ''' 

59 flask_app = flask.Flask('Shekels') 

60 swg.Swagger(flask_app) 

61 flask_app.register_blueprint(API) 

62 

63 # healthz endpoints 

64 flask_app.register_blueprint(healthz, url_prefix="/healthz") 

65 flask_app.config.update(HEALTHZ={ 

66 "live": liveness, 

67 "ready": readiness, 

68 }) 

69 

70 # flask monitoring 

71 fmdb.config.link = 'monitor' 

72 fmdb.config.monitor_level = 3 

73 fmdb.config.git = 'https://theNewFlesh.github.io/shekels/' 

74 fmdb.config.brand_name = 'Shekels Monitoring Dashboard' 

75 fmdb.config.title_name = 'Shekels Monitoring Dashboard' 

76 fmdb.config.description = 'Monitor Shekels App Performance' 

77 fmdb.config.enable_logging = False 

78 fmdb.config.show_login_banner = True 

79 fmdb.config.show_login_footer = True 

80 fmdb.config.database_name = 'sqlite:////tmp/shekels_monitor.db' 

81 fmdb.bind(flask_app) 

82 

83 app = svc.get_dash_app(flask_app) 

84 app.api = API 

85 app.client = flask_app.test_client() 

86 app.cache = Cache(flask_app, config={'CACHE_TYPE': 'SimpleCache'}) 

87 

88 # register event listener 

89 app.event_listener = sev.EventListener(app, {}) \ 

90 .listen('config-query', svt.config_query_event) \ 

91 .listen('config-search-button', svt.config_query_event) \ 

92 .listen('config-table', svt.config_edit_event) \ 

93 .listen('query', svt.data_query_event) \ 

94 .listen('search-button', svt.data_query_event) \ 

95 .listen('init-button', svt.init_event) \ 

96 .listen('update-button', svt.update_event) \ 

97 .listen('upload', svt.upload_event) \ 

98 .listen('save-button', svt.save_event) 

99 return app 

100 

101 

102APP = get_app() 

103 

104 

105@APP.server.route('/static/<stylesheet>') 

106def serve_stylesheet(stylesheet): 

107 # type: (str) -> flask.Response 

108 ''' 

109 Serve stylesheet to app. 

110 

111 Args: 

112 stylesheet (str): stylesheet filename. 

113 

114 Returns: 

115 flask.Response: Response. 

116 ''' 

117 temp = APP.api.config or {} 

118 color_scheme = copy(cfg.COLOR_SCHEME) 

119 cs = temp.get('color_scheme', {}) 

120 color_scheme.update(cs) 

121 

122 params = dict( 

123 COLOR_SCHEME=color_scheme, 

124 FONT_FAMILY=temp.get('font_family', cfg.Config.font_family.default), 

125 ) 

126 content = svt.render_template('style.css.j2', params) 

127 return flask.Response(content, mimetype='text/css') 

128 

129 

130# EVENTS------------------------------------------------------------------------ 

131@APP.callback( 

132 Output('store', 'data'), 

133 [ 

134 Input('config-query', 'value'), 

135 Input('config-search-button', 'n_clicks'), 

136 Input('config-table', 'data'), 

137 Input('query', 'value'), 

138 Input('init-button', 'n_clicks'), 

139 Input('update-button', 'n_clicks'), 

140 Input('search-button', 'n_clicks'), 

141 Input('upload', 'contents'), 

142 Input('save-button', 'n_clicks'), 

143 ], 

144 [State('config-table', 'data_previous')] 

145) 

146def on_event(*inputs): 

147 # type: (Tuple[Any, ...]) -> Dict[str, Any] 

148 ''' 

149 Update database instance, and updates store with input data. 

150 

151 Args: 

152 inputs (tuple): Input elements. 

153 

154 Returns: 

155 dict: Store data. 

156 ''' 

157 event = dash.callback_context.triggered[0]['prop_id'].split('.')[0] 

158 value = dash.callback_context.triggered[0]['value'] 

159 store = APP.event_listener.store 

160 

161 if event == 'config-table': 

162 value = dict(new=value[0], old=inputs[-1][0]) 

163 

164 if event == 'update-button' and store.get('/api/initialize') is None: 

165 APP.event_listener.emit('init-button', None) 

166 store = APP.event_listener.emit(event, value).store 

167 return store 

168 

169 

170@APP.callback( 

171 Output('plots-content', 'children'), 

172 [Input('store', 'data')] 

173) 

174@APP.cache.memoize(100) 

175def on_plots_update(store): 

176 # type: (Dict) -> dash_table.DataTable 

177 ''' 

178 Updates plots with read information from store. 

179 

180 Args: 

181 store (dict): Store data. 

182 

183 Returns: 

184 list[dcc.Graph]: Plots. 

185 ''' 

186 comp = svt.solve_component_state(store) 

187 if comp is not None: 

188 return comp 

189 config = store.get('/config', deepcopy(APP.api.config)) 

190 plots = config.get('plots', []) 

191 return svc.get_plots(store['/api/search']['response'], plots) 

192 

193 

194@APP.callback( 

195 Output('data-content', 'children'), 

196 [Input('store', 'data')] 

197) 

198@APP.cache.memoize(100) 

199def on_datatable_update(store): 

200 # type: (Dict) -> dash_table.DataTable 

201 ''' 

202 Updates datatable with read information from store. 

203 

204 Args: 

205 store (dict): Store data. 

206 

207 Returns: 

208 DataTable: Dash DataTable. 

209 ''' 

210 comp = svt.solve_component_state(store) 

211 if comp is not None: 

212 return comp 

213 return svc.get_datatable(store['/api/search']['response']) 

214 

215 

216@APP.callback( 

217 Output('config-content', 'children'), 

218 [Input('store', 'data')] 

219) 

220@APP.cache.memoize(100) 

221def on_config_update(store): 

222 # type: (Dict[str, Any]) -> List[flask.Response] 

223 ''' 

224 Updates config table with config information from store. 

225 

226 Args: 

227 store (dict): Store data. 

228 

229 Returns: 

230 flask.Response: Response. 

231 ''' 

232 config = store.get('/config', deepcopy(APP.api.config)) 

233 store['/config'] = config 

234 store['/config/search'] = store.get('/config/search', store['/config']) 

235 comp = svt.solve_component_state(store, config=True) 

236 if comp is not None: 

237 return [ 

238 comp, 

239 html.Div(className='dummy', children=[ 

240 dash_table.DataTable(id='config-table') 

241 ]) 

242 ] 

243 return svc.get_key_value_table( 

244 store['/config/search'], 

245 id_='config', 

246 header='config', 

247 editable=True, 

248 ) 

249 

250 

251@APP.callback( 

252 Output('content', 'children'), 

253 [Input('tabs', 'value')], 

254 [State('store', 'data')] 

255) 

256def on_get_tab(tab, store): 

257 # type: (str, Dict) -> Union[flask.Response, List, None] 

258 ''' 

259 Serve content for app tabs. 

260 

261 Args: 

262 tab (str): Name of tab to render. 

263 store (dict): Store. 

264 

265 Returns: 

266 flask.Response: Response. 

267 ''' 

268 store = store or {} 

269 

270 if tab == 'plots': 

271 query = store.get('/api/search/query', APP.api.config['default_query']) 

272 return svc.get_plots_tab(query) 

273 

274 elif tab == 'data': 

275 query = store.get('/api/search/query', APP.api.config['default_query']) 

276 return svc.get_data_tab(query) 

277 

278 elif tab == 'config': 

279 config = store.get('/config', deepcopy(APP.api.config)) 

280 return svc.get_config_tab(config) 

281 

282 elif tab == 'api': # pragma: no cover 

283 return dcc.Location(id='api', pathname='/api') 

284 

285 elif tab == 'docs': # pragma: no cover 

286 return dcc.Location( 

287 id='docs', 

288 href='https://theNewFlesh.github.io/shekels/' 

289 ) 

290 

291 elif tab == 'monitor': # pragma: no cover 

292 return dcc.Location(id='monitor', pathname='/monitor') 

293# ------------------------------------------------------------------------------ 

294 

295 

296def run(app, config_path, debug=False, test=False): 

297 ''' 

298 Runs a given Shekels app. 

299 

300 Args: 

301 Dash: Shekels app. 

302 config_path (str or Path): Path to configuration JSON. 

303 debug (bool, optional): Whether debug mode is turned on. Default: False. 

304 test (bool, optional): Calls app.run_server if False. Default: False. 

305 ''' 

306 config_path = Path(config_path).as_posix() 

307 with open(config_path) as f: 

308 config = jsonc.JsonComment().load(f) 

309 app.api.config = config 

310 app.api.config_path = config_path 

311 app.event_listener.state.clear() 

312 app.event_listener.state.append({}) 

313 if not test: 

314 app.run_server(debug=debug, host='0.0.0.0', port=8080) # pragma: no cover 

315 

316 

317if __name__ == '__main__': # pragma: no cover 

318 debug = 'DEBUG_MODE' in os.environ.keys() 

319 

320 config_path = '/mnt/storage/shekels_config.json' 

321 if debug: 

322 config_path = '/mnt/storage/test_config.json' 

323 run(APP, config_path, debug=debug)