Coverage for /home/ubuntu/hidebound/python/hidebound/server/app.py: 42%

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

2from dash import dash_table # noqa F401 

3 

4import os 

5 

6from dash import dcc 

7from dash.dependencies import Input, Output, State 

8from dash.exceptions import PreventUpdate 

9import dash 

10import flask 

11from flask import current_app 

12 

13import hidebound.core.tools as hbt 

14import hidebound.server.components as components 

15import hidebound.server.extensions as ext 

16import hidebound.server.server_tools as hst 

17 

18 

19TESTING = hbt.str_to_bool(os.environ.get('HIDEBOUND_TESTING', 'True')) # pragma: no cover 

20 

21# setup directories in /tmp/mnt 

22if hbt.str_to_bool(os.environ.get('_HIDEBOUND_TEST_DIRS', 'False')): 

23 hst.setup_hidebound_directories('/tmp/mnt') # pragma: no cover 

24# ------------------------------------------------------------------------------ 

25 

26 

27''' 

28Hidebound service used for displaying and interacting with Hidebound database. 

29''' 

30 

31 

32EP = hst.EndPoints() 

33 

34 

35# TODO: implement this 

36def liveness(): 

37 # type: () -> None 

38 '''Liveness probe for kubernetes.''' 

39 pass 

40 

41 

42# TODO: implement this 

43def readiness(): 

44 # type: () -> None 

45 ''' 

46 Readiness probe for kubernetes. 

47 ''' 

48 pass 

49 

50 

51def get_app(testing=False): 

52 # type: (bool) -> dash.Dash 

53 ''' 

54 Creates a Hidebound app. 

55 

56 Returns: 

57 Dash: Dash app. 

58 ''' 

59 app = flask.Flask('hidebound') # type: Union[flask.Flask, dash.Dash] 

60 app.config['TESTING'] = testing 

61 app.config['HEALTHZ'] = dict( 

62 live=liveness, 

63 ready=readiness, 

64 ) 

65 

66 ext.swagger.init_app(app) 

67 ext.hidebound.init_app(app) 

68 ext.healthz.init_app(app) 

69 

70 app = components.get_dash_app(app, seconds=0.8) 

71 return app 

72 

73 

74APP = get_app(testing=TESTING) 

75SERVER = APP.server 

76 

77 

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

79def serve_stylesheet(stylesheet): 

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

81 ''' 

82 Serve stylesheet to app. 

83 

84 Args: 

85 stylesheet (str): stylesheet filename. 

86 

87 Returns: 

88 flask.Response: Response. 

89 ''' 

90 params = dict( 

91 COLOR_SCHEME=components.COLOR_SCHEME, 

92 FONT_FAMILY=components.FONT_FAMILY, 

93 ) 

94 content = hst.render_template(stylesheet + '.j2', params) 

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

96 

97 

98# EVENTS------------------------------------------------------------------------ 

99# TODO: Find a way to test events. 

100@APP.callback( 

101 output=Output('store', 'data'), 

102 inputs=[ 

103 Input('workflow-button', 'n_clicks'), 

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

105 Input('create-button', 'n_clicks'), 

106 Input('export-button', 'n_clicks'), 

107 Input('delete-button', 'n_clicks'), 

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

109 Input('dropdown', 'value'), 

110 Input('query', 'value'), 

111 Input('query', 'n_submit'), 

112 ], 

113 state=[State('store', 'data')], 

114 prevent_initial_call=True, 

115) 

116def on_event(*inputs): 

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

118 ''' 

119 Update Hidebound database instance, and updates store with input data. 

120 

121 Args: 

122 inputs (tuple): Input elements. 

123 

124 Returns: 

125 dict: Store data. 

126 ''' 

127 APP.logger.debug(f'on_event called with inputs: {str(inputs)[:50]}') 

128 hb = current_app.extensions['hidebound'] 

129 

130 # get context values 

131 context = dash.callback_context 

132 store = context.states['store.data'] or {} # type: dict 

133 trigger = context.triggered_id 

134 query = context.inputs['query.value'] 

