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

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

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

4# wuttaweb -- Web App for Wutta Framework

7# This file is part of Wutta Framework.

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"""

24Views for Alembic

25"""

27import datetime

28import logging

29import os

30import re

32from alembic import command as alembic_command

33from alembic.migration import MigrationContext

34from alembic.util import CommandError

36from wuttjamaican.db.conf import (

37 make_alembic_config,

38 get_alembic_scriptdir,

39 check_alembic_current,

40)

42import colander

43from webhelpers2.html import tags, HTML

45from wuttaweb.views import View, MasterView

46from wuttaweb.forms import widgets

47from wuttaweb.forms.schema import WuttaDateTime

49log = logging.getLogger(__name__)

52def normalize_revision(

53 config, rev, json_safe=False

54): # pylint: disable=missing-function-docstring

55 app = config.get_app()

57 created = None

58 if match := re.search(r"Create Date: (\d{4}-\d{2}-\d{2}[\d:\. ]+\d)", rev.longdoc):

59 created = datetime.datetime.fromisoformat(match.group(1))

60 created = app.localtime(created, from_utc=False)

61 if json_safe:

62 created = app.render_datetime(created)

64 return {

65 "revision": rev.revision,

66 "branch_labels": ", ".join(rev.branch_labels),

67 "doc": rev.doc,

68 "longdoc": rev.longdoc,

69 "path": rev.path,

70 "dependencies": rev.dependencies or "",

71 "down_revision": rev.down_revision or "",

72 "nextrev": ", ".join(rev.nextrev),

73 "is_base": app.render_boolean(rev.is_base),

74 "is_branch_point": app.render_boolean(rev.is_branch_point),

75 "is_head": rev.is_head,

76 "created": created,

77 }

80class AlembicDashboardView(View):

81 """

82 Custom views for the Alembic Dashboard.

83 """

85 def dashboard(self):

86 """

87 Main view for the Alembic Dashboard.

89 Route name is ``alembic.dashboard``; URL is

90 ``/alembic/dashboard``

91 """

92 script = get_alembic_scriptdir(self.config)

93 with self.config.appdb_engine.begin() as conn:

94 context = MigrationContext.configure(conn)

95 current_heads = context.get_current_heads()

97 def normalize(rev):

98 normal = normalize_revision(self.config, rev, json_safe=True)

99 normal["is_current"] = rev.revision in current_heads

100

101 normal["revision"] = tags.link_to(

102 normal["revision"],

103 self.request.route_url(

104 "alembic.migrations.view", revision=normal["revision"]

105 ),

106 )

107

108 if normal["down_revision"]:

109 normal["down_revision"] = tags.link_to(

110 normal["down_revision"],

111 self.request.route_url(

112 "alembic.migrations.view", revision=normal["down_revision"]

113 ),

114 )

115

116 return normal

117

118 script_heads = []

119 for head in script.get_heads():

120 rev = script.get_revision(head)

121 script_heads.append(normalize(rev))

122

123 db_heads = []

124 for head in current_heads:

125 rev = script.get_revision(head)

126 db_heads.append(normalize(rev))

127

128 script_heads.sort(key=lambda rev: rev["branch_labels"])

129 db_heads.sort(key=lambda rev: rev["branch_labels"])

130

131 return {

132 "index_title": "Alembic Dashboard",

133 "script": {

134 "dir": script.dir,

135 "version_locations": sorted(script.version_locations),

136 "env_py_location": script.env_py_location,

137 "file_template": script.file_template,

138 },

139 "script_heads": script_heads,

140 "db_heads": db_heads,

141 }

142

143 def migrate(self):

144 """

145 Action view to migrate the database. POST request must be used.

146

147 This directly invokes the :func:`alembic upgrade

148 <alembic:alembic.command.upgrade>` (or :func:`alembic

149 downgrade <alembic:alembic.command.downgrade>`) command.

150

151 It then sets a flash message per the command status, and

152 redirects user back to the Dashboard (or other referrer).

153

154 The request must specify a ``revspec`` param, which we pass

155 along as-is to the ``alembic`` command. We assume ``alembic

156 upgrade`` unless the request sets ``direction`` param to

157 ``"downgrade"``.

158 """

