ProjectPyramid/MigrateSummaryDialog.py

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

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

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

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

from eric7.EricGui.EricOverrideCursor import EricOverrideCursor, EricOverridenCursor
from eric7.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, parent=None):
        """
        Constructor

        @param project reference to the project object
        @type Project
        @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.__process = None
        self.__currentItemIndex = 1000000
        self.__currentRevision = ""

    def showSummary(self):
        """
        Public method to show the migrations summary.
        """
        projectPath = self.__project.projectPath()

        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.getAlembicCommand()

        self.__process = QProcess()
        self.__process.setWorkingDirectory(projectPath)

        args = ["-c", "development.ini", "history", "--indicate-current"]

        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 'alembic' process did not finish"""
                                """ within 10 seconds."""
                            ),
                        )
            else:
                with EricOverridenCursor():
                    EricMessageBox.critical(
                        None,
                        self.tr("Migrations Summary"),
                        self.tr("""The 'alembic' 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.__project.upgradeDatabase(revision=rev)
        else:
            self.__project.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
                )