ProjectFlask/FlaskMigrateExtension/MigrateSummaryDialog.py

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

# Copyright (c) 2020 - 2021 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)