# -*- coding: utf-8; -*-
################################################################################
#
# wuttaweb -- Web App for Wutta Framework
# Copyright © 2024-2026 Lance Edgar
#
# This file is part of Wutta Framework.
#
# Wutta Framework is free software: you can redistribute it and/or modify it
# under the terms of the GNU General Public License as published by the Free
# Software Foundation, either version 3 of the License, or (at your option) any
# later version.
#
# Wutta Framework is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
# FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
# more details.
#
# You should have received a copy of the GNU General Public License along with
# Wutta Framework. If not, see <http://www.gnu.org/licenses/>.
#
################################################################################
"""
Upgrade Views
"""
import json
import logging
import os
import re
import sqlalchemy as sa
from sqlalchemy import orm
from deform.widget import SelectWidget
from wuttjamaican.db.model import Upgrade
from wuttaweb.views import MasterView
from wuttaweb.forms.schema import UserRef, WuttaEnum, FileDownload
from wuttaweb.forms.widgets import DiffWidget
from wuttaweb.progress import get_progress_session
log = logging.getLogger(__name__)
[docs]
class UpgradeView(MasterView): # pylint: disable=abstract-method
"""
Master view for upgrades.
Default route prefix is ``upgrades``.
Notable URLs provided by this class:
* ``/upgrades/``
* ``/upgrades/new``
* ``/upgrades/XXX``
* ``/upgrades/XXX/edit``
* ``/upgrades/XXX/delete``
"""
model_class = Upgrade
executable = True
execute_progress_template = "/upgrade.mako"
downloadable = True
configurable = True
labels = {
"target_key": "Target",
}
grid_columns = [
"created",
"target_key",
"description",
"status",
"executed",
"executed_by",
]
sort_defaults = ("created", "desc")
def configure_grid(self, grid): # pylint: disable=empty-docstring
""" """
g = grid
super().configure_grid(g)
model = self.app.model
enum = self.app.enum
handler = self.app.get_upgrade_handler()
# target_key
targets = handler.get_all_targets()
if len(targets) == 1:
g.remove("target_key")
else:
g.set_renderer(
"target_key", self.render_target_key_for_grid, targets=targets
)
# description
g.set_link("description")
# created_by
g.set_link("created_by")
Creator = orm.aliased(model.User) # pylint: disable=invalid-name
g.set_joiner(
"created_by",
lambda q: q.join(Creator, Creator.uuid == model.Upgrade.created_by_uuid),
)
g.set_filter("created_by", Creator.username, label="Created By Username")
# status
g.set_renderer("status", self.grid_render_enum, enum=enum.UpgradeStatus)
# executed_by
g.set_link("executed_by")
Executor = orm.aliased(model.User) # pylint: disable=invalid-name
g.set_joiner(
"executed_by",
lambda q: q.outerjoin(
Executor, Executor.uuid == model.Upgrade.executed_by_uuid
),
)
g.set_filter("executed_by", Executor.username, label="Executed By Username")
def render_target_key_for_grid(
self, upgrade, field, value, targets=None
): # pylint: disable=empty-docstring,unused-argument
""" """
key = value or "default"
target = targets.get(key)
return target.get("label", key) if target else key
def grid_row_class( # pylint: disable=empty-docstring,unused-argument
self, upgrade, data, i
):
""" """
enum = self.app.enum
if upgrade.status == enum.UpgradeStatus.EXECUTING:
return "has-background-warning"
if upgrade.status == enum.UpgradeStatus.FAILURE:
return "has-background-warning"
return None
def configure_form(
self, form
): # pylint: disable=empty-docstring,too-many-branches,too-many-statements
""" """
f = form
super().configure_form(f)
enum = self.app.enum
upgrade = f.model_instance
handler = self.app.get_upgrade_handler()
# never show these
f.remove("created_by_uuid", "executing", "executed_by_uuid")
# sequence sanity
f.fields.set_sequence(
[
"target_key",
"description",
"notes",
"status",
"created",
"created_by",
"executed",
"executed_by",
]
)
# target_key
if self.creating:
targets = handler.get_all_targets()
if len(targets) == 1:
f.remove("target_key")
else:
values = [(p["key"], p["label"]) for p in targets.values()]
f.set_widget("target_key", SelectWidget(values=values))
f.set_required("target_key")
else:
target = handler.get_target(upgrade.target_key)
f.fields.insert_after("target_key", "_target_key")
f.remove("target_key")
f.set_readonly("_target_key")
f.set_label("_target_key", "Target")
f.set_default("_target_key", target["label"])
# created
if self.creating or self.editing:
f.remove("created")
# created_by
if self.creating or self.editing:
f.remove("created_by")
else:
f.set_node("created_by", UserRef(self.request))
# notes
f.set_widget("notes", "notes")
# status
if self.creating:
f.remove("status")
else:
f.set_node("status", WuttaEnum(self.request, enum.UpgradeStatus))
# executed
if self.creating or self.editing or not upgrade.executed:
f.remove("executed")
# executed_by
if self.creating or self.editing or not upgrade.executed:
f.remove("executed_by")
else:
f.set_node("executed_by", UserRef(self.request))
# exit_code
if self.creating or self.editing or not upgrade.executed:
f.remove("exit_code")
# stdout / stderr (and package_diff)
if not (self.creating or self.editing) and upgrade.status in (
enum.UpgradeStatus.SUCCESS,
enum.UpgradeStatus.FAILURE,
):
# stdout_file
f.append("stdout_file")
f.set_label("stdout_file", "STDOUT")
url = self.get_action_url(
"download", upgrade, _query={"filename": "stdout.log"}
)
f.set_node("stdout_file", FileDownload(self.request, url=url))
f.set_default(
"stdout_file", handler.get_upgrade_filepath(upgrade, "stdout.log")
)
# stderr_file
f.append("stderr_file")
f.set_label("stderr_file", "STDERR")
url = self.get_action_url(
"download", upgrade, _query={"filename": "stderr.log"}
)
f.set_node("stderr_file", FileDownload(self.request, url=url))
f.set_default(
"stderr_file", handler.get_upgrade_filepath(upgrade, "stderr.log")
)
# package_diff
if self.viewing and upgrade.target_key in ("default", None):
f.append("package_diff")
try:
before = self.read_pip_freeze_output(upgrade, "before")
after = self.read_pip_freeze_output(upgrade, "after")
except Exception: # pylint: disable=broad-exception-caught
f.set_default(
"package_diff", "(failed to read package data for this upgrade)"
)
else:
f.set_widget(
"package_diff",
DiffWidget(
self.request,
before,
after,
columns=["package", "old version", "new version"],
show_filter=True,
filtered_by_default=True,
repr_values=False,
),
)
def read_pip_freeze_output(
self, upgrade, occurrence
): # pylint: disable=empty-docstring
""" """
handler = self.app.get_upgrade_handler()
path = handler.get_upgrade_filepath(upgrade, f"requirements.{occurrence}.txt")
packages = {}
with open(path, "rt", encoding="utf_8") as f:
for line in f:
if line := line.strip():
if line.startswith("#"):
continue
if req := self.parse_requirement(line):
packages[req["name"]] = req["version"]
else:
log.warning("could not parse req from line: %s", line)
return packages
def parse_requirement(self, line): # pylint: disable=empty-docstring
""" """
if match := re.match(r"^.*@(.*)#egg=(.*)$", line):
return {"name": match.group(2), "version": match.group(1)}
if match := re.match(r"^(.*)==(.*)$", line):
return {"name": match.group(1), "version": match.group(2)}
return None
[docs]
def delete_instance(self, obj):
"""
We override this method to delete any files associated with
the upgrade, in addition to deleting the upgrade record. It
calls
:meth:`~wuttjamaican.upgrades.UpgradeHandler.delete_files()`
to accomplish this.
"""
upgrade = obj
handler = self.app.get_upgrade_handler()
# delete output files
handler.delete_files(upgrade)
# continue as normal
super().delete_instance(upgrade)
def objectify(self, form): # pylint: disable=empty-docstring
""" """
upgrade = super().objectify(form)
enum = self.app.enum
# set user, status when creating
if self.creating:
upgrade.created_by = self.request.user
upgrade.status = enum.UpgradeStatus.PENDING
return upgrade
def download_path(self, obj, filename): # pylint: disable=empty-docstring
""" """
upgrade = obj
if filename:
handler = self.app.get_upgrade_handler()
return handler.get_upgrade_filepath(upgrade, filename)
return None
[docs]
def execute_instance(self, obj, user, progress=None):
"""
This will invoke
:meth:`~wuttjamaican.upgrades.UpgradeHandler.do_execute()` on
the :term:`upgrade handler`.
It returns a different "success" message based on whether or
not the execute call failed. (Even if the upgrade *command*
fails, the overall execution did technically complete okay.)
"""
upgrade = obj
handler = self.app.get_upgrade_handler()
if not handler.do_execute(upgrade, user):
# TODO: would be nice to give this msg a "warning" background
return "Upgrade was executed, but it failed"
return "Upgrade was executed okay"
def execute_progress(self): # pylint: disable=empty-docstring
""" """
route_prefix = self.get_route_prefix()
handler = self.app.get_upgrade_handler()
upgrade = self.get_instance()
session = get_progress_session(self.request, f"{route_prefix}.execute")
# session has 'complete' flag set when operation is over
if session.get("complete"):
# set a flash msg for user if one is defined. this is the
# time to do it since user is about to get redirected.
msg = session.get("success_msg")
if msg:
self.request.session.flash(msg)
elif session.get("error"): # uh-oh
# set an error flash msg for user. this is the time to do it
# since user is about to get redirected.
msg = session.get("error_msg", "An unspecified error occurred.")
self.request.session.flash(msg, "error")
# our return value will include all from progress session
data = dict(session)
# add whatever might be new from upgrade process STDOUT
path = handler.get_upgrade_filepath(upgrade, filename="stdout.log")
offset = session.get("stdout.offset", 0)
if os.path.exists(path):
size = os.path.getsize(path) - offset
if size > 0:
with open(path, "rt", encoding="utf_8") as f:
f.seek(offset)
chunk = f.read(size)
data["stdout"] = chunk.replace("\n", "<br />")
session["stdout.offset"] = offset + size
session.save()
return data
def configure_get_context(
self, simple_settings=None
): # pylint: disable=empty-docstring
""" """
context = super().configure_get_context(simple_settings=simple_settings)
handler = self.app.get_upgrade_handler()
# targets
targets = handler.get_all_targets()
context["upgrade_targets"] = list(targets.values())
return context
def configure_gather_settings(
self, data, simple_settings=None
): # pylint: disable=empty-docstring
""" """
settings = super().configure_gather_settings(
data, simple_settings=simple_settings
)
keys = []
for target in json.loads(data["upgrade_targets"]):
key = target["key"]
keys.append(key)
settings.append(
{
"name": f"{self.config.appname}.upgrades.{key}.command",
"value": target["command"],
}
)
if key != "default":
if label := target.get("label"):
settings.append(
{
"name": f"{self.config.appname}.upgrades.{key}.label",
"value": label,
}
)
settings.append(
{
"name": f"{self.config.appname}.upgrades.keys",
"value": ", ".join(keys),
}
)
return settings
def configure_remove_settings(
self, simple_settings=None, session=None
): # pylint: disable=empty-docstring
""" """
super().configure_remove_settings(
simple_settings=simple_settings, session=session
)
model = self.app.model
session = session or self.Session()
to_delete = (
session.query(model.Setting)
.filter(
sa.or_(
model.Setting.name == f"{self.config.appname}.upgrades.keys",
model.Setting.name
== f"{self.config.appname}.upgrades.command", # deprecated
model.Setting.name.like(f"{self.config.appname}.upgrades.%.label"),
model.Setting.name.like(
f"{self.config.appname}.upgrades.%.command"
),
)
)
.all()
)
for setting in to_delete:
self.app.delete_setting(session, setting.name)
@classmethod
def defaults(cls, config): # pylint: disable=empty-docstring
""" """
# nb. Upgrade may come from custom model
wutta_config = config.registry.settings["wutta_config"]
app = wutta_config.get_app()
cls.model_class = app.model.Upgrade
cls._defaults(config)
cls._upgrade_defaults(config)
@classmethod
def _upgrade_defaults(cls, config):
route_prefix = cls.get_route_prefix()
permission_prefix = cls.get_permission_prefix()
instance_url_prefix = cls.get_instance_url_prefix()
# execution progress
config.add_route(
f"{route_prefix}.execute_progress",
f"{instance_url_prefix}/execute/progress",
)
config.add_view(
cls,
attr="execute_progress",
route_name=f"{route_prefix}.execute_progress",
permission=f"{permission_prefix}.execute",
renderer="json",
)
def defaults(config, **kwargs): # pylint: disable=missing-function-docstring
base = globals()
UpgradeView = kwargs.get( # pylint: disable=invalid-name,redefined-outer-name
"UpgradeView", base["UpgradeView"]
)
UpgradeView.defaults(config)
def includeme(config): # pylint: disable=missing-function-docstring
defaults(config)