Coverage for .tox/coverage/lib/python3.13/site-packages/wuttaweb/views/tables.py: 100%

234 statements  

« prev     ^ index     » next       coverage.py v7.17.0, created at 2026-10-24 16:36 -0500

1# -*- coding: utf-8; -*- 

2################################################################################ 

3# 

4# wuttaweb -- Web App for Wutta Framework 

5# Copyright © 2024-2026 Lance Edgar 

6# 

7# This file is part of Wutta Framework. 

8# 

9# Wutta Framework is free software: you can redistribute it and/or modify it 

10# under the terms of the GNU General Public License as published by the Free 

11# Software Foundation, either version 3 of the License, or (at your option) any 

12# later version. 

13# 

14# Wutta Framework is distributed in the hope that it will be useful, but 

15# WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or 

16# FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for 

17# more details. 

18# 

19# You should have received a copy of the GNU General Public License along with 

20# Wutta Framework. If not, see <http://www.gnu.org/licenses/>. 

21# 

22################################################################################ 

23""" 

24Table Views 

25""" 

26 

27import os 

28import sys 

29 

30from alembic import command as alembic_command 

31from sqlalchemy_utils import get_mapper 

32from mako.lookup import TemplateLookup 

33from webhelpers2.html import HTML 

34 

35from wuttjamaican.db.conf import ( 

36 check_alembic_current, 

37 make_alembic_config, 

38 get_alembic_scriptdir, 

39) 

40 

41from wuttaweb.views import MasterView 

42from wuttaweb.forms.widgets import CopyableTextWidget 

43 

44 

45class AppTableView(MasterView): # pylint: disable=abstract-method 

46 """ 

47 Master view showing all tables in the :term:`app database`. 

48 

49 Default route prefix is ``app_tables``. 

50 

51 Notable URLs provided by this class: 

52 

53 * ``/tables/app/`` 

54 * ``/tables/app/XXX`` 

55 """ 

56 

57 # pylint: disable=duplicate-code 

58 model_name = "app_table" 

59 model_title = "App Table" 

60 model_key = "name" 

61 url_prefix = "/tables/app" 

62 filterable = False 

63 sortable = True 

64 sort_on_backend = False 

65 paginated = True 

66 paginate_on_backend = False 

67 creatable = True 

68 editable = False 

69 deletable = False 

70 # pylint: enable=duplicate-code 

71 

72 labels = { 

73 "name": "Table Name", 

74 "module_name": "Module", 

75 "module_file": "File", 

76 } 

77 

78 grid_columns = [ 

79 "name", 

80 "schema", 

81 # "row_count", 

82 ] 

83 

84 sort_defaults = "name" 

85 

86 form_fields = [ 

87 "name", 

88 "schema", 

89 "model_name", 

90 "description", 

91 # "row_count", 

92 "module_name", 

93 "module_file", 

94 ] 

95 

96 has_rows = True 

97 row_model_title = "Column" 

98 row_model_title_plural = "Columns" 

99 rows_filterable = False 

100 rows_sort_defaults = "sequence" 

101 rows_sort_on_backend = False 

102 rows_paginated = True 

103 rows_paginate_on_backend = False 

104 rows_viewable = False 

105 

106 row_grid_columns = [ 

107 "sequence", 

108 "column_name", 

109 "data_type", 

110 "nullable", 

111 "description", 

112 ] 

113 

114 def normalize_table(self, table): # pylint: disable=missing-function-docstring 

115 record = { 

116 # nb. sometimes need the real table reference later when 

117 # dealing with an instance view 

118 "table": table, 

119 "name": table.name, 

120 "schema": table.schema or "", 

121 # "row_count": 42, 

122 } 

123 

124 try: 

125 cls = get_mapper(table).class_ 

126 except ValueError: 

127 pass 

128 else: 

129 record.update( 

130 { 

131 "model_class": cls, 

132 "model_name": cls.__name__, 

133 "model_name_dotted": f"{cls.__module__}.{cls.__name__}", 

134 "description": (cls.__doc__ or "").strip(), 

135 "module_name": cls.__module__, 

136 "module_file": sys.modules[cls.__module__].__file__, 

137 } 

138 ) 

139 

140 return record 

141 

142 def get_grid_data( # pylint: disable=empty-docstring 

143 self, columns=None, session=None 

144 ): 

145 """ """ 

146 model = self.app.model 

147 data = [] 

148 

149 for table in model.Base.metadata.tables.values(): 