159 referrer = self.request.get_referrer(

160 default=self.request.route_url("alembic.dashboard")

161 )

162 if self.request.method != "POST":

163 return self.redirect(referrer)

164

165 revspec = self.request.POST.get("revspec")

166 if not revspec:

167 self.request.session.flash("You must provide a target revspec.", "error")

168

169 else:

170 direction = self.request.POST.get("direction")

171 if direction != "downgrade":

172 direction = "upgrade"

173 alembic = make_alembic_config(self.config)

174 command = (

175 alembic_command.downgrade

176 if direction == "downgrade"

177 else alembic_command.upgrade

178 )

179

180 # invoke alembic upgrade/downgrade

181 try:

182 command(alembic, revspec)

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

184 log.exception(

185 "database failed to %s using revspec: %s", direction, revspec

186 )

187 self.request.session.flash(f"Migrate failed: {err}", "error")

188 else:

189 self.request.session.flash("Database has been migrated.")

190

191 return self.redirect(referrer)

192

193 @classmethod

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

195 """ """

196 cls._defaults(config)

197

198 @classmethod

199 def _defaults(cls, config):

200

201 # permission group

202 config.add_wutta_permission_group(

203 "alembic", "Alembic (General)", overwrite=False

204 )

205

206 # dashboard

207 config.add_wutta_permission(

208 "alembic",

209 "alembic.dashboard",

210 "Basic (view) access to the Alembic Dashboard",

211 )

212 config.add_route("alembic.dashboard", "/alembic/dashboard")

213 config.add_view(

214 cls,

215 attr="dashboard",

216 route_name="alembic.dashboard",

217 renderer="/alembic/dashboard.mako",

218 permission="alembic.dashboard",

219 )

220

221 # migrate

222 config.add_wutta_permission(

223 "alembic",

224 "alembic.migrate",

225 "Run migration scripts on the database",

226 )

227 config.add_route("alembic.migrate", "/alembic/migrate")

228 config.add_view(

229 cls,

230 attr="migrate",

231 route_name="alembic.migrate",

232 permission="alembic.migrate",

233 )

234

235

236class AlembicMigrationView(MasterView): # pylint: disable=abstract-method

237 """

238 Master view for Alembic Migrations.

239

240 Route prefix is ``alembic.migrations``; notable URLs include:

241

242 * ``/alembic/migrations/``

243 * ``/alembic/migrations/new``

244 * ``/alembic/migrations/XXX``

245 """

246

247 # pylint: disable=duplicate-code

248 model_name = "alembic_migration"

249 model_key = "revision"

250 model_title = "Alembic Migration"

251 route_prefix = "alembic.migrations"

252 url_prefix = "/alembic/migrations"

253 filterable = False

254 sortable = True

255 sort_on_backend = False

256 paginated = True

257 paginate_on_backend = False

258 editable = False

259 configurable = True

260 # pylint: enable=duplicate-code

261

262 labels = {

263 "doc": "Description",

264 "longdoc": "Long Description",

265 "nextrev": "Next Revision",

266 }

267

268 grid_columns = [

269 "is_head",

270 "revision",

271 "doc",

272 "branch_labels",

273 "down_revision",

274 "created",

275 ]

276

277 sort_defaults = ("is_head", "desc")

278

279 form_fields = [

280 "revision",

281 "doc",

282 "longdoc",

283 "branch_labels",

284 "dependencies",

285 "down_revision",

286 "nextrev",

287 "is_base",

288 "is_branch_point",

289 "is_head",

290 "path",

291 "created",

292 ]

293

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

295 self, columns=None, session=None

296 ):

297 """ """

298 data = []

299 script = get_alembic_scriptdir(self.config)

300 for rev in script.walk_revisions():

