3
0
Fork 0

feat: add support for Create Alembic Migration

just a wrapper around `alembic revision` but it seems useful...
This commit is contained in:
Lance Edgar 2025-12-24 00:15:37 -06:00
parent 58bf8b4bbb
commit b1ebe1095e
6 changed files with 497 additions and 30 deletions

View file

@ -43,6 +43,13 @@
label="Alembic Migrations"
once />
% endif
% if request.has_perm("alembic.migrations.create"):
<wutta-button type="is-primary"
tag="a" href="${url('alembic.migrations.create')}"
icon-left="plus"
label="New Migration"
once />
% endif
</div>
% if request.has_perm("alembic.migrate"):
<div class="buttons">

View file

@ -0,0 +1,31 @@
## -*- coding: utf-8; -*-
<%inherit file="/configure.mako" />
<%def name="form_content()">
<h3 class="block is-size-3">Basics</h3>
<div class="block" style="padding-left: 2rem; width: 50%;">
<b-field label="Default Branch for new Migrations">
<b-select name="${config.appname}.alembic.default_revise_branch"
v-model="simpleSettings['${config.appname}.alembic.default_revise_branch']"
@input="settingsNeedSaved = true">
<option :value="null">(none)</option>
<option v-for="branch in reviseBranchOptions"
:value="branch">
{{ branch }}
</option>
</b-select>
</b-field>
</div>
</%def>
<%def name="modify_vue_vars()">
${parent.modify_vue_vars()}
<script>
ThisPageData.reviseBranchOptions = ${json.dumps(revise_branch_options)|n}
</script>
</%def>

View file

@ -0,0 +1,70 @@
## -*- coding: utf-8; -*-
<%inherit file="/master/create.mako" />
<%def name="page_content()">
<br />
<div class="buttons">
% if request.has_perm("alembic.dashboard"):
<wutta-button type="is-primary"
tag="a" href="${url('alembic.dashboard')}"
icon-left="forward"
label="Alembic Dashboard"
once />
% endif
</div>
${parent.page_content()}
</%def>
<%def name="form_vue_fields()">
${form.render_vue_field("description", horizontal=False)}
${form.render_vue_field("autogenerate", horizontal=False, label=False, static_text="Auto-generate migration logic based on current app model")}
<br />
<b-field label="Branching Options">
<div style="margin: 1rem;">
<div class="field">
<b-radio name="branching_option"
v-model="${form.get_field_vmodel('branching_option')}"
native-value="revise">
Revise existing branch
</b-radio>
</div>
<div v-show="${form.get_field_vmodel('branching_option')} == 'revise'"
style="padding: 1rem 0;">
${form.render_vue_field("revise_branch", horizontal=True)}
</div>
<div class="field">
<b-radio name="branching_option"
v-model="${form.get_field_vmodel('branching_option')}"
native-value="new">
Start new branch
</b-radio>
</div>
<div v-show="${form.get_field_vmodel('branching_option')} == 'new'"
style="padding: 1rem 0;">
${form.render_vue_field("new_branch", horizontal=True)}
${form.render_vue_field("version_location", horizontal=True)}
<p class="block is-italic">
NOTE: New version locations must be added to the
<span class="is-family-monospace">[alembic]</span> section of
your config file (and app restarted) before they will appear as
options here.
</p>
</div>
</div>
</b-field>
</%def>

View file

@ -0,0 +1,17 @@
## -*- coding: utf-8; -*-
<%inherit file="/master/view.mako" />
<%def name="page_content()">
<br />
<div class="buttons">
% if request.has_perm("alembic.dashboard"):
<wutta-button type="is-primary"
tag="a" href="${url('alembic.dashboard')}"
icon-left="forward"
label="Alembic Dashboard"
once />
% endif
</div>
${parent.page_content()}
</%def>

View file

