Mercurial Repositories > Flask Project / file revision

# -*- coding: utf-8 -*-

# Copyright (c) 2020 - 2022 Detlev Offenbach <detlev@die-offenbachs.de>
#

"""
Module implementing a dialog showing a summary of all created.migrations.
"""

from PyQt6.QtCore import pyqtSlot, Qt, QProcess, QEventLoop, QTimer
from PyQt6.QtGui import QGuiApplication
from PyQt6.QtWidgets import (
    QDialog,
    QDialogButtonBox,
    QAbstractButton,
    QTreeWidgetItem,
    QAbstractItemView,
)

from EricGui.EricOverrideCursor import EricOverrideCursor, EricOverridenCursor
from EricWidgets import EricMessageBox

from .Ui_MigrateSummaryDialog import Ui_MigrateSummaryDialog


class MigrateSummaryDialog(QDialog, Ui_MigrateSummaryDialog):
    """
    Class implementing a dialog showing a summary of all created.migrations.
    """

    def __init__(self, project, migrateProject, migrations="", parent=None):
        """
        Constructor

        @param project reference to the project object
        @type Project
        @param migrateProject reference to the migrate project extension
        @type MigrateProject
        @param migrations directory path containing the migrations
        @type str
        @param parent reference to the parent widget
        @type QWidget
        """
        super().__init__(parent)
        self.setupUi(self)

        self.__refreshButton = self.buttonBox.addButton(
            self.tr("Refresh"), QDialogButtonBox.ButtonRole.ActionRole
        )
        self.__refreshButton.clicked.connect(self.showSummary)

        self.__project = project
        self.__migrateProject = migrateProject
        self.__migrations = migrations

        self.__process = None
        self.__currentItemIndex = 1000000
        self.__currentRevision = ""

    def showSummary(self):
        """
        Public method to show the migrations summary.
        """
        workdir, env = self.__project.prepareRuntimeEnvironment()
        if env is not None:
            self.show()
            self.raise_()

            self.buttonBox.button(QDialogButtonBox.StandardButton.Close).setEnabled(
                False
            )
            self.buttonBox.button(QDialogButtonBox.StandardButton.Cancel).setDefault(
                True
            )
            self.buttonBox.button(QDialogButtonBox.StandardButton.Cancel).setEnabled(
                True
            )
            self.buttonBox.button(QDialogButtonBox.StandardButton.Cancel).setFocus(
                Qt.FocusReason.OtherFocusReason
            )
            QGuiApplication.processEvents(
                QEventLoop.ProcessEventsFlag.ExcludeUserInputEvents
            )

            command = self.__project.getFlaskCommand()

            self.__process = QProcess()
            self.__process.setProcessEnvironment(env)
            self.__process.setWorkingDirectory(workdir)

            args = ["db", "history", "--indicate-current"]
            if self.__migrations:
                args += ["--directory", self.__migrations]

            with EricOverrideCursor():
                self.__process.start(command, args)
                ok = self.__process.waitForStarted(10000)
                if ok:
                    ok = self.__process.waitForFinished(10000)
                    if ok:
                        out = str(self.__process.readAllStandardOutput(), "utf-8")
                        self.__processOutput(out)
                        self.__selectItem(self.__currentRevision)
                    else:
                        with EricOverridenCursor():
                            EricMessageBox.critical(
                                None,
                                self.tr("Migrations Summary"),
                                self.tr(
                                    """The Flask process did not finish"""
                                    """ within 10 seconds."""
                                ),
                            )
                else:
                    with EricOverridenCursor():
                        EricMessageBox.critical(
                            None,
                            self.tr("Migrations Summary"),
                            self.tr(
                                """The Flask process could not be""" """ started."""
                            ),
                        )
                for column in range(self.summaryWidget.columnCount()):
                    self.summaryWidget.resizeColumnToContents(column)

            self.buttonBox.button(QDialogButtonBox.StandardButton.Cancel).setEnabled(
                False
            )
            self.buttonBox.button(QDialogButtonBox.StandardButton.Close).setEnabled(
                True
            )
            self.buttonBox.button(QDialogButtonBox.StandardButton.Close).setDefault(
                True
            )
            self.buttonBox.button(QDialogButtonBox.StandardButton.Close).setFocus(
                Qt.FocusReason.OtherFocusReason
            )

    def __processOutput(self, output):
        """
        Private method to process the flask output and populate the summary
        list.

        @param output output of the flask process
        @type str
        """
        self.summaryWidget.clear()
        self.upDownButton.setEnabled(False)
        self.__currentItemIndex = 1000000
        self.__currentRevision = ""

        lines = output.splitlines()
        for line in lines:
            isCurrent = False
            oldRev, rest = line.split("->")
            rest, message = rest.split(",", 1)
            newRev, *labels = rest.split()
            if labels:
                labelList = [
                    label.replace("(", "").replace(")", "") for label in labels
                ]
                labelsStr = ", ".join(labelList)
                if "current" in labelList:
                    isCurrent = True
            else:
                labelsStr = ""

            itm = QTreeWidgetItem(
                self.summaryWidget,
                [
                    oldRev.strip(),
                    newRev.strip(),
                    message.strip(),
                    labelsStr,
                ],
            )
            if isCurrent:
                font = itm.font(0)
                font.setBold(True)
                for column in range(self.summaryWidget.columnCount()):
                    itm.setFont(column, font)

                self.__currentItemIndex = self.summaryWidget.indexOfTopLevelItem(itm)
                self.__currentRevision = newRev.strip()

    @pyqtSlot()
    def on_summaryWidget_itemSelectionChanged(self):
        """
        Private slot to handle the selection of an entry.
        """
        items = self.summaryWidget.selectedItems()
        if items:
            index = self.summaryWidget.indexOfTopLevelItem(items[0])
            if index < self.__currentItemIndex:
                self.upDownButton.setText(self.tr("Upgrade"))
            elif index > self.__currentItemIndex:
                self.upDownButton.setText(self.tr("Downgrade"))
            self.upDownButton.setEnabled(index != self.__currentItemIndex)
        else:
            self.upDownButton.setEnabled(False)

    @pyqtSlot()
    def on_upDownButton_clicked(self):
        """
        Private slot to upgrade/downgrade to the selected revision.
        """
        itm = self.summaryWidget.selectedItems()[0]
        rev = itm.text(1)
        if self.upDownButton.text() == self.tr("Upgrade"):
            self.__migrateProject.upgradeDatabase(revision=rev)
        else:
            self.__migrateProject.downgradeDatabase(revision=rev)
        self.showSummary()

    @pyqtSlot(QAbstractButton)
    def on_buttonBox_clicked(self, button):
        """
        Private slot handling a button press of the button box.

        @param button reference to the pressed button
        @type QAbstractButton
        """
        if button is self.buttonBox.button(QDialogButtonBox.StandardButton.Cancel):
            self.__cancelProcess()

    @pyqtSlot()
    def __cancelProcess(self):
        """
        Private slot to terminate the current process.
        """
        if (
            self.__process is not None
            and self.__process.state() != QProcess.ProcessState.NotRunning
        ):
            self.__process.terminate()
            QTimer.singleShot(2000, self.__process.kill)
            self.__process.waitForFinished(3000)

        self.__process = None

    def __selectItem(self, revision):
        """
        Private method to select an item given its revision.

        @param revision revision of the item to select
        @type str
        """
        if revision:
            items = self.summaryWidget.findItems(revision, Qt.MatchFlag.MatchExactly, 1)
            if items:
                # select the first item
                items[0].setSelected(True)
                self.summaryWidget.scrollToItem(
                    items[0], QAbstractItemView.ScrollHint.PositionAtCenter
                )