ProjectDjango/DjangoMigrationsListDialog.py

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

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

"""
Module implementing a dialog show a list of all available migrations.
"""

from PyQt6.QtCore import pyqtSlot, Qt, QProcess, QTimer, QPoint
from PyQt6.QtWidgets import (
    QDialog, QDialogButtonBox, QAbstractButton,
    QHeaderView, QTreeWidgetItem, QMenu, QAbstractItemView, QInputDialog,
    QLineEdit
)

from EricWidgets import EricMessageBox

from .Ui_DjangoMigrationsListDialog import Ui_DjangoMigrationsListDialog

import Preferences


class DjangoMigrationsListDialog(QDialog, Ui_DjangoMigrationsListDialog):
    """
    Class implementing a dialog show a list of all available migrations.
    """
    MigrationsListMode = "L"
    MigrationsPlanMode = "P"
    
    def __init__(self, mode, django, parent=None):
        """
        Constructor
        
        @param mode mode of the dialog
        @type str
        @param django reference to the Django project object
        @type Project
        @param parent reference to the parent widget
        @type QWidget
        """
        super().__init__(parent)
        self.setupUi(self)
        
        self.buttonBox.button(
            QDialogButtonBox.StandardButton.Close).setEnabled(False)
        self.buttonBox.button(
            QDialogButtonBox.StandardButton.Cancel).setDefault(True)
        
        self.ioEncoding = Preferences.getSystem("IOEncoding")
        
        self.proc = QProcess()
        self.proc.finished.connect(self.__procFinished)
        self.proc.readyReadStandardOutput.connect(self.__readStdout)
        self.proc.readyReadStandardError.connect(self.__readStderr)
        
        self.__pyExe = ""
        self.__sitePath = ""
        
        self.__django = django
        
        self.__mode = mode
        if self.__mode == DjangoMigrationsListDialog.MigrationsListMode:
            self.setWindowTitle(self.tr("Available Migrations"))
            self.migrationsList.setHeaderLabels([
                self.tr("Name"),
            ])
            self.migrationsList.setSelectionMode(
                QAbstractItemView.SelectionMode.ExtendedSelection)
        else:
            self.setWindowTitle(self.tr("Migrations Plan"))
            self.migrationsList.setHeaderLabels([
                self.tr("Migration"),
                self.tr("Dependencies"),
            ])
            self.migrationsList.setSelectionMode(
                QAbstractItemView.SelectionMode.SingleSelection)
        
        self.refreshButton = self.buttonBox.addButton(
            self.tr("&Refresh"), QDialogButtonBox.ButtonRole.ActionRole)
        self.refreshButton.setToolTip(
            self.tr("Press to refresh the list"))
        self.refreshButton.setEnabled(False)
    
    @pyqtSlot(QAbstractButton)
    def on_buttonBox_clicked(self, button):
        """
        Private slot called by a button of the button box clicked.
        
        @param button button that was clicked
        @type QAbstractButton
        """
        if button == self.buttonBox.button(
            QDialogButtonBox.StandardButton.Close
        ):
            self.close()
        elif button == self.buttonBox.button(
            QDialogButtonBox.StandardButton.Cancel
        ):
            self.__finish()
        elif button == self.refreshButton:
            self.on_refreshButton_clicked()
    
    def __finish(self):
        """
        Private slot called when the process finished or the user pressed the
        button.
        """
        if (
            self.proc is not None and
            self.proc.state() != QProcess.ProcessState.NotRunning
        ):
            self.proc.terminate()
            QTimer.singleShot(2000, self.proc.kill)
            self.proc.waitForFinished(3000)
        
        self.buttonBox.button(
            QDialogButtonBox.StandardButton.Close).setEnabled(True)
        self.buttonBox.button(
            QDialogButtonBox.StandardButton.Cancel).setEnabled(False)
        self.buttonBox.button(
            QDialogButtonBox.StandardButton.Close).setDefault(True)
        
        self.refreshButton.setEnabled(True)
        
        self.__resizeColumns()
    
    def __procFinished(self, exitCode, exitStatus):
        """
        Private slot connected to the finished signal.
        
        @param exitCode exit code of the process
        @type int
        @param exitStatus exit status of the process
        @type QProcess.ExitStatus
        """
        self.__finish()
    
    def __resizeColumns(self):
        """
        Private method to resize the list columns.
        """
        self.migrationsList.header().resizeSections(
            QHeaderView.ResizeMode.ResizeToContents)
        if self.__mode == DjangoMigrationsListDialog.MigrationsListMode:
            self.migrationsList.header().setStretchLastSection(True)
    
    def start(self, pythonExecutable, sitePath, databaseName):
        """
        Public slot used to start the process.
        
        @param pythonExecutable Python executable to be used
        @type str
        @param sitePath path of the site
        @type str
        @param databaseName name of the database to be used
        @type str
        @return flag indicating a successful start of the process
        @rtype bool
        """
        self.__pyExe = pythonExecutable
        self.__sitePath = sitePath
        
        self.errorGroup.hide()
        self.migrationsList.clear()
        
        self.__lastTopItem = None
        
        if sitePath:
            self.proc.setWorkingDirectory(sitePath)
        
        args = []
        args.append("manage.py")
        args.append("showmigrations")
        if self.__mode == DjangoMigrationsListDialog.MigrationsListMode:
            args.append("--list")
        else:
            args.append("--plan")
            args.append("--verbosity")
            args.append("2")
        if databaseName:
            args.append("--database={0}".format(databaseName))
        
        self.proc.start(pythonExecutable, args)
        procStarted = self.proc.waitForStarted()
        if not procStarted:
            self.buttonBox.setFocus()
            EricMessageBox.critical(
                self,
                self.tr('Process Generation Error'),
                self.tr(
                    'The process {0} could not be started. '
                    'Ensure, that it is in the search path.'
                ).format(pythonExecutable))
        return procStarted
    
    def __readStdout(self):
        """
        Private slot to handle the readyReadStdout signal.
        
        It reads the output of the process, formats it and inserts it into
        the contents pane.
        """
        while self.proc.canReadLine():
            s = str(self.proc.readLine(), self.ioEncoding, 'replace').rstrip()
            if self.__mode == DjangoMigrationsListDialog.MigrationsListMode:
                self.__createListItem(s)
            else:
                self.__createPlanItem(s)
    
    def __createListItem(self, line):
        """
        Private method to create an item for list mode.
        
        @param line line of text
        @type str
        """
        if not line.startswith(" "):
            # application name
            self.__lastTopItem = QTreeWidgetItem(
                self.migrationsList, [line.strip()])
            self.__lastTopItem.setExpanded(True)
        else:
            # migration name
            line = line.strip()
            if line.startswith("["):
                applied = line[:3]
                name = line[3:].strip()
                if self.__lastTopItem:
                    itm = QTreeWidgetItem(self.__lastTopItem, [name])
                else:
                    itm = QTreeWidgetItem(self.migrationsList, [name])
                if applied[1] == " ":
                    itm.setCheckState(0, Qt.CheckState.Unchecked)
                else:
                    itm.setCheckState(0, Qt.CheckState.Checked)
    
    def __createPlanItem(self, line):
        """
        Private method to create an item for plan mode.
        
        @param line line of text
        @type str
        """
        line = line.strip()
        applied = line[:3]
        parts = line[3:].strip().split(None, 2)
        if len(parts) == 3:
            dependencies = "\n".join([
                d.strip() for d in parts[2].strip()[1:-1].split(",")
            ])
            itm = QTreeWidgetItem(self.migrationsList, [
                parts[0].strip(),
                dependencies,
            ])
        else:
            itm = QTreeWidgetItem(self.migrationsList, [
                parts[0].strip(),
                "",
            ])
        if applied[1] != " ":
            itm.setCheckState(0, Qt.CheckState.Checked)
    
    def __readStderr(self):
        """
        Private slot to handle the readyReadStderr signal.
        
        It reads the error output of the process and inserts it into the
        error pane.
        """
        if self.proc is not None:
            self.errorGroup.show()
            s = str(self.proc.readAllStandardError(), self.ioEncoding,
                    'replace')
            self.errors.insertPlainText(s)
            self.errors.ensureCursorVisible()
    
    @pyqtSlot()
    def on_refreshButton_clicked(self):
        """
        Private slot to refresh the log.
        """
        self.buttonBox.button(
            QDialogButtonBox.StandardButton.Close).setEnabled(False)
        self.buttonBox.button(
            QDialogButtonBox.StandardButton.Cancel).setEnabled(True)
        self.buttonBox.button(
            QDialogButtonBox.StandardButton.Cancel).setDefault(True)
        
        self.refreshButton.setEnabled(False)
        
        self.start(self.__pyExe, self.__sitePath)
    
    @pyqtSlot(QPoint)
    def on_migrationsList_customContextMenuRequested(self, pos):
        """
        Private slot to show the context menu.
        
        @param pos position the context menu was requested at
        @type QPoint
        """
        menu = QMenu(self.migrationsList)
        menu.addAction(self.tr("Apply All Migrations"),
                       self.__applyAllMigrations)
        
        if self.__mode == DjangoMigrationsListDialog.MigrationsListMode:
            selItems = self.migrationsList.selectedItems()
            if len(selItems) > 0:
                selApps = []
                for itm in selItems:
                    if itm.parent() is None:
                        selApps.append(itm)
                
                menu.addAction(
                    self.tr("Apply Selected Migrations"),
                    self.__applyMigration).setEnabled(len(selItems) == 1)
                menu.addAction(
                    self.tr("Unapply Migrations"),
                    self.__unapplyMigration).setEnabled(len(selApps) == 1)
                menu.addSeparator()
                menu.addAction(
                    self.tr("Make Migrations"),
                    self.__makeMigrations).setEnabled(len(selApps) > 0)
                act = menu.addAction(
                    self.tr("Make Empty Migrations"),
                    lambda: self.__makeMigrations(empty=True))
                act.setEnabled(len(selApps) > 0)
                act = menu.addAction(
                    self.tr("Make Migrations (dry-run)"),
                    lambda: self.__makeMigrations(dryRun=True))
                act.setEnabled(len(selApps) > 0)
        else:
            menu.addAction(self.tr("Apply Selected Migrations"),
                           self.__applyMigration)
        
        menu.popup(self.migrationsList.mapToGlobal(pos))
    
    def __applyAllMigrations(self):
        """
        Private slot to apply all migrations.
        """
        self.__django.applyMigrations()
        self.on_refreshButton_clicked()
    
    def __applyMigration(self):
        """
        Private slot to apply the selected migrations.
        """
        itm = self.migrationsList.selectedItems()[0]
        if self.__mode == DjangoMigrationsListDialog.MigrationsListMode:
            if itm.parent() is None:
                # app item
                app = itm.text(0)
                migration = None
            else:
                migration = itm.text(0)
                app = itm.parent().text(0)
        else:
            app, migration = itm.text(0).split(".", 1)
        
        self.__django.applyMigrations(app=app, migration=migration)
        
        self.on_refreshButton_clicked()
    
    def __unapplyMigration(self):
        """
        Private slot to unapply the selected migrations.
        """
        if self.__mode == DjangoMigrationsListDialog.MigrationsListMode:
            itm = self.migrationsList.selectedItems()[0]
            if itm.parent() is None:
                # only valid for app entries
                app = itm.text(0)
                self.__django.applyMigrations(app=app, migration="zero")
                
                self.on_refreshButton_clicked()
    
    def __makeMigrations(self, dryRun=False, empty=False):
        """
        Private slot to make migrations for the selected apps.
        
        @param dryRun flag indicating a dry-run
        @type bool
        @param empty flag indicating an empty migration
        @type bool
        """
        apps = []
        for itm in self.migrationsList.selectedItems():
            if itm.parent() is None:
                apps.append(itm.text(0))
        
        if apps:
            migration, ok = QInputDialog.getText(
                self,
                self.tr("Make Migrations"),
                self.tr("Enter a name for the migrations (leave empty to"
                        " use system supplied name):"),
                QLineEdit.EchoMode.Normal)
            
            if ok:
                self.__django.makeMigrations(apps, migration, dryRun, empty,
                                             True)
            
            self.on_refreshButton_clicked()