@ -26,16 +26,21 @@ Views for Alembic
import datetime
import logging
import os
import re
from alembic import command as alembic_command
from alembic.migration import MigrationContext
from alembic.util import CommandError
from wuttjamaican.db.conf import make_alembic_config, get_alembic_scriptdir
from wuttjamaican.db.conf import (
make_alembic_config,
get_alembic_scriptdir,
check_alembic_current,
)
import colander
from webhelpers2.html import tags
from webhelpers2.html import tags, HTML
from wuttaweb.views import View, MasterView
from wuttaweb.forms import widgets
@ -44,7 +49,7 @@ from wuttaweb.forms import widgets
log = logging.getLogger(__name__)
def normalize_revision(config, rev):
def normalize_revision(config, rev): # pylint: disable=missing-function-docstring
app = config.get_app()
created = None
@ -124,7 +129,7 @@ class AlembicDashboardView(View):
"index_title": "Alembic Dashboard",
"script": {
"dir": script.dir,
"version_locations": script.version_locations,
"version_locations": sorted(script.version_locations),
"env_py_location": script.env_py_location,
"file_template": script.file_template,
},
@ -169,22 +174,22 @@ class AlembicDashboardView(View):
else alembic_command.upgrade
)
# invoke alembic upgrade/downgrade
try:
command(alembic, revspec)
except Exception as err:
except Exception as err: # pylint: disable=broad-exception-caught
log.exception(
"database failed to %s using revspec: %s", direction, revspec
)
self.request.session.flash(
f"Database failed to migrate: {err}", "error"
)
self.request.session.flash(f"Migrate failed: {err}", "error")
else:
self.request.session.flash("Database has been migrated.")
return self.redirect(referrer)
@classmethod
def defaults(cls, config):
def defaults(cls, config): # pylint: disable=empty-docstring
""" """
cls._defaults(config)
@classmethod
@ -201,7 +206,7 @@ class AlembicDashboardView(View):
"alembic.dashboard",
"Basic (view) access to the Alembic Dashboard",
)
config.add_route("alembic.dashboard", f"/alembic/dashboard")
config.add_route("alembic.dashboard", "/alembic/dashboard")
config.add_view(
cls,
attr="dashboard",
@ -216,11 +221,7 @@ class AlembicDashboardView(View):
"alembic.migrate",
"Run migration scripts on the database",
)
config.add_route(
"alembic.migrate",
"/alembic/migrate",
# request_method="POST"
)
config.add_route("alembic.migrate", "/alembic/migrate")
config.add_view(
cls,
attr="migrate",
@ -229,16 +230,18 @@ class AlembicDashboardView(View):
)
class AlembicMigrationView(MasterView):
class AlembicMigrationView(MasterView): # pylint: disable=abstract-method
"""
Master view for Alembic Migrations.
Route prefix is ``alembic.migrations``; notable URLs include:
* ``/alembic/migrations/``
* ``/alembic/migrations/new``
* ``/alembic/migrations/XXX``
"""
# pylint: disable=duplicate-code
model_name = "alembic_migration"
model_key = "revision"
model_title = "Alembic Migration"
@ -249,9 +252,9 @@ class AlembicMigrationView(MasterView):
sort_on_backend = False
paginated = True
paginate_on_backend = False
creatable = False
editable = False
deletable = False
configurable = True
# pylint: enable=duplicate-code
labels = {
"doc": "Description",
@ -315,10 +318,14 @@ class AlembicMigrationView(MasterView):
g.set_label("is_head", "Head")
g.set_renderer("is_head", self.render_is_head)
def render_is_head(self, rev, field, value):
return "Yes" if rev.get("is_head") else ""
def render_is_head( # pylint: disable=missing-function-docstring,unused-argument
self, rev, field, value
):
return self.app.render_boolean(value) if value else ""
def get_instance(self): # pylint: disable=empty-docstring
def get_instance(
self, **kwargs
): # pylint: disable=empty-docstring,arguments-differ,unused-argument
""" """
if "_cached_instance" not in self.__dict__:
revision = self.request.matchdict["revision"]
@ -344,6 +351,9 @@ class AlembicMigrationView(MasterView):
f = form
super().configure_form(f)
# revision
f.set_widget("revision", widgets.CopyableTextWidget())
# longdoc
f.set_widget("longdoc", "notes")
@ -356,21 +366,190 @@ class AlembicMigrationView(MasterView):
# is_head
f.set_node("is_head", colander.Boolean())
# path
f.set_widget("path", widgets.CopyableTextWidget())
def make_create_form(self): # pylint: disable=empty-docstring
""" """
alembic = make_alembic_config(self.config)
script = get_alembic_scriptdir(self.config, alembic)
schema = colander.Schema()
schema.add(colander.SchemaNode(colander.String(), name="description"))
schema.add(
colander.SchemaNode(
colander.Boolean(),
name="autogenerate",
default=check_alembic_current(self.config, alembic),
)
)
schema.add(
colander.SchemaNode(
colander.String(), name="branching_option", default="revise"
)
)
branch_options = self.get_revise_branch_options(script)
revise_branch = colander.SchemaNode(
colander.String(),
name="revise_branch",
missing=colander.null,
validator=colander.OneOf(branch_options),
widget=widgets.SelectWidget(values=[(b, b) for b in branch_options]),
)
branch = self.config.get(f"{self.config.appname}.alembic.default_revise_branch")
if not branch and len(branch_options) == 1:
branch = branch_options[0]
if branch:
revise_branch.default = branch
schema.add(revise_branch)
schema.add(
colander.SchemaNode(
colander.String(), name="new_branch", missing=colander.null
)
)
version_locations = sorted(
self.config.parse_list(alembic.get_main_option("version_locations"))
)
schema.add(
colander.SchemaNode(
colander.String(),
name="version_location",
missing=colander.null,
validator=colander.OneOf(version_locations),
widget=widgets.SelectWidget(values=[(v, v) for v in version_locations]),
)
)
schema.validator = colander.All(
self.validate_revise_branch, self.validate_new_branch
)
form = self.make_form(
schema=schema,
cancel_url_fallback=self.get_index_url(),
button_label_submit="Write Script File",
)
form.set_label("revise_branch", "Branch")
return form
def validate_revise_branch(self, node, value):
if value["branching_option"] == "revise":
if not value["revise_branch"]:
node["revise_branch"].raise_invalid(
"Must specify which branch to revise."
)
def validate_new_branch(self, node, value):
if value["branching_option"] == "new":
if not value["new_branch"]:
node["new_branch"].raise_invalid("New branch requires a name.")
if not value["version_location"]:
node["version_location"].raise_invalid(
"New branch requires a version location."
)
def save_create_form(self, form): # pylint: disable=empty-docstring
""" """
alembic = make_alembic_config(self.config)
script = get_alembic_scriptdir(self.config, alembic)
data = form.validated
# kwargs for `alembic revision` command
kw = {
"message": data["description"],
"autogenerate": data["autogenerate"],
}
if data["branching_option"] == "new":
kw["head"] = "base"
kw["branch_label"] = data["new_branch"]
kw["version_path"] = self.app.resource_path(data["version_location"])
else:
assert data["branching_option"] == "revise"
kw["head"] = f"{data['revise_branch']}@head"
# run `alembic revision`
revision = alembic_command.revision(alembic, **kw)
intro = HTML.tag(
"p",
class_="block",
c="New migration script has been created. Please review and modify the file contents as needed:",
)
path = HTML.tag(
"p",
class_="block has-background-white has-text-black is-family-monospace",
style="padding: 0.5rem;",
c=[HTML.tag("wutta-copyable-text", text=revision.path)],
)
outro = HTML.tag(
"p",
class_="block",
c=[
"When satisfied, proceed to ",
tags.link_to(
"Migrate Database", self.request.route_url("alembic.dashboard")
),
".",
],
)
self.request.session.flash(HTML.tag("div", c=[intro, path, outro]))
return revision
def save_delete_form(self, form): # pylint: disable=empty-docstring
""" """
rev = self.get_instance()
os.remove(rev["path"])
def get_revise_branch_options(self, script):
branches = set()
for rev in script.get_revisions(script.get_heads()):
branches.update(rev.branch_labels)
return sorted(branches)
def configure_get_simple_settings(self): # pylint: disable=empty-docstring
""" """
return [
{"name": f"{self.config.appname}.alembic.default_revise_branch"},
]
def configure_get_context( # pylint: disable=empty-docstring,arguments-differ
self, **kwargs
):
""" """
context = super().configure_get_context(**kwargs)
script = get_alembic_scriptdir(self.config)
context["revise_branch_options"] = self.get_revise_branch_options(script)
return context
def defaults(config, **kwargs): # pylint: disable=missing-function-docstring
base = globals()
AlembicDashboardView = (
kwargs.get( # pylint: disable=invalid-name,redefined-outer-name
"AlembicDashboardView", base["AlembicDashboardView"]
)
AlembicDashboardView = ( # pylint: disable=invalid-name,redefined-outer-name
kwargs.get("AlembicDashboardView", base["AlembicDashboardView"])
)
AlembicDashboardView.defaults(config)
AlembicMigrationView = (
kwargs.get( # pylint: disable=invalid-name,redefined-outer-name
"AlembicMigrationView", base["AlembicMigrationView"]
)
AlembicMigrationView = ( # pylint: disable=invalid-name,redefined-outer-name
kwargs.get("AlembicMigrationView", base["AlembicMigrationView"])
)
AlembicMigrationView.defaults(config)

View file

@ -1,5 +1,6 @@
# -*- coding: utf-8; -*-
import os
from unittest.mock import patch
import sqlalchemy as sa
@ -12,9 +13,11 @@ from wuttjamaican.db.conf import (
make_alembic_config,
)
import colander
from pyramid.httpexceptions import HTTPNotFound, HTTPFound
from wuttaweb.views import alembic as mod
from wuttaweb.forms import Form
from wuttaweb.testing import WebTestCase
from wuttaweb.forms.widgets import AlembicRevisionWidget
@ -133,6 +136,166 @@ class TestAlembicMigrationView(WebTestCase):
form = view.make_model_form(rev)
self.assertIsInstance(form.widgets["down_revision"], AlembicRevisionWidget)
def test_make_create_form(self):
self.pyramid_config.add_route("alembic.migrations", "/alembic/migrations/")
view = self.make_view()
# sanity / coverage
form = view.make_create_form()
self.assertIsInstance(form, Form)
self.assertIn("branching_option", form)
def test_validate_revise_branch(self):
self.pyramid_config.add_route("alembic.migrations", "/alembic/migrations/")
view = self.make_view()
form = view.make_create_form()
schema = form.get_schema()
# good example
self.assertIsNone(
view.validate_revise_branch(
schema,
{
"branching_option": "revise",
"revise_branch": "wutta",
},
)
)
# branch is required
self.assertRaises(
colander.Invalid,
view.validate_revise_branch,
schema,
{
"branching_option": "revise",
"revise_branch": None,
},
)
def test_validate_new_branch(self):
self.pyramid_config.add_route("alembic.migrations", "/alembic/migrations/")
view = self.make_view()
form = view.make_create_form()
schema = form.get_schema()
# good example
self.assertIsNone(
view.validate_revise_branch(
schema,
{
"branching_option": "new",
"new_branch": "poser",
"version_location": "wuttjamaican.db:alembic/versions",
},
)
)
# name is required
self.assertRaises(
colander.Invalid,
view.validate_new_branch,
schema,
{
"branching_option": "new",
"new_branch": None,
"version_location": "wuttjamaican.db:alembic/versions",
},
)
# version_location is required
self.assertRaises(
colander.Invalid,
view.validate_new_branch,
schema,
{
"branching_option": "new",
"new_branch": "poser",
"version_location": None,
},
)
def test_save_create_form(self):
self.pyramid_config.add_route("alembic.migrations", "/alembic/migrations/")
self.pyramid_config.add_route("alembic.dashboard", "/alembic/dashboard")
view = self.make_view()
form = view.make_create_form()
# revise branch
form.validated = {
"description": "test revision",
"autogenerate": False,
"branching_option": "revise",
"revise_branch": "wutta",
}
revision = view.save_create_form(form)
# file was saved in wutta dir
self.assertTrue(
revision.path.startswith(
self.app.resource_path("wuttjamaican.db:alembic/versions")
)
)
# get rid of that file!
os.remove(revision.path)
# new branch
form.validated = {
"description": "test revision",
"autogenerate": False,
"branching_option": "new",
"new_branch": "wuttatest",
"version_location": "wuttjamaican.db:alembic/versions",
}
revision = view.save_create_form(form)
# file was saved in wutta dir
self.assertTrue(
revision.path.startswith(
self.app.resource_path("wuttjamaican.db:alembic/versions")
)
)
# get rid of that file!
os.remove(revision.path)
def test_save_delete_form(self):
self.pyramid_config.add_route(
"alembic.migrations.view", "/alembic/migrations/{revision}"
)
view = self.make_view()
alembic = make_alembic_config(self.config)
# write new empty migration script
revision = alembic_command.revision(
alembic,
head="base",
branch_label="wuttatest",
version_path=self.app.resource_path("wuttjamaican.db:alembic/versions"),
message="test revision",
)
# script exists
self.assertTrue(os.path.exists(revision.path))
with patch.object(
self.request, "matchdict", new={"revision": revision.revision}
):
form = view.make_delete_form(revision)
view.save_delete_form(form)
# script gone
self.assertFalse(os.path.exists(revision.path))
def test_configure(self):
self.pyramid_config.add_route("home", "/")
self.pyramid_config.add_route("login", "/auth/login")
self.pyramid_config.add_route("alembic.migrations", "/alembic/migrations")
view = self.make_view()
# sanity/coverage
view.configure()
class TestAlembicDashboardView(WebTestCase):
@ -294,4 +457,4 @@ version_locations = wuttjamaican.db:alembic/versions
self.assertFalse(self.request.session.peek_flash())
self.assertTrue(self.request.session.peek_flash("error"))
[msg] = self.request.session.pop_flash("error")
self.assertTrue(msg.startswith("Database failed to migrate: "))
self.assertTrue(msg.startswith("Migrate failed: "))