301 data.append(normalize_revision(self.config, rev))

302 return data

303

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

305 """ """

306 g = grid

307 super().configure_grid(g)

308

309 # revision

310 g.set_link("revision")

311 g.set_searchable("revision")

312

313 # doc

314 g.set_link("doc")

315 g.set_searchable("doc")

316

317 # branch_labels

318 g.set_searchable("branch_labels")

319

320 # is_head

321 g.set_label("is_head", "Head")

322 g.set_renderer("is_head", self.render_is_head)

323

324 # created

325 g.set_renderer("created", "datetime")

326

327 def render_is_head( # pylint: disable=missing-function-docstring,unused-argument

328 self, rev, field, value

329 ):

330 return self.app.render_boolean(value) if value else ""

331

332 def get_instance(

333 self, **kwargs

334 ): # pylint: disable=empty-docstring,arguments-differ,unused-argument

335 """ """

336 if "_cached_instance" not in self.__dict__:

337 revision = self.request.matchdict["revision"]

338 script = get_alembic_scriptdir(self.config)

339 try:

340 rev = script.get_revision(revision)

341 except CommandError:

342 rev = None

343 if not rev:

344 raise self.notfound()

345 self.__dict__["_cached_instance"] = normalize_revision(self.config, rev)

346 return self.__dict__["_cached_instance"]

347

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

349 """ """

350 text = f"({instance['branch_labels']}) {instance['doc']}"

351 if instance.get("is_head"):

352 text += " [head]"

353 return text

354

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

356 """ """

357 f = form

358 super().configure_form(f)

359

360 # revision

361 f.set_widget("revision", widgets.CopyableTextWidget())

362

363 # longdoc

364 f.set_widget("longdoc", "notes")

365

366 # down_revision

367 f.set_widget("down_revision", widgets.AlembicRevisionWidget(self.request))

368

369 # nextrev

370 f.set_widget("nextrev", widgets.AlembicRevisionsWidget(self.request))

371

372 # is_head

373 f.set_node("is_head", colander.Boolean())

374

375 # path

376 f.set_widget("path", widgets.CopyableTextWidget())

377

378 # created

379 f.set_node("created", WuttaDateTime())

380 f.set_widget("created", widgets.WuttaDateTimeWidget(self.request))

381

382 def make_create_form(self): # pylint: disable=empty-docstring

383 """ """

384 alembic = make_alembic_config(self.config)

385 script = get_alembic_scriptdir(self.config, alembic)

386

387 schema = colander.Schema()

388 schema.add(colander.SchemaNode(colander.String(), name="description"))

389

390 schema.add(

391 colander.SchemaNode(

392 colander.Boolean(),

393 name="autogenerate",

394 default=check_alembic_current(self.config, alembic),

395 )

396 )

397

398 schema.add(

399 colander.SchemaNode(

400 colander.String(), name="branching_option", default="revise"

401 )

402 )

403

404 branch_options = self.get_revise_branch_options(script)

405

406 revise_branch = colander.SchemaNode(

407 colander.String(),

408 name="revise_branch",

409 missing=colander.null,

410 validator=colander.OneOf(branch_options),

411 widget=widgets.SelectWidget(values=[(b, b) for b in branch_options]),

412 )

413

414 branch = self.config.get(f"{self.config.appname}.alembic.default_revise_branch")

415 if not branch and len(branch_options) == 1:

416 branch = branch_options[0]

417 if branch:

418 revise_branch.default = branch

419

420 schema.add(revise_branch)

421

422 schema.add(

423 colander.SchemaNode(

424 colander.String(), name="new_branch", missing=colander.null

425 )

426 )

427

428 version_locations = sorted(

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

430 )

431

432 schema.add(

433 colander.SchemaNode(

434 colander.String(),

435 name="version_location",

436 missing=colander.null,

437 validator=colander.OneOf(version_locations),

438 widget=widgets.SelectWidget(values=[(v, v) for v in version_locations]),

439 )

440 )

441

