Coverage for .tox/coverage/lib/python3.13/site-packages/wuttjamaican/upgrades.py: 100%
85 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-10-08 21:29 -0500
1# -*- coding: utf-8; -*-
2################################################################################
3#
4# WuttJamaican -- Base package for Wutta Framework
5# Copyright © 2023-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"""
24Upgrade Handler
25"""
27import logging
28import os
29import shutil
30import subprocess
31import sys
32import warnings
33from collections import OrderedDict
35from wuttjamaican.app import GenericHandler
37log = logging.getLogger(__name__)
40class UpgradeHandler(GenericHandler):
41 """
42 Base class and default implementation for the :term:`upgrade
43 handler`.
45 See also
46 :meth:`~wuttjamaican.app.AppHandler.get_upgrade_handler()`.
47 """
49 def get_all_target_keys(self):
50 """
51 Retrieve a list of keys for all configured :term:`upgrade
52 targets <upgrade target>`.
54 Note that a ``default`` target will always be included, first
55 in the list.
57 :returns: List of string keys
58 """
59 keys = self.config.get_list(f"{self.config.appname}.upgrades.keys", default=[])
61 # make 'default' the first key
62 if "default" in keys:
63 keys.remove("default")
64 keys.insert(0, "default")
66 return keys
68 def get_all_targets(self):
69 """
70 Retrieve an ordered dict of all configured :term:`upgrade
71 targets <upgrade target>`.
73 The result would be something like::
75 OrderedDict([
76 ("default", {
77 "key": "default",
78 "label": "Poser",
79 "command": "/srv/envs/poser/app/upgrade.sh"
80 }),
81 ("another", {
82 "key": "another",
83 "label": "My Other App",
84 "command": "/srv/envs/poser/app/upgrade-another.sh"
85 }),
86 ])
88 Note that a ``default`` target will always be included, first
89 in the ordering.
91 :returns: :class:`~python:collections.OrderedDict` with all
92 upgrade targets.
93 """
94 targets = OrderedDict()
95 for key in self.get_all_target_keys():
96 targets[key] = self.get_target(key)
97 return targets
99 def get_target(self, key):
100 """
101 Retrieve the configured :term:`upgrade target` for the given
102 key.
104 The target is represented as a simple dict, e.g.::
106 {
107 "key": "default",
108 "label": "Poser",
109 "command": "/srv/envs/poser/app/upgrade.sh",
110 }
112 :param key: Config key for the desired target.
114 :returns: Target as dict.
115 """
116 if not key:
117 key = "default"
119 default_label = key
120 if key == "default":
121 default_label = self.app.get_node_title()
122 label = self.config.get(
123 f"{self.config.appname}.upgrades.{key}.label", default=default_label
124 )
126 command = self.config.get(f"{self.config.appname}.upgrades.{key}.command")
127 if not command and key == "default":
128 if command := self.config.get(f"{self.config.appname}.upgrades.command"):
129 warnings.warn(
130 f"setting '{self.config.appname}.upgrades.command' "
131 "is deprecated; please use setting "
132 f"'{self.config.appname}.upgrades.default.command' instead",
133 DeprecationWarning,
134 )
136 return {"key": key, "label": label, "command": command}
138 def do_execute(self, upgrade, user, **kwargs):
139 """
140 Perform all steps required to fully execute an
141 :term:`upgrade`.
143 This first calls :meth:`execute()` and then will update the
144 given upgrade record to reflect the result.
146 For the ``default`` :term:`upgrade target` only, this will
147 also call :meth:`capture_pip_freeze_output()` before and after
148 the upgrade runs. This allows viewing the diff between them
149 e.g. in web app.
151 :param upgrade: Reference to the
152 :class:`~wuttjamaican.db.model.upgrades.Upgrade` record to
153 be executed.
155 :param user: Reference to the in-app
156 :class:`~wuttjamaican.db.model.auth.User` responsible for
157 executing the upgrade.
159 :param \\**kwargs: Any additional kwargs are passed along to
160 :meth:`execute()`.
162 :returns: Boolean indicating success of the upgrade.
163 """
164 enum = self.app.enum
166 # record the fact that execution has begun for this upgrade
167 # nb. this is done in separate session to ensure it sticks,
168 # but also must update local object to reflect the change
169 with self.app.short_session(commit=True) as s:
170 alt = s.merge(upgrade)
171 alt.executing = True
172 alt.status = enum.UpgradeStatus.EXECUTING
173 upgrade.executing = True
174 upgrade.status = enum.UpgradeStatus.EXECUTING
176 # maybe capture pip freeze output
177 if upgrade.target_key in ("default", None):
178 self.capture_pip_freeze_output(upgrade, "before")
180 # execute proper
181 success = self.execute(upgrade, user, **kwargs)
183 # maybe capture pip freeze output
184 if upgrade.target_key in ("default", None):
185 self.capture_pip_freeze_output(upgrade, "after")
187 # upgrade has finished executing
188 upgrade.executing = False
189 upgrade.executed = self.app.make_utc()
190 upgrade.executed_by = user
191 if success:
192 upgrade.status = enum.UpgradeStatus.SUCCESS
193 else:
194 upgrade.status = enum.UpgradeStatus.FAILURE
196 return success
198 def capture_pip_freeze_output(self, upgrade, occurrence):
199 """
200 Run the ``pip freeze`` command and capture its output to file,
201 in the given upgrade's data folder.
203 This is called from :meth:`do_execute()` but only for the
204 ``default`` :term:`upgrade target`.
206 See also :meth:`get_pip_path()`.
208 :param upgrade: Reference to the
209 :class:`~wuttjamaican.db.model.upgrades.Upgrade` record
210 being executed.
212 :param occurrence: String identifier, should be ``"before"``
213 or ``"after"`` only.
215 :returns: Output file path as string.
216 """
217 # nb. specify create=True to ensure *folder* exists
218 path = self.get_upgrade_filepath(
219 upgrade, f"requirements.{occurrence}.txt", create=True
220 )
222 pip = self.get_pip_path()
223 with open(path, "wb") as stdout:
224 subprocess.call([pip, "freeze"], stdout=stdout)
225 return path
227 def get_pip_path(self):
228 """
229 Retrieve the path to ``pip`` executable.
231 This is called from :meth:`capture_pip_freeze_output()`.
233 :returns: File path as string
234 """
235 # TODO: ugh how should we really do this
236 return os.path.join(sys.prefix, "bin", "pip")
238 def execute(self, upgrade, user, **kwargs): # pylint: disable=unused-argument
239 """
240 Execute the given upgrade as the given user.
242 This will run the :term:`upgrade target` command "as-is" via
243 :func:`python:subprocess.call()` with ``shell=True``, and
244 capture the STDOUT/STDERR streams to file.
246 Callers should generally avoid using this method directly and
247 instead use :meth:`do_execute()`.
249 :param upgrade: Reference to the
250 :class:`~wuttjamaican.db.model.upgrades.Upgrade` record to
251 be executed.
253 :param user: Reference to the in-app
254 :class:`~wuttjamaican.db.model.auth.User` responsible for
255 executing the upgrade.
257 :param \\**kwargs: Any additional kwargs are ignored by
258 default; subclass may override if needed.
260 :returns: Boolean indicating success of the command (i.e. if
261 exit code was 0).
262 """
263 # command
264 cmd = self.get_target(upgrade.target_key)["command"]
265 if not cmd:
266 raise RuntimeError(f"could not determine command for upgrade: {upgrade}")
268 # output file paths
269 # nb. specify create=True the first time to ensure *folder* exists
270 stdout_path = self.get_upgrade_filepath(upgrade, "stdout.log", create=True)
271 stderr_path = self.get_upgrade_filepath(upgrade, "stderr.log")
273 # run the command
274 log.debug("running upgrade command: %s", cmd)
275 with open(stdout_path, "wb") as stdout:
276 with open(stderr_path, "wb") as stderr:
277 upgrade.exit_code = subprocess.call(
278 cmd, shell=True, text=True, stdout=stdout, stderr=stderr
279 )
280 success = upgrade.exit_code == 0
281 logger = log.debug if success else log.warning
282 logger("upgrade command exited with code %s", upgrade.exit_code)
284 return success
286 def get_upgrade_filepath(self, upgrade, filename=None, create=False):
287 """
288 Get the path to upgrade's data folder, or a file within it.
290 :param upgrade: Reference to the
291 :class:`~wuttjamaican.db.model.upgrades.Upgrade` record.
293 :param filename: Optional filename to append as suffix to the
294 data folder path.
296 :param create: Pass ``True`` here if you want to ensure the
297 data *folder* exists, creating it if necessary. This does
298 not auto-create any files.
300 :returns: Folder/file path as string.
301 """
302 uuid = str(upgrade.uuid)
303 path = self.app.get_appdir(
304 "data", "upgrades", uuid[:2], uuid[2:], create=create
305 )
306 if filename:
307 path = os.path.join(path, filename)
308 return path
310 def delete_files(self, upgrade):
311 """
312 Delete all data files for the given upgrade.
314 Caller must use this method (first!) for full cleanup when
315 deleting an upgrade record from the database.
317 :param upgrade: Reference to the
318 :class:`~wuttjamaican.db.model.upgrades.Upgrade` record.
319 """
320 path = self.get_upgrade_filepath(upgrade, create=False)
321 if os.path.exists(path):
322 shutil.rmtree(path)