135 group_by_asset = context.inputs['dropdown.value'] == 'asset' 

136 

137 initial_query = context.inputs['query.n_submit'] == 0 

138 ready = getattr(APP, 'ready', False) 

139 if initial_query and not ready: 

140 APP.ready = True 

141 raise PreventUpdate 

142 

143 if trigger == 'workflow-button': 

144 hst.request(store, EP.workflow, dict(steps=hb.config['workflow'])) 

145 store = hst.search(store, query, group_by_asset) 

146 

147 elif trigger == 'update-button': 

148 hst.request(store, EP.update) 

149 store = hst.search(store, query, group_by_asset) 

150 

151 elif trigger == 'create-button': 

152 hst.request(store, EP.create) 

153 

154 elif trigger == 'export-button': 

155 hst.request(store, EP.export) 

156 

157 elif trigger == 'delete-button': 

158 hst.request(store, EP.delete) 

159 

160 elif trigger in ['search-button', 'query']: 

161 if store.get('ready', False): 

162 store = hst.search(store, query, group_by_asset) 

163 

164 return store 

165 

166 

167@APP.callback( 

168 Output('table-content', 'children'), 

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

170) 

171def on_datatable_update(store): 

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

173 ''' 

174 Updates datatable with read information from store. 

175 

176 Args: 

177 store (dict): Store data. 

178 

179 Returns: 

180 DataTable: Dash DataTable. 

181 ''' 

182 APP.logger.debug( 

183 f'on_datatable_update called with store: {str(store)[:50]}' 

184 ) 

185 

186 if store in [{}, None]: 

187 raise PreventUpdate 

188 data = store.get('content', None) 

189 if data is None: 

190 raise PreventUpdate 

191 

192 if 'error' in data.keys(): 

193 return components.get_key_value_card(data, header='error', id_='error') 

194 return components.get_datatable(data['response']) 

195 

196 

197@APP.callback( 

198 Output('content', 'children'), 

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

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

201) 

202def on_get_tab(tab, store): 

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

204 ''' 

205 Serve content for app tabs. 

206 

207 Args: 

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

209 store (dict): Store. 

210 

211 Returns: 

212 flask.Response: Response. 

213 ''' 

214 hb = current_app.extensions['hidebound'] 

215 

216 APP.logger.debug( 

217 f'on_get_tab called with tab: {tab} and store: {str(store)[:50]}' 

218 ) 

219 store = store or {} 

220 

221 if tab == 'data': 

222 query = store.get('query', None) 

223 return components.get_data_tab(query) 

224 

225 elif tab == 'graph': 

226 data = store.get('content', None) 

227 if data is None: 

228 return None 

229 

230 if 'error' in data.keys(): 

231 return components.get_key_value_card( 

232 data, header='error', id_='error' 

233 ) 

234 graph = data['response'] 

235 if len(graph) == 0: 

236 return None 

237 return components.get_asset_graph(graph) 

238 

239 elif tab == 'config': 

240 config = hst.format_config( 

241 hb.config, 

242 redact_regex=hb.config['redact_regex'], 

243 redact_hash=hb.config['redact_hash'], 

244 ) 

245 return components.get_config_tab(config) 

246 

247 elif tab == 'api': 

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

249 

250 elif tab == 'docs': 

251 return dcc.Location( 

252 id='docs', 

253 href='https://thenewflesh.github.io/hidebound' 

254 ) 

255 

256 

257@APP.callback( 

258 Output('progressbar-container', 'children'), 

259 [Input('clock', 'n_intervals')], 

260) 

261def on_progress(timestamp): 

262 # type: (int) -> flask.Response 

263 ''' 

264 Updates progressbar. 

265 

266 Args: 

267 timestamp (int): Store modification timestamp. 

268 

269 Returns: 

270 flask.Response: Response. 

271 ''' 

272 return components.get_progressbar(hst.get_progress()) 

273# ------------------------------------------------------------------------------ 

274 

275 

276if __name__ == '__main__': 

277 debug = os.environ.get('HIDEBOUND_TESTING', False) 

278 APP.run_server(debug=debug, host=EP.host, port=EP.port)