ProjectFlask/FlaskMigrateExtension/MigrateProjectExtension.py

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

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

"""
Module implementing the project support for flask-migrate.
"""

import os
import glob

from PyQt6.QtCore import pyqtSlot, QObject, QProcess
from PyQt6.QtWidgets import QMenu, QDialog, QInputDialog, QLineEdit

from eric7 import Utilities
from eric7.EricGui.EricAction import EricAction
from eric7.EricWidgets import EricMessageBox
from eric7.EricWidgets.EricApplication import ericApp

from ..FlaskCommandDialog import FlaskCommandDialog


# TODO: add a submenu with action for the commands with command options
class MigrateProject(QObject):
    """
    Class implementing the flask-migrate project support.
    """

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

        @param plugin reference to the plugin object
        @type ProjectFlaskPlugin
        @param project reference to the project object
        @type Project
        @param parent parent
        @type QObject
        """
        super().__init__(parent)

        self.__plugin = plugin
        self.__project = project

        self.__ericProject = ericApp().getObject("Project")

        self.__migrationSummaryDialog = None

    def initActions(self):
        """
        Public method to define the flask-migrate actions.
        """
        self.actions = []

        self.migrateConfigAct = EricAction(
            self.tr("Configure Migrate"),
            self.tr("C&onfigure Migrate"),
            0,
            0,
            self,
            "flask_config_migrate",
        )
        self.migrateConfigAct.setStatusTip(
            self.tr("Shows a dialog to edit the configuration for flask-migrate")
        )
        self.migrateConfigAct.setWhatsThis(
            self.tr(
                """<b>Configure Migrate</b>"""
                """<p>Shows a dialog to edit the configuration for"""
                """ flask-migrate.</p>"""
            )
        )
        self.migrateConfigAct.triggered.connect(self.__configureMigrate)
        self.actions.append(self.migrateConfigAct)

        self.migrateInstallAct = EricAction(
            self.tr("Install flask-migrate"),
            self.tr("Install &flask-migrate"),
            0,
            0,
            self,
            "flask_install_migrate",
        )
        self.migrateInstallAct.setStatusTip(
            self.tr(
                "Installs the flask-migrate extension into the configured"
                " environment"
            )
        )
        self.migrateInstallAct.setWhatsThis(
            self.tr(
                """<b>Install flask-migrate</b>"""
                """<p>Installs the flask-migrate extension into the configured"""
                """ environment using the pip interface.</p>"""
            )
        )
        self.migrateInstallAct.triggered.connect(self.__installFlaskMigrate)
        self.actions.append(self.migrateInstallAct)

        self.migrateAvailabilityAct = EricAction(
            self.tr("Check flask-migrate Availability"),
            self.tr("Check flask-migrate &Availability"),
            0,
            0,
            self,
            "flask_check_migrate",
        )
        self.migrateAvailabilityAct.setStatusTip(
            self.tr("Check the availability of the flask-migrate extension")
        )
        self.migrateAvailabilityAct.setWhatsThis(
            self.tr(
                """<b>Check flask-migrate Availability</b>"""
                """<p>Check the availability of the flask-migrate extension.</p>"""
            )
        )
        self.migrateAvailabilityAct.triggered.connect(self.__checkAvailability)
        self.actions.append(self.migrateAvailabilityAct)

        #########################################################
        ## action to initialize the database migration system
        #########################################################

        self.migrateInitAct = EricAction(
            self.tr("Initialize Migrations"),
            self.tr("&Initialize Migrations"),
            0,
            0,
            self,
            "flask_init_migrations",
        )
        self.migrateInitAct.setStatusTip(
            self.tr("Initialize support for database migrations")
        )
        self.migrateInitAct.setWhatsThis(
            self.tr(
                """<b>Initialize Migrations</b>"""
                """<p>Initializes the support for database migrations to be"""
                """ stored in the configured migrations directory.</p>"""
            )
        )
        self.migrateInitAct.triggered.connect(self.__initMigrations)
        self.actions.append(self.migrateInitAct)

        #########################################################
        ## action to create a new database migration
        #########################################################

        self.migrateCreateAct = EricAction(
            self.tr("Create Migration"),
            self.tr("&Create Migration"),
            0,
            0,
            self,
            "flask_create_migration",
        )
        self.migrateCreateAct.setStatusTip(
            self.tr("Create a new migration for the current database")
        )
        self.migrateCreateAct.setWhatsThis(
            self.tr(
                """<b>Create Migration</b>"""
                """<p>Creates a new migration for the current database"""
                """ and stores it  in the configured migrations directory.</p>"""
            )
        )
        self.migrateCreateAct.triggered.connect(self.__createMigration)
        self.actions.append(self.migrateCreateAct)

        #########################################################
        ## action to up- and downgrade a database
        #########################################################

        self.upgradeDatabaseAct = EricAction(
            self.tr("Upgrade Database"),
            self.tr("&Upgrade Database"),
            0,
            0,
            self,
            "flask_upgrade_database",
        )
        self.upgradeDatabaseAct.setStatusTip(
            self.tr("Upgrade the database to the current migration")
        )
        self.upgradeDatabaseAct.setWhatsThis(
            self.tr(
                """<b>Upgrade Database</b>"""
                """<p>Upgrades the database to the current migration.</p>"""
            )
        )
        self.upgradeDatabaseAct.triggered.connect(self.upgradeDatabase)
        self.actions.append(self.upgradeDatabaseAct)

        self.downgradeDatabaseAct = EricAction(
            self.tr("Downgrade Database"),
            self.tr("&Downgrade Database"),
            0,
            0,
            self,
            "flask_downgrade_database",
        )
        self.downgradeDatabaseAct.setStatusTip(
            self.tr("Downgrade the database to the previous version")
        )
        self.downgradeDatabaseAct.setWhatsThis(
            self.tr(
                """<b>Downgrade Database</b>"""
                """<p>Downgrades the database to the previous version.</p>"""
            )
        )
        self.downgradeDatabaseAct.triggered.connect(self.downgradeDatabase)
        self.actions.append(self.downgradeDatabaseAct)

        #########################################################
        ## actions to show migrations history information
        #########################################################

        self.migrationSummaryAct = EricAction(
            self.tr("Show Migrations Summary"),
            self.tr("Show Migrations &Summary"),
            0,
            0,
            self,
            "flask_show_migrations_summary",
        )
        self.migrationSummaryAct.setStatusTip(
            self.tr("Show a summary of the created database migrations")
        )
        self.migrationSummaryAct.setWhatsThis(
            self.tr(
                """<b>Show Migrations Summary</b>"""
                """<p>Shows a summary list of the created database"""
                """ migrations.</p>"""
            )
        )
        self.migrationSummaryAct.triggered.connect(self.__showMigrationsSummary)
        self.actions.append(self.migrationSummaryAct)

        self.migrationHistoryAct = EricAction(
            self.tr("Show Migrations History"),
            self.tr("Show Migrations &History"),
            0,
            0,
            self,
            "flask_show_migrations_history",
        )
        self.migrationHistoryAct.setStatusTip(
            self.tr("Show the full history of the created database migrations")
        )
        self.migrationHistoryAct.setWhatsThis(
            self.tr(
                """<b>Show Migrations History</b>"""
                """<p>Shows the full history of the created database"""
                """ migrations.</p>"""
            )
        )
        self.migrationHistoryAct.triggered.connect(self.__showMigrationsHistory)
        self.actions.append(self.migrationHistoryAct)

    def initMenu(self):
        """
        Public method to initialize the flask-migrate menu.

        @return the menu generated
        @rtype QMenu
        """
        menu = QMenu(self.tr("Database"))
        menu.setTearOffEnabled(True)

        menu.addAction(self.migrateConfigAct)
        menu.addSeparator()
        menu.addAction(self.migrateInitAct)
        menu.addSeparator()
        menu.addAction(self.migrateCreateAct)
        menu.addSeparator()
        menu.addAction(self.upgradeDatabaseAct)
        menu.addAction(self.downgradeDatabaseAct)
        menu.addSeparator()
        menu.addAction(self.migrationSummaryAct)
        menu.addAction(self.migrationHistoryAct)
        menu.addSeparator()
        menu.addAction(self.migrateAvailabilityAct)
        menu.addAction(self.migrateInstallAct)

        return menu

    def determineCapability(self):
        """
        Public method to determine the availability of flask-migrate.
        """
        available = (
            self.__project.getData("flask", "flask_migrate_available")
            if self.__project.getData("flask", "flask_migrate_override")
            else self.__flaskMigrateAvailable()
        )
        self.__project.setCapability("flask-migrate", available)

        self.migrateInstallAct.setEnabled(not available)

        for act in (
            self.migrateConfigAct,
            self.migrateInitAct,
            self.migrateCreateAct,
            self.upgradeDatabaseAct,
            self.downgradeDatabaseAct,
            self.migrationSummaryAct,
            self.migrationHistoryAct,
        ):
            act.setEnabled(available)

    def __flaskMigrateAvailable(self):
        """
        Private method to check, if the 'flask-babel' package is available.

        @return flag indicating the availability of 'flask-babel'
        @rtype bool
        """
        interpreter = self.__project.getVirtualenvInterpreter()
        if interpreter and Utilities.isinpath(interpreter):
            detector = os.path.join(
                os.path.dirname(__file__), "FlaskMigrateDetector.py"
            )
            proc = QProcess()
            proc.setProcessChannelMode(QProcess.ProcessChannelMode.MergedChannels)
            proc.start(interpreter, [detector])
            finished = proc.waitForFinished(30000)
            if finished and proc.exitCode() == 0:
                return True

        return False

    def __migrationsDirectory(self, abspath=False):
        """
        Private method to calculate the path of the configured migrations
        directory.

        @param abspath flag indicating to return an absolute path
        @type bool
        @return path of the migrations directory
        @rtype str
        """
        migrations = ""

        self.__ensureMigrateConfigured()

        migrations = self.__project.getData("flask-migrate", "migrationsDirectory")
        if migrations:
            if abspath:
                migrations = self.__ericProject.getAbsoluteUniversalPath(migrations)
            else:
                workdir = self.__project.getApplication()[0]
                migrations = os.path.relpath(
                    self.__ericProject.getAbsoluteUniversalPath(migrations), workdir
                )
        else:
            if abspath:
                migrations = self.__ericProject.getAbsoluteUniversalPath("migrations")

        return migrations

    def projectClosed(self):
        """
        Public method to handle the closing of a project.
        """
        for dlg in (self.__migrationSummaryDialog,):
            if dlg is not None:
                dlg.close()

    ########################################################
    ## Menu related slots below
    ########################################################

    @pyqtSlot()
    def __configureMigrate(self):
        """
        Private slot to show a dialog to edit the migrate configuration.
        """
        from .MigrateConfigDialog import MigrateConfigDialog

        config = self.__project.getData("flask-migrate", "")
        dlg = MigrateConfigDialog(config)
        if dlg.exec() == QDialog.DialogCode.Accepted:
            config = dlg.getConfiguration()
            self.__project.setData("flask-migrate", "", config)

    def __ensureMigrateConfigured(self):
        """
        Private method to ensure, that flask-migrate has been configured.
        """
        config = self.__project.getData("flask-migrate", "")
        if not config:
            self.__configureMigrate()

    @pyqtSlot()
    def __installFlaskMigrate(self):
        """
        Private slot to install the flask-migrate extension into the configured
        environment.
        """
        venvName = self.__project.getVirtualEnvironment()
        if venvName:
            interpreter = self.__project.getFullCommand("python")
            pip = ericApp().getObject("Pip")
            pip.installPackages(["flask-migrate"], interpreter=interpreter)
            self.determineCapability()
        else:
            EricMessageBox.critical(
                None,
                self.tr("Install flask-migrate"),
                self.tr(
                    "The 'flask-migrate' extension could not be installed"
                    " because no virtual environment has been"
                    " configured."
                ),
            )

    @pyqtSlot()
    def __checkAvailability(self):
        """
        Private slot to check the availability of the 'flask-babel' extension.
        """
        self.determineCapability()
        msg = (
            self.tr("The 'flask-migrate' extension is installed.")
            if self.__project.hasCapability("flask-migrate")
            else self.tr("The 'flask-migrate' extension is not installed.")
        )
        EricMessageBox.information(None, self.tr("flask-migrate Availability"), msg)

    #########################################################
    ## slot to initialize the database migration system
    #########################################################

    @pyqtSlot()
    def __initMigrations(self):
        """
        Private slot to initialize the database migration system.
        """
        title = self.tr("Initialize Migrations")

        self.__ensureMigrateConfigured()
        migrations = self.__migrationsDirectory()

        args = ["init"]
        if migrations:
            args += ["--directory", migrations]

        multidb = EricMessageBox.yesNo(
            None,
            self.tr("Multiple Databases"),
            self.tr(
                """Shall the support for multiple databases be""" """ activated?"""
            ),
        )
        if multidb:
            args.append("--multidb")

        dlg = FlaskCommandDialog(
            self.__project,
            title=title,
            msgSuccess=self.tr("\nMigrations initialized successfully."),
        )
        if dlg.startCommand("db", args):
            dlg.exec()
            if dlg.normalExit():
                for root, _dirs, files in os.walk(
                    self.__migrationsDirectory(abspath=True)
                ):
                    for fileName in files:
                        fullName = os.path.join(root, fileName)
                        self.__ericProject.appendFile(fullName)

                browser = (
                    ericApp().getObject("ProjectBrowser").getProjectBrowser("others")
                )
                alembic = os.path.join(
                    self.__migrationsDirectory(abspath=True), "alembic.ini"
                )
                browser.sourceFile.emit(alembic)

    #########################################################
    ## slot to create a new database migration
    #########################################################

    @pyqtSlot()
    def __createMigration(self):
        """
        Private slot to create a new database migration.
        """
        title = self.tr("Create Migration")

        self.__ensureMigrateConfigured()
        migrations = self.__migrationsDirectory()

        message, ok = QInputDialog.getText(
            None,
            title,
            self.tr("Enter a short message for the migration:"),
            QLineEdit.EchoMode.Normal,
        )
        if ok:
            args = ["migrate"]
            if migrations:
                args += ["--directory", migrations]
            if message:
                args += ["--message", message]

            dlg = FlaskCommandDialog(
                self.__project,
                title=title,
                msgSuccess=self.tr("\nMigration created successfully."),
            )
            if dlg.startCommand("db", args):
                dlg.exec()
                if dlg.normalExit():
                    versionsPattern = os.path.join(
                        self.__migrationsDirectory(abspath=True), "versions", "*.py"
                    )
                    for fileName in glob.iglob(versionsPattern):
                        self.__ericProject.appendFile(fileName)

    #########################################################
    ## slots to up- and downgrade a databse
    #########################################################

    @pyqtSlot()
    def upgradeDatabase(self, revision=None):
        """
        Public slot to upgrade the database to the current migration.

        @param revision migration revision to upgrade to
        @type str
        """
        title = self.tr("Upgrade Database")

        self.__ensureMigrateConfigured()
        migrations = self.__migrationsDirectory()

        args = ["upgrade"]
        if migrations:
            args += ["--directory", migrations]
        if revision:
            args.append(revision)

        dlg = FlaskCommandDialog(
            self.__project,
            title=title,
            msgSuccess=self.tr("\nDatabase upgraded successfully."),
        )
        if dlg.startCommand("db", args):
            dlg.exec()

    @pyqtSlot()
    def downgradeDatabase(self, revision=None):
        """
        Public slot to downgrade the database to the previous version.

        @param revision migration revision to downgrade to
        @type str
        """
        title = self.tr("Downgrade Database")

        self.__ensureMigrateConfigured()
        migrations = self.__migrationsDirectory()

        args = ["downgrade"]
        if migrations:
            args += ["--directory", migrations]
        if revision:
            args.append(revision)

        dlg = FlaskCommandDialog(
            self.__project,
            title=title,
            msgSuccess=self.tr("\nDatabase downgraded successfully."),
        )
        if dlg.startCommand("db", args):
            dlg.exec()

    #########################################################
    ## slots to show migrations history information
    #########################################################

    @pyqtSlot()
    def __showMigrationsSummary(self):
        """
        Private slot to show a migrations history summary.
        """
        from .MigrateSummaryDialog import MigrateSummaryDialog

        self.__ensureMigrateConfigured()
        migrations = self.__migrationsDirectory()

        if self.__migrationSummaryDialog is None:
            self.__migrationSummaryDialog = MigrateSummaryDialog(
                self.__project, self, migrations=migrations
            )

        self.__migrationSummaryDialog.showSummary()

    @pyqtSlot()
    def __showMigrationsHistory(self):
        """
        Private slot to show the full migrations history.
        """
        title = self.tr("Migrations History")

        self.__ensureMigrateConfigured()
        migrations = self.__migrationsDirectory()

        args = ["history", "--indicate-current", "--verbose"]
        if migrations:
            args += ["--directory", migrations]

        dlg = FlaskCommandDialog(self.__project, title=title)
        if dlg.startCommand("db", args):
            dlg.exec()