ProjectFlask/FlaskMigrateExtension/MigrateProjectExtension.py

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

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

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

import glob
import os

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

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

try:
    from eric7.SystemUtilities.FileSystemUtilities import isinpath
except ImportError:
    # imports for eric < 23.1
    from eric7.Utilities import isinpath

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 reference to the parent widget
        @type QWidget
        """
        super().__init__(parent)

        self.__plugin = plugin
        self.__project = project
        self.__ui = parent

        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 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, parent=self.__ui)
        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."),
            parent=self.__ui,
        )
        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."),
                parent=self.__ui,
            )
            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."),
            parent=self.__ui,
        )
        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."),
            parent=self.__ui,
        )
        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, parent=self.__ui)
        if dlg.startCommand("db", args):
            dlg.exec()