442 schema.validator = colander.All(

443 self.validate_revise_branch, self.validate_new_branch

444 )

445

446 form = self.make_form(

447 schema=schema,

448 cancel_url_fallback=self.get_index_url(),

449 button_label_submit="Write Script File",

450 )

451

452 form.set_label("revise_branch", "Branch")

453

454 return form

455

456 def validate_revise_branch( # pylint: disable=missing-function-docstring

457 self, node, value

458 ):

459 if value["branching_option"] == "revise":

460 if not value["revise_branch"]:

461 node["revise_branch"].raise_invalid(

462 "Must specify which branch to revise."

463 )

464

465 def validate_new_branch( # pylint: disable=missing-function-docstring

466 self, node, value

467 ):

468 if value["branching_option"] == "new":

469

470 if not value["new_branch"]:

471 node["new_branch"].raise_invalid("New branch requires a name.")

472

473 if not value["version_location"]:

474 node["version_location"].raise_invalid(

475 "New branch requires a version location."

476 )

477

478 def save_create_form(self, form): # pylint: disable=empty-docstring

479 """ """

480 alembic = make_alembic_config(self.config)

481 data = form.validated

482

483 # TODO: this is "mostly" duplicated in AppTableView.write_revision_script()

484

485 # kwargs for `alembic revision` command

486 kw = {

487 "message": data["description"],

488 "autogenerate": data["autogenerate"],

489 }

490 # pylint: disable=duplicate-code

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

492 kw["head"] = "base"

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

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

495 else:

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

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

498

499 # run `alembic revision`

500 revision = alembic_command.revision(alembic, **kw)

501 # pylint: enable=duplicate-code

502

503 intro = HTML.tag(

504 "p",

505 class_="block",

506 c="New migration script has been created. "

507 "Please review and modify the file contents as needed:",

508 )

509

510 path = HTML.tag(

511 "p",

512 class_="block has-background-white has-text-black is-family-monospace",

513 style="padding: 0.5rem;",

514 c=[HTML.tag("wutta-copyable-text", text=revision.path)],

515 )

516

517 outro = HTML.tag(

518 "p",

519 class_="block",

520 c=[

521 "When satisfied, proceed to ",

522 tags.link_to(

523 "Migrate Database", self.request.route_url("alembic.dashboard")

524 ),

525 ".",

526 ],

527 )

528

529 self.request.session.flash(HTML.tag("div", c=[intro, path, outro]))

530 return revision

531

532 def save_delete_form(self, form): # pylint: disable=empty-docstring

533 """ """

534 rev = self.get_instance()

535 os.remove(rev["path"])

536

537 # TODO: this is effectivey duplicated in AppTableView.get_migration_branch_options()

538 def get_revise_branch_options( # pylint: disable=missing-function-docstring

539 self, script

540 ):

541 branches = set()

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

543 branches.update(rev.branch_labels)

544 return sorted(branches)

545

546 def configure_get_simple_settings(self): # pylint: disable=empty-docstring

547 """ """

548 return [

549 {"name": f"{self.config.appname}.alembic.default_revise_branch"},

550 ]

551

552 def configure_get_context( # pylint: disable=empty-docstring,arguments-differ

553 self, **kwargs

554 ):

555 """ """

556 context = super().configure_get_context(**kwargs)

557

558 script = get_alembic_scriptdir(self.config)

559 context["revise_branch_options"] = self.get_revise_branch_options(script)

560

561 return context

562

563

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

565 base = globals()

566

567 AlembicDashboardView = ( # pylint: disable=invalid-name,redefined-outer-name

568 kwargs.get("AlembicDashboardView", base["AlembicDashboardView"])

569 )

570 AlembicDashboardView.defaults(config)

571

572 AlembicMigrationView = ( # pylint: disable=invalid-name,redefined-outer-name

573 kwargs.get("AlembicMigrationView", base["AlembicMigrationView"])

574 )

575 AlembicMigrationView.defaults(config)

576

577

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

579 defaults(config)