150 data.append(self.normalize_table(table)) 

151 

152 return data 

153 

154 def configure_grid(self, grid): # pylint: disable=empty-docstring 

155 """ """ 

156 g = grid 

157 super().configure_grid(g) 

158 

159 # nb. show more tables by default 

160 g.pagesize = 50 

161 

162 # schema 

163 g.set_searchable("schema") 

164 

165 # name 

166 g.set_searchable("name") 

167 g.set_link("name") 

168 

169 def get_instance( # pylint: disable=empty-docstring,arguments-differ,unused-argument 

170 self, **kwargs 

171 ): 

172 """ """ 

173 if "_cached_instance" not in self.__dict__: 

174 model = self.app.model 

175 name = self.request.matchdict["name"] 

176 table = model.Base.metadata.tables[name] 

177 self.__dict__["_cached_instance"] = self.normalize_table(table) 

178 

179 return self.__dict__["_cached_instance"] 

180 

181 def get_instance_title(self, instance): # pylint: disable=empty-docstring 

182 """ """ 

183 return instance["name"] 

184 

185 def configure_form(self, form): # pylint: disable=empty-docstring 

186 """ """ 

187 f = form 

188 super().configure_form(f) 

189 

190 # description 

191 f.set_widget("description", "notes") 

192 

193 # module_name 

194 f.set_widget("module_name", CopyableTextWidget()) 

195 

196 # module_file 

197 f.set_widget("module_file", CopyableTextWidget()) 

198 

199 def get_xref_buttons(self, obj): 

200 """ 

201 By default this returns a list of buttons for each 

202 :class:`~wuttaweb.views.master.MasterView` subclass registered 

203 in the app for the current table model. Also a button to make 

204 a new Master View class, if permissions allow. 

205 

206 See also parent method docs, 

207 :meth:`~wuttaweb.views.master.MasterView.get_xref_buttons()` 

208 """ 

209 table = obj 

210 buttons = [] 

211 

212 # nb. we do not omit any buttons due to lack of permission 

213 # here. all buttons are shown for anyone seeing this page. 

214 # this is for sake of clarity so admin users are aware of what 

215 # is *possible* within the app etc. 

216 master_views = self.request.registry.settings.get("wuttaweb_master_views", {}) 

217 model_views = master_views.get(table["model_class"], []) 

218 for view in model_views: 

219 buttons.append( 

220 self.make_button( 

221 view.get_model_title_plural(), 

222 primary=True, 

223 url=self.request.route_url(view.get_route_prefix()), 

224 icon_left="eye", 

225 ) 

226 ) 

227 

228 # only add "new master view" button if user has perm 

229 if self.request.has_perm("master_views.create"): 

230 query = None 

231 if table.get("model_name"): 

232 query = {"modelClass": table["model_name"]} 

233 

234 # nb. separate slightly from others 

235 buttons.append(HTML.tag("br")) 

236 buttons.append( 

237 self.make_button( 

238 "New Master View", 

239 url=self.request.route_url("master_views.create", _query=query), 

240 icon_left="plus", 

241 ) 

242 ) 

243 

244 return buttons 

245 

246 def get_row_grid_data(self, obj): # pylint: disable=empty-docstring 

247 """ """ 

248 table = obj 

249 data = [] 

250 for i, column in enumerate(table["table"].columns, 1): 

251 data.append( 

252 { 

253 "column": column, 

254 "sequence": i, 

255 "column_name": column.name, 

256 "data_type": str(repr(column.type)), 

257 "nullable": column.nullable, 

258 "description": (column.doc or "").strip(), 

259 } 

260 ) 

261 return data 

262 

263 def configure_row_grid(self, grid): # pylint: disable=empty-docstring 

264 """ """ 

265 g = grid 

266 super().configure_row_grid(g) 

267 

268 # nb. try not to hide any columns by default 

269 g.pagesize = 100 

270 

271 # sequence 

272 g.set_label("sequence", "Seq.") 

273 

274 # column_name 

275 g.set_searchable("column_name") 

276 

277 # data_type 

278 g.set_searchable("data_type") 

279 

280 # nullable 

281 g.set_renderer("nullable", "boolean") 

282 

283 # description 

284 g.set_searchable("description") 

285 g.set_renderer("description", self.render_column_description) 

286 

287 def render_column_description( # pylint: disable=missing-function-docstring,unused-argument 

288 self, column, field, value 

289 ): 

290 if not value: 

291 return "" 

292 

293 max_length = 100 

294 if len(value) <= max_length: 

295 return value 

296 

297 return HTML.tag("span", title=value, c=f"{value[:max_length]} ...") 

298 

299 def get_template_context(self, context): # pylint: disable=empty-docstring 

300 """ """ 

301 context = super().get_template_context(context) 

302 

303 if self.creating: 

304 model = self.app.model 

305 

306 # existing tables 

307 # TODO: any reason this should check grid data instead of metadata? 

308 unwanted = ["transaction", "transaction_meta"] 

309 context["existing_tables"] = [ 

310 {"name": table} 

311 for table in sorted(model.Base.metadata.tables) 

312 if table not in unwanted and not table.endswith("_version") 

313 ] 

314 

315 # model dir 

316 context["model_dir"] = os.path.dirname(model.__file__) 

317 

318 # everything alembic 

319 self.add_alembic_context(context) 

320 

321 return context 

322 

323 def add_alembic_context(self, context): # pylint: disable=empty-docstring 

324 """ """ 

325 

326 # is alembic current? 

327 context["alembic_is_current"] = check_alembic_current(self.config) 

328 

329 # get alembic api 

330 alembic = make_alembic_config(self.config) 

331 script = get_alembic_scriptdir(self.config, alembic) 

332 

333 # collect branch and location for script heads 

334 branches = set() 

335 found_locations = set() 

336 for rev in script.get_revisions(script.get_heads()): 

337 branches.update(rev.branch_labels) 

338 found_locations.add(os.path.dirname(rev.path)) 

339 

340 # known revision branches 

341 context["alembic_branches"] = sorted(branches) 

342 

343 # default branch may be configured 

344 revise_branch = self.config.get( 

345 f"{self.config.appname}.alembic.default_revise_branch" 

346 ) 

347 context["alembic_revise_branch"] = revise_branch 

348 context["alembic_new_branch"] = None 

349 

350 # known version locations 

351 version_locations = sorted( 

352 self.config.parse_list(alembic.get_main_option("version_locations")) 

353 ) 

354 context["alembic_version_locations"] = version_locations 

355 context["alembic_version_location"] = None 

356 

357 # this next bit only if revise branch not configured 

358 if not revise_branch: 

359 

360 # look for "new" version location for which no revision 

361 # scripts were found; if so choose it for new branch 

362 full_locations = { 

363 self.app.resource_path(verloc): verloc for verloc in version_locations 

364 } 

365 new_locations = [ 

366 verloc for verloc in full_locations if verloc not in found_locations 

367 ] 

368 if new_locations: 

369 verloc = full_locations[new_locations[0]] 

370 context["alembic_version_location"] = verloc 

371 context["alembic_new_branch"] = verloc.split(":")[0].split(".")[0] 

372 

373 def wizard_action(self): # pylint: disable=too-many-return-statements 

374 """ 

375 AJAX view to handle various actions for the "new table" wizard. 

376 """ 

377 data = self.request.json_body 

378 action = data.get("action", "").strip() 

379 try: 

380 

381 # nb. cannot use match/case statement until python 3.10, but this 

382 # project technically still supports python 3.8 

383 if action == "write_model_file": 

384 return self.write_model_file(data) 

385 if action == "check_model": 

386 return self.check_model(data) 

387 if action == "write_revision_script": 

388 return self.write_revision_script(data) 

389 if action == "migrate_db": 

390 return self.migrate_db(data) 

391 if action == "check_table": 

392 return self.check_table(data) 

393 if action == "": 

394 return {"error": "Must specify the action to perform."} 

395 return {"error": f"Unknown action requested: {action}"} 

396 

397 except Exception as err: # pylint: disable=broad-exception-caught 

398 return {"error": f"Unexpected error occurred: {err}"} 

399 

400 def write_model_file(self, data): # pylint: disable=missing-function-docstring 

401 model = self.app.model 

402 path = data["module_file"] 

403 

404 if os.path.exists(path): 

405 if data["overwrite"]: 

406 os.remove(path) 

407 else: 

408 return {"error": "File already exists"} 

409 

410 for column in data["columns"]: 

411 if column["data_type"]["type"] == "_fk_uuid_" and column["relationship"]: 

412 name = column["relationship"] 

413 

414 table = model.Base.metadata.tables[column["data_type"]["reference"]] 

415 mapper = get_mapper(table) 

416 reference_model = mapper.class_.__name__ 

417 

418 column["relationship"] = { 

419 "name": name, 

420 "reference_model": reference_model, 

421 } 

422 

423 # TODO: make templates dir configurable? 

424 templates = [self.app.resource_path("wuttaweb:code-templates")] 

425 table_templates = TemplateLookup(directories=templates) 

426 

427 template = table_templates.get_template("/new-table.mako") 

428 content = template.render(**data) 

429 with open(path, "wt", encoding="utf_8") as f: 

430 f.write(content) 

431 

432 return {} 

433 

434 def check_model(self, data): # pylint: disable=missing-function-docstring 

435 model = self.app.model 

436 model_name = data["model_name"] 

437 

438 if not hasattr(model, model_name): 

439 return { 

440 "problem": "class not found in app model", 

441 "model": model.__name__, 

442 } 

443 

444 return {} 

445 

446 def write_revision_script(self, data): # pylint: disable=missing-function-docstring 

447 alembic_config = make_alembic_config(self.config) 

448 

449 # TODO: this is "mostly" duplicated in AlembicMigrationView.save_create_form() 

450 

451 # kwargs for `alembic revision` command 

452 kw = { 

453 "autogenerate": True, 

454 "message": data["message"], 

455 } 

456 # pylint: disable=duplicate-code 

457 if data["branching_option"] == "new": 

458 kw["head"] = "base" 

459 kw["branch_label"] = data["new_branch"] 

460 kw["version_path"] = self.app.resource_path(data["version_location"]) 

461 else: 

462 assert data["branching_option"] == "revise" 

463 kw["head"] = f"{data['revise_branch']}@head" 

464 

465 # run `alembic revision` 

466 script = alembic_command.revision(alembic_config, **kw) 

467 # pylint: enable=duplicate-code 

468 

469 # maybe save default branch 

470 if data["branching_option"] == "revise" and data["make_default_branch"]: 

471 self.app.save_setting( 

472 self.Session(), 

473 f"{self.config.appname}.alembic.default_revise_branch", 

474 data["revise_branch"], 

475 ) 

476 

477 return {"script": script.path} 

478 

479 def migrate_db( # pylint: disable=missing-function-docstring,unused-argument 

480 self, data 

481 ): 

482 alembic_config = make_alembic_config(self.config) 

483 alembic_command.upgrade(alembic_config, "heads") 

484 return {} 

485 

486 def check_table(self, data): # pylint: disable=missing-function-docstring 

487 model = self.app.model 

488 name = data["name"] 

489 

490 table = model.Base.metadata.tables.get(name) 

491 if table is None: 

492 return {"problem": "table does not exist in app model"} 

493 

494 session = self.Session() 

495 count = session.query(table).count() 

496 

497 route_prefix = self.get_route_prefix() 

498 url = self.request.route_url(f"{route_prefix}.view", name=name) 

499 return {"url": url, "count": count} 

500 

501 @classmethod 

502 def defaults(cls, config): # pylint: disable=empty-docstring 

503 """ """ 

504 cls._apptable_defaults(config) 

505 cls._defaults(config) 

506 

507 # pylint: disable=duplicate-code 

508 @classmethod 

509 def _apptable_defaults(cls, config): 

510 route_prefix = cls.get_route_prefix() 

511 permission_prefix = cls.get_permission_prefix() 

512 model_title_plural = cls.get_model_title_plural() 

513 url_prefix = cls.get_url_prefix() 

514 

515 # fix permission group 

516 config.add_wutta_permission_group( 

517 permission_prefix, model_title_plural, overwrite=False 

518 ) 

519 

520 # wizard actions 

521 config.add_route( 

522 f"{route_prefix}.wizard_action", 

523 f"{url_prefix}/new/wizard-action", 

524 request_method="POST", 

525 ) 

526 config.add_view( 

527 cls, 

528 attr="wizard_action", 

529 route_name=f"{route_prefix}.wizard_action", 

530 renderer="json", 

531 permission=f"{permission_prefix}.create", 

532 ) 

533 

534 # pylint: enable=duplicate-code 

535 

536 

537def defaults(config, **kwargs): # pylint: disable=missing-function-docstring 

538 base = globals() 

539 

540 AppTableView = kwargs.get( # pylint: disable=invalid-name,redefined-outer-name 

541 "AppTableView", base["AppTableView"] 

542 ) 

543 AppTableView.defaults(config) 

544 

545 

546def includeme(config): # pylint: disable=missing-function-docstring 

547 defaults(config)