msm_plugin/lib/management.py

# Copyright (c) 2025, Oracle and/or its affiliates.
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License, version 2.0,
# as published by the Free Software Foundation.
#
# This program is designed to work with certain software (including
# but not limited to OpenSSL) that is licensed under separate terms, as
# designated in a particular file or component or in included license
# documentation.  The authors of MySQL hereby grant you an additional
# permission to link the program and your derivative works with the
# separately licensed software that they have either included with
# the program or referenced in the documentation.
#
# This program is distributed in the hope that it will be useful,  but
# WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See
# the GNU General Public License, version 2.0, for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA

# Define plugin version

from msm_plugin import lib
import json
import os
from pathlib import Path
import re
import shutil
from string import Template
from datetime import date
from textwrap import indent
import mysqlsh


# Regex to match MSM sections, excluding the license and the last section
MSM_SECTIONS_REGEX = r'--\s+#+\s+--\s+MSM\s+Section\s+(\d+):\s*(.*?)$\s+.*?--\s+#+'

# Regex to match the MSM LOOP UPDATABLE-VERSIONS
MSM_LOOP_UPDATABLE_VERSIONS_REGEX = \
    r'--\s+###\s+MSM-LOOP-START:UPDATABLE-VERSIONS(\(indent=(\d+)\))?.*?\n(.*?)' \
    r'--\s+###\s+MSM-LOOP-END:UPDATABLE-VERSIONS.*?\n'

# Reges to match the MSM section placeholder
MSM_SECTION_PLACEHOLDER_REGEX = r'\$\{section_(\d+).*?\}\n'

# Regex to match the leading comments and empty lines before SQL commands,
# but ensure that comments before SQL commands are kept
REMOVE_LEADING_COMMENTS_AND_EMPTY_LINES = r'((^--.*?\n)+(^\s*\n)*)'

# Regex to match all empty lines, including whitespace, at the end of the file
REMOVE_TRAILING_EMPTY_LINES = r'(\s*\Z)'

MSM_SCHEMA_VERSION_VIEW_VALUES = r'CREATE.*?VIEW.*?`msm_schema_version`.*?' \
    r'\(.*?`major`.*?`minor`.*?`patch`.*?\).*?AS.*?SELECT.*?' \
    r'((\d+).*?,.*?(\d+).*?,.*?(\d+))'

MSM_SECTION_SOURCE_REGEX = \
    r'^([ \t]*)SOURCE\s+[\'"](.*?)[\'"]\s*\[(\d*:-?\d*)\]\s*;.*?$\n'

MSM_SECTION_REMOVE_COMMENTS_AND_DELIMITERS = \
    r'(^\s*--\s+.*?$)|(^DELIMITER\s+.*?$)|(\/\*.*?\*\/)'


def remove_leading_comments_trailing_lines(section: str) -> str:
    """Removes leading comments and empty lines at the beginning and end from the given section

    Args:
        section (str): The section to remove the leading comments from

    Returns:
        The trimmed section text
    """
    lines = section.splitlines()

    # Remove leading comments and empty lines
    while len(lines) > 0 and (lines[0].startswith("-- ") or lines[0].strip() == ""):
        lines = lines[1:]

    # Remove trailing empty lines
    while len(lines) > 0 and lines[len(lines) - 1].strip() == "":
        lines = lines[:len(lines) - 1]

    return "\n".join(lines)


def get_sql_content(full_section_content: str) -> tuple[str, int, int]:
    """Returns the sql content of a MSM section

    Args:
        full_section_content: The full content of the section

    Returns:
        The sql content as well as the start and the end position
    """
    # Search start position of sql content
    start_position = 0
    match = re.search(
        REMOVE_LEADING_COMMENTS_AND_EMPTY_LINES, full_section_content,
        re.MULTILINE | re.DOTALL)
    if match is not None:
        start_position = match.end()

    # Search end position of sql content
    end_position = len(full_section_content)
    match = re.search(
        REMOVE_TRAILING_EMPTY_LINES, full_section_content,
        re.MULTILINE | re.DOTALL)
    if match is not None:
        end_position = match.start()

    return (
        full_section_content[start_position:end_position],
        start_position,
        end_position)


def set_sql_content(section: dict, sql_content: str) -> None:
    """Sets the sql content of a MSM section

    Args:
        section: The MSM section as dict

    Returns:
        None
    """
    if not ("sql_content" in section and "sql_start" in section
            and "sql_end" in section):
        raise ValueError("The given section does not contain SQL content.")

    section["full_content"] = (
        section["full_content"][:section["sql_start"]] + "\n" + sql_content.strip("\n"))
    section["sql_content"] = sql_content
    section["sql_end"] = len(section["full_content"])


def get_script_sections(script: str) -> dict[str, dict]:
    """Returns the MSM sections of a script

    Args:
        script (str): The script as string

    Returns:
        A list of script dicts.
    """

    # Search for MSM sections
    matches = re.finditer(
        MSM_SECTIONS_REGEX, script, re.MULTILINE | re.DOTALL)

    # for match_id, match in enumerate(matches, start=1):
    #     print("Match {match_id} was found at {start}-{end}: {match}".format(
    #         match_id=match_id, start=match.start(), end=match.end(), match=match.group()))
    #     for group_id in range(0, len(match.groups())):
    #         group_id = group_id + 1
    #         print("Group {group_id} found at {start}-{end}: {group}".format(group_id=group_id,
    #               start=match.start(group_id), end=match.end(group_id), group=match.group(group_id)))

    section_headers = []
    for match_id, match in enumerate(matches, start=1):
        # Add license section if available
        if match_id == 1 and match.start() > 0:
            section_headers.append({
                "section_id": "license",
                "start_position": 0,
                "header_length": 0,
            })

        section_headers.append({
            "section_id": match.group(1),
            "start_position": match.start(),
            "header_length": match.end() - match.start(),
        })

    sections = {}

    for idx, header in enumerate(section_headers):
        # Lookup the end of the section by checking the start of the next section,
        # while the last section ends at the end of the script.
        if idx + 1 < len(section_headers):
            section_end = section_headers[idx + 1]["start_position"] - 1
        else:
            section_end = len(script)

        sections[header["section_id"]] = {
            "full_content": script[header["start_position"]:section_end],
            "sql_content": script[header["start_position"] + header["header_length"]:section_end],
            "sql_start": header["header_length"],
            "sql_end": section_end,
            "start_position": header["start_position"],
            "end_position": section_end,
        }

    return sections


def get_file_sections(file_path: str) -> str:
    """Returns the MSM sections of a given file

    Args:
        file_path (str): The path of the SQL file.
        section_id (str): The id of the section.

    Returns:
        The SQL content as string
    """
    if not os.path.exists(file_path):
        raise ValueError(
            f"The file `{file_path}` could not be found.")

    with open(file_path, "r") as f:
        script = f.read()

    return get_script_sections(script)


def get_file_section(file_path: str, section_id: str) -> str:
    """Returns the SQL content of a MSM section of a given file

    Args:
        file_path (str): The path of the SQL file.
        section_id (str): The id of the section.

    Returns:
        The SQL content as string
    """
    sections = get_file_sections(file_path)

    if section_id in sections:
        return sections[section_id]
    else:
        raise ValueError(
            f"The MSM section `{section_id}` could not be found in the file `{file_path}`.")


def set_section_sql_content(
        section_id: str, sql_content: str, file_path: str = None,
        sections: dict[str, dict] = None) -> None:
    """Sets the SQL content of a MSM section of a file

    Args:
        section_id: The id of the section.
        sql_content: The sql content to set
        file_path: The path of the SQL file.
        sections: The MSM sections.

    Returns:
        None
    """
    if sections is None and file_path is not None:
        sections = get_file_sections(file_path)
    elif sections is None:
        raise ValueError("Either a file_path or sections must be set.")

    if not section_id in sections:
        raise ValueError(f"The given section `{section_id}` was not found.")

    section = sections[section_id]
    set_sql_content(section=section, sql_content=sql_content)

    if file_path is not None:
        write_sections_to_file(
            file_path=file_path, sections=sections, overwrite_existing=True)


def write_sections_to_file(file_path: str, sections: dict[str, dict], overwrite_existing: bool = False):
    """Writes all MSM sections to a file

    Args:
        file_path: The path write the sections to.
        sections: The MSM sections to write out.
        overwrite_existing: Whether an existing file should be replaced. Defaults to False.

    Returns:
        None
    """
    if os.path.exists(file_path) and not overwrite_existing:
        raise ValueError(
            f"The file {file_path} already exists. Please explicitly allow to replace existing files.")

    with open(file_path, "w") as f:
        f.write("\n".join(
            sections[section_id].get("full_content")
            for section_id in sections))


def convert_string_to_valid_filename(name: str) -> str:
    """Converts a string to a valid filename

    Replaces all spaces with underscores and remove characters that are not allowed.

    Args:
        name: The string to convert

    Returns:
        The converted filename
    """
    # Replace all spaces with underscores
    r = re.compile(r"\s+", re.MULTILINE)
    file_name = r.sub("_", name)

    # Remove characters that are not allowed
    r = re.compile(r"[\\\\/:*?\"<>|]", re.MULTILINE)
    file_name = r.sub("", file_name)

    return file_name


def check_mysql_identifier(identifier: str, must_be_usable_when_unquoted: bool = True):
    """Checks if the given identifier is valid

    Args:
        identifier: The identifier to be checked
        allow_special_chars: Wether

    Returns:
        The converted filename
    """
    if not must_be_usable_when_unquoted:
        if not re.match(r"^[A-Za-z0-9_]+[A-Za-z0-9_$]*$", identifier):
            raise ValueError(
                "Only basic Latin letters, digits 0-9, dollar, underscore are allowed to be used as schema name. "
                "Use the allow_special_chars option to overwrite this behavior.")
        if re.match(r"^[0-9]+$", identifier):
            raise ValueError(
                "A schema name must not consist solely of digits. "
                "Use the allow_special_chars option to overwrite this behavior.")


def get_license_text(schema_project_path: str = None, project_settings: dict = None) -> str:
    """Gets the correct license text for the given project

    The license text is fetch from the right license template and the variables are
    substituted accordingly.

    Args:
        schema_project_path: The MSM project path
        project_settings: Optional project settings

    Returns:
        The converted filename
    """
    if project_settings is None and schema_project_path is not None:
        project_settings = get_project_settings(schema_project_path)
    elif project_settings is None and schema_project_path is None:
        raise ValueError(
            "No schema_project_path nor project_settings parameter given.")

    try:
        copyright_holder = project_settings.get("copyrightHolder")
        license = project_settings.get("license")
        custom_license = project_settings.get("customLicense")
        year_of_creation = project_settings.get("yearOfCreation")
    except Exception as e:
        raise ValueError(
            f"The project settings must include copyrightHolder, license, customLicense and yearOfCreation. {e}")

    license_template = None
    if license.upper() == "CUSTOM":
        if custom_license is not None and custom_license != "":
            license_template = Template(custom_license)
        else:
            raise ValueError("No custom license text specified.")
    else:
        template_folder = os.path.join(
            Path(__file__).parent.parent, "templates")
        license_file_path = os.path.join(
            template_folder, "license", f"{license}.txt")
        if not os.path.exists(license_file_path):
            raise ValueError(
                f"No license stored under the given license name `{license}`. Please use a custom license text.")
        with open(license_file_path) as f:
            license_template = Template("".join(f.readlines()[1:]))

    current_year = date.today().strftime("%Y")
    try:
        license_text = license_template.substitute({
            "copyright_holder": copyright_holder,
            "year": year_of_creation if year_of_creation == current_year else f"{year_of_creation}, {current_year}"
        })
    except:
        raise ValueError(
            "The license template is either missing the ${year} or ${copyright_holder} placeholders.")

    return license_text


def copy_template_file_and_substitute(
        source_file_path: str, target_file_path: str, substitutions: dict):
    """Copies a template file and substitutes the given variables

    Args:
        source_file_path: The path to the source file
        target_file_path: The path to the target file
        substitutions: The dict holding the substitutions

    Returns:
        None
    """
    # Create schema development script
    with open(source_file_path, "r") as f:
        # Remove copyright line and replace placeholders
        script = Template("".join(f.readlines()[1:]))
        script = script.substitute(substitutions)

    with open(target_file_path, "w") as f:
        f.write(script)


def create_schema_project_folder(
        schema_name: str, target_path: str, copyright_holder: str, license: str = None,
        overwrite_existing: bool = False, allow_special_chars: bool = False,
        enforce_target_path: bool = False) -> str:
    """Creates a new schema project folder.

    Args:
        schema_name (str): The name of the schema.
        target_path (str): The path to the schema project folder.
        copyright_holder (str): The name of the copyright holder.
        license (str): The license to use for the project.
        overwrite_existing (bool): If the project folder already exists, overwrite it.
        allow_special_chars (bool): If set to True, allows all characters
        overwrite_existing (bool): If the project folder already exists, overwrite it.
        enforce_target_path (bool): If set to true, the target_path is created if it does not yet exist.

    Returns:
        The path of the project folder that was created.
    """

    check_mysql_identifier(
        identifier=schema_name,
        must_be_usable_when_unquoted=not allow_special_chars)

    target_path = os.path.abspath(os.path.expanduser(target_path))
    if not os.path.exists(target_path):
        if enforce_target_path:
            Path(target_path).mkdir(parents=True, exist_ok=True)
        else:
            raise ValueError(
                f'The project folder cannot be created inside the directory `{target_path}` as this path '
                'does not exist.')

    # When used in file names, critical characters need to be removed
    schema_file_name = convert_string_to_valid_filename(schema_name)

    # Create the project folder
    project_path = os.path.join(target_path, f"{schema_file_name}.msm.project")
    if os.path.exists(project_path):
        if overwrite_existing:
            shutil.rmtree(project_path)
        else:
            raise ValueError(
                f'The project folder "{project_path}" already exists.')

    os.makedirs(os.path.join(project_path), exist_ok=True)

    # Copy the template project file structure
    template_folder = os.path.join(Path(__file__).parent.parent, "templates")
    shutil.copytree(
        Path(os.path.join(template_folder, "msm.project")),
        project_path,
        dirs_exist_ok=True)

    # Get current year
    year_of_creation = date.today().strftime("%Y")

    project_settings = {
        "copyrightHolder": copyright_holder,
        "customLicense": "",
        "license": license,
        "schemaDependencies": [],
        "schemaName": schema_name,
        "schemaFileName": schema_file_name,
        "yearOfCreation": year_of_creation,
    }

    # Write the msm.project.json file
    with open(os.path.join(project_path, "msm.project.json"), "w") as f:
        f.write(json.dumps(project_settings, indent=4))

    # Replace placeholders in all markdown files
    files = list(Path(project_path).rglob("*.md"))
    for file in files:
        with open(file, "r") as f:
            script = Template(f.read())
            script = script.substitute({
                "schema_name": schema_name,
            })

        r = re.compile(r"Copyright.*$", re.MULTILINE)
        script = r.sub(
            f"Copyright (c) {year_of_creation}, {copyright_holder}.", script)

        with open(file, "w") as f:
            f.write(script)

    copy_template_file_and_substitute(
        source_file_path=os.path.join(
            template_folder, "scripts", "schema_next.sql"),
        target_file_path=os.path.join(
            project_path, "development", f"{schema_file_name}_next.sql"),
        substitutions={
            "license": get_license_text(project_settings=project_settings),
            "schema_name": schema_name,
            "version_str": "0.0.1",
            "version_comma_str": "0, 0, 1",
        })

    return project_path


def get_schema_development_file_path(schema_project_path: str = None) -> str:
    schema_file_name = get_project_settings(
        schema_project_path).get("schemaFileName", None)
    if schema_file_name is None:
        raise ValueError(
            f"The settings files of the project `{schema_project_path}` does not contain "
            "a schemaFileName value.")

    schema_dev_file_path = os.path.join(
        schema_project_path, "development", f"{schema_file_name}_next.sql")
    if not os.path.exists(schema_dev_file_path):
        raise ValueError(
            f"The MSM project folder does not contain a schema development file `{schema_dev_file_path}`.")

    return schema_dev_file_path


def get_schema_development_sections(schema_project_path: str = None) -> dict[str, dict]:
    schema_dev_file_path = get_schema_development_file_path(
        schema_project_path)

    with open(schema_dev_file_path, "r") as f:
        script = f.read()

    return get_script_sections(script)


def write_schema_development_file(schema_project_path: str, sections: dict[str, dict]) -> None:
    schema_dev_file_path = get_schema_development_file_path(
        schema_project_path)
    write_sections_to_file(
        file_path=schema_dev_file_path,
        sections=sections,
        overwrite_existing=True)


def schema_development_version(
        schema_project_path: str = None, sections: dict[str, dict] = None, new_version: str = None,
        write_to_file: bool = False) -> str:
    """Returns the development version of the current schema

    Args:
        schema_project_path: The path to the schema project folder.
        sections: The list of sections of the schema development file.
        new_version: If not None, the given version is set as new development version.
        write_to_file: Whether the development file should be updated. Defaults to False.

    Returns:
        A dict with information about the project
    """
    if sections is None and schema_project_path is None:
        raise ValueError(
            "No sections nor schema_project_path parameters given.")

    if sections is None:
        sections = get_schema_development_sections(
            schema_project_path=schema_project_path)

    # Process MSM Section 910: Database Schema Version
    version_section = sections.get("910", None)
    if version_section is None:
        for section_id in sections:
            section = sections[section_id]
            print(
                f"{section_id=}\n{section['full_content']=}\n\n{section['sql_content']=}\n\n\n")
        raise ValueError(
            "The script section `MSM Section 910: Database Schema Version` could not be found.")

    section_content = version_section.get("full_content")
    version_values = re.search(
        MSM_SCHEMA_VERSION_VIEW_VALUES, section_content, re.MULTILINE | re.DOTALL)
    if version_values is None:
        raise ValueError(
            "The script section `MSM Section 910: Database Schema Version` does not include the "
            "CREATE VIEW statement to create the `msm_schema_version` VIEW.")

    # If no new_version is given, return the current one
    if new_version is None:
        return f"{version_values.group(2)}.{version_values.group(3)}.{version_values.group(4)}"

    new_version_list = lib.core.convert_version_str_to_list(new_version)

    # Update the version
    updated_content = (
        section_content[:version_values.start(1)]
        + ", ".join(str(number) for number in new_version_list)
        + section_content[version_values.end(1):])

    version_section["full_content"] = updated_content

    sections["910"] = version_section

    if write_to_file:
        write_schema_development_file(
            schema_project_path=schema_project_path,
            sections=sections)

    return new_version


def get_schema_development_version(schema_project_path: str = None, sections: dict[str, dict] = None) -> str:
    """Returns the development version of the current schema

    Args:
        schema_project_path: The path to the schema project folder.
        sections: The list of sections of the schema development file.

    Returns:
        A dict with information about the project
    """
    return schema_development_version(
        schema_project_path=schema_project_path, sections=sections)


def set_development_version(
        new_version: str, schema_project_path: str = None, sections: dict = None, write_to_file: bool = False) -> dict:
    """Sets the development version inside the development/schema_next.sql file

    Args:
        version: The new version to create.
        schema_project_path: The path to the schema project folder.
        sections: The list of sections of the schema development file.
        write_to_file: Whether the development file should be updated. Defaults to False.

    Returns:
        None
    """
    return schema_development_version(
        schema_project_path=schema_project_path, sections=sections,
        new_version=new_version, write_to_file=write_to_file)


def get_project_settings(schema_project_path: str) -> dict:
    """Returns the project settings

    Args:
        schema_project_path: The path to the schema project folder.

    Returns:
        A dict with the project settings
    """
    if schema_project_path is None:
        raise ValueError("No schema_project_path given.")

    project_settings_file = os.path.join(
        schema_project_path, "msm.project.json")
    if not os.path.exists(project_settings_file):
        raise ValueError(
            f"The path {schema_project_path} does not contain a MSM project folder.")

    with open(project_settings_file, "r") as f:
        try:
            project_settings = json.loads(f.read())
        except:
            raise ValueError(
                f"The contents of the MSM project settings file `{project_settings_file}` is corrupted.")

    return project_settings


def get_released_versions(schema_project_path: str) -> list[list[int, int, int]]:
    """Returns the all released version of the database schema

    Args:
        schema_project_path: The path to the schema project folder.

    Returns:
        The list of released database versions as list of list of 3 ints.
    """
    if schema_project_path is None:
        raise ValueError("No schema_project_path given.")

    # Look at all files and get the highest version
    files = []
    for file in next(os.walk(os.path.join(schema_project_path, "releases", "versions")))[2]:
        files.append(file)

    versions = []
    for file in files:
        version_match = re.match(r".*?(\d+)\.(\d+)\.(\d+)\.sql", file)
        if version_match is not None:
            file_version = [int(version_match.group(
                1)), int(version_match.group(2)), int(version_match.group(3))]
            versions.append(file_version)

    versions.sort()

    return versions


def get_updatable_versions(schema_project_path: str) -> list[list[int, int, int]]:
    """Returns the all version that have update scripts

    Args:
        schema_project_path: The path to the schema project folder.

    Returns:
        The list of updatable database versions as list of list of 3 ints.
    """
    if schema_project_path is None:
        raise ValueError("No schema_project_path given.")

    # Look at all files and get the highest version
    files = []
    for file in next(os.walk(os.path.join(schema_project_path, "releases", "updates")))[2]:
        files.append(file)

    versions = []
    versions_to = []
    for file in files:
        version_match = re.match(
            r".*?(\d+)\.(\d+)\.(\d+)_to_(\d+)\.(\d+)\.(\d+)\.sql", file)
        if version_match is not None:
            file_version = [int(version_match.group(
                1)), int(version_match.group(2)), int(version_match.group(3))]
            versions.append(file_version)
            to_version = [int(version_match.group(
                4)), int(version_match.group(5)), int(version_match.group(6))]
            versions_to.append(to_version)

    versions.sort()
    versions_to.sort()

    # Check if there are any missing version updates
    if len(versions) > 0:
        versions_to.insert(0, versions[0])
        del versions_to[-1]

        missing_upgrade_from = None
        missing_upgrade_to = None
        for i in range(len(versions)):
            if versions[i] != versions_to[i]:
                missing_upgrade_from = versions[i]
                missing_upgrade_to = versions_to[i]
            else:
                continue
            break

        if missing_upgrade_from is not None:
            raise Exception(
                "The list of upgrade scripts misses an upgrade step from "
                f"version {missing_upgrade_from} to {missing_upgrade_to}. "
                "Please create the required update script.")

    return versions


def get_last_released_version(schema_project_path: str) -> list[int, int, int] | None:
    """Returns the last released version of the database schema

    Args:
        schema_project_path: The path to the schema project folder.

    Returns:
        The last released version or None if no version was released yet
    """
    versions = get_released_versions(schema_project_path)

    if len(versions) > 0:
        return versions[-1]
    else:
        return None


def get_deployment_script_versions(schema_project_path: str) -> list[list[int, int, int]]:
    """Returns the all deployed version of the database schema

    Args:
        schema_project_path: The path to the schema project folder.

    Returns:
        The list of deployed database versions as list of list of 3 ints.
    """
    if schema_project_path is None:
        raise ValueError("No schema_project_path given.")

    # Look at all files and get the highest version
    files = []
    for file in next(os.walk(os.path.join(schema_project_path, "releases", "deployment")))[2]:
        files.append(file)

    versions = []
    for file in files:
        version_match = re.match(r".*?(\d+)\.(\d+)\.(\d+)\.sql", file)
        if version_match is not None:
            file_version = [int(version_match.group(
                1)), int(version_match.group(2)), int(version_match.group(3))]
            versions.append(file_version)

    versions.sort()

    return versions


def get_last_deployment_script_version(schema_project_path: str) -> list[int, int, int] | None:
    """Returns the last deployment version of the database schema

    Args:
        schema_project_path: The path to the schema project folder.

    Returns:
        The version of the last deployment or None if no version was released yet
    """
    versions = get_deployment_script_versions(schema_project_path)

    if len(versions) > 0:
        return versions[-1]
    else:
        return None


def get_project_info(schema_project_path: str) -> dict:
    """Returns information about the project

    Args:
        schema_project_path: The path to the schema project folder.

    Returns:
        A dict with information about the project
    """

    project_info = get_project_settings(schema_project_path)
    project_info["currentDevelopmentVersion"] = get_schema_development_version(
        schema_project_path=schema_project_path)
    last_released_version = get_last_released_version(
        schema_project_path)
    project_info["lastReleasedVersion"] = (
        '%d.%d.%d' % tuple(last_released_version) if last_released_version is not None else None)
    project_info["schemaDevelopmentFilePath"] = get_schema_development_file_path(
        schema_project_path)

    return project_info


def sql_content_has_no_statement(sql_content: str, pattern: re.Pattern[str] = None) -> bool:
    """Removes all comments and delimiter statements

    Args:
        section_content: The section content

    Returns:
        True if the section content only contains comments or delimiter statements
    """

    if pattern is None:
        pattern = re.compile(
            MSM_SECTION_REMOVE_COMMENTS_AND_DELIMITERS, re.MULTILINE | re.DOTALL)

    return re.sub(pattern, "", sql_content, 0).strip() == ""


def remove_empty_sections(sections: dict, keep_even_if_empty=list[str]) -> None:
    """Removes all empty sections from the given section dict

    Args:
        sections: The section dict with all sections
        keep_even_if_empty: A list of section ids that should kept even if empty

    Returns:
        None
    """

    pattern = re.compile(
        MSM_SECTION_REMOVE_COMMENTS_AND_DELIMITERS, re.MULTILINE | re.DOTALL)

    sections_to_remove = []

    for section_id in sections:
        if section_id not in keep_even_if_empty:
            section = sections[section_id]
            full_content = section.get("full_content", "")
            if sql_content_has_no_statement(full_content, pattern):
                sections_to_remove.append(section_id)

    for section_id in sections_to_remove:
        sections.pop(section_id, None)


def prepare_release(
        schema_project_path: str, version: str, next_version: str, allow_to_stay_on_same_version: bool = False,
        overwrite_existing: bool = False) -> list[str]:
    """Adds a new release to the schema project

    Args:
        schema_project_path: The path to the schema project folder.
        version: The new version to create.
        next_version: The next development version.
        allow_to_stay_on_same_version: Whether to allow to stay on the
            same version for further development work
        overwrite_existing: Whether existing files should be overwritten. Defaults to False.

    Returns:
        A list of filenames that have been created or changed for the release
    """

    project_settings = get_project_settings(schema_project_path)
    schema_name = project_settings.get("schemaName")
    schema_file_name = project_settings.get("schemaFileName")

    last_released_version = get_last_released_version(schema_project_path)
    version_as_ints = lib.core.convert_version_str_to_list(version)

    files_for_release = []

    # Ensure provided version are OK
    if last_released_version is not None:
        if last_released_version > version_as_ints:
            raise ValueError(
                f"The given version {version} is lower than the last released "
                f"version {'%d.%d.%d' % tuple(last_released_version)}.")

    if not allow_to_stay_on_same_version and lib.core.convert_version_str_to_list(next_version) <= version_as_ints:
        raise ValueError(
            "The next development version needs to be higher than the version for release "
            f"{version}. Please explicitly pass the according flag to stay on the same version.")
    elif allow_to_stay_on_same_version and lib.core.convert_version_str_to_list(next_version) < version_as_ints:
        raise ValueError(
            "The next development version needs to be at least at the same version as the version for release "
            f"{version}.")

    # Get the current schema development file sections
    schema_dev_file_path = os.path.join(
        schema_project_path, "development", f"{schema_file_name}_next.sql")
    if not os.path.exists(schema_dev_file_path):
        raise ValueError(
            f"The MSM project folder does not contain a schema development file `{schema_dev_file_path}`.")
    with open(schema_dev_file_path, "r") as f:
        schema_dev_script = f.read()
    # Make sure to resolve SOURCE statements with actual content
    schema_dev_script = substitute_source_statements_with_content(
        schema_dev_script, os.path.dirname(schema_dev_file_path))
    schema_dev_expanded_sections = get_script_sections(schema_dev_script)

    # Prepare the schema version template file
    template_folder = os.path.join(Path(__file__).parent.parent, "templates")
    schema_version_template_file_path = os.path.join(
        template_folder, "scripts", "schema_a.b.c.sql")
    if not os.path.exists(schema_version_template_file_path):
        raise ValueError(
            f"The schema release version template file `{schema_version_template_file_path}` does not exist.")
    with open(schema_version_template_file_path, "r") as f:
        # Remove copyright line and replace placeholders
        schema_version_script = Template("".join(f.readlines()[1:]))
    schema_version_script = schema_version_script.substitute({
        # get_license_text(project_settings=project_settings),
        "license": "License Placeholder",
        "schema_name": schema_name,
        "version_str": version,
        "version_comma_str": ", ".join(str(number) for number in version_as_ints),
    })
    schema_version_sections = get_script_sections(schema_version_script)

    # Get the version section of the current schema development file
    schema_dev_version_section = schema_dev_expanded_sections.get("910", None)
    if schema_dev_version_section is None:
        raise ValueError(
            "The script section `MSM Section 910: Database Schema Version` could not be found in "
            f"`{schema_dev_file_path}`.")

    # Loop over all sections of the version file and replace the sections with the ones from the development file
    for section_id in schema_version_sections:
        section = schema_dev_expanded_sections.get(section_id, None)
        if section is not None:
            schema_version_sections[section_id] = section

    remove_empty_sections(
        sections=schema_version_sections,
        keep_even_if_empty=["license", "001"])

    # Write the schema version file out
    schema_version_file_path = os.path.join(
        schema_project_path, "releases", "versions", f"{schema_file_name}_{version}.sql")
    write_sections_to_file(
        file_path=schema_version_file_path,
        sections=schema_version_sections,
        overwrite_existing=overwrite_existing)
    files_for_release.append(schema_version_file_path)

    # Check if update script is needed and if so, write it out
    if last_released_version is not None:
        last_released_version_str = '%d.%d.%d' % tuple(last_released_version)
        if last_released_version_str != version:
            schema_update_file_path = os.path.join(
                schema_project_path, "releases", "updates",
                f"{schema_file_name}_{last_released_version_str}_to_{version}.sql")
            copy_template_file_and_substitute(
                source_file_path=os.path.join(
                    template_folder, "scripts", "schema_x.y.z_to_a.b.c.sql"),
                target_file_path=schema_update_file_path,
                substitutions={
                    "license": get_license_text(project_settings=project_settings),
                    "schema_name": schema_name,
                    "version_from": last_released_version_str,
                    "version_to": version,
                    "version_comma_str": ", ".join(
                        str(number) for number in lib.core.convert_version_str_to_list(version)),
                })
            files_for_release.append(schema_update_file_path)

    # Update the version of the development schema file store in section
    # MSM Section 910: Database Schema Version
    set_development_version(new_version=next_version,
                            schema_project_path=schema_project_path,
                            write_to_file=True)

    return files_for_release


def generate_deployment_script(
        schema_project_path: str, version: str, overwrite_existing: bool = False) -> str:
    """Generate the deployment script for a release

    Args:
        schema_project_path: The path to the schema project folder.
        version: The version to generate the deployment script for.
        overwrite_existing: Whether existing files should be overwritten. Defaults to False.

    Returns:
        The file name path of the generated script
    """
    project_settings = get_project_settings(schema_project_path)
    schema_name = project_settings.get("schemaName")
    schema_file_name = project_settings.get("schemaFileName")

    # Build deployment file path that will be used for output and check that it does not exist
    deployment_file_path = os.path.join(
        schema_project_path, "releases", "deployment",
        f"{schema_file_name}_deployment_{version}.sql")
    if os.path.exists(deployment_file_path) and not overwrite_existing:
        raise ValueError(
            f"The file {deployment_file_path} already exists. Please explicitly allow to replace existing files.")

    # Build the version file path that is used as a source and check that it does exist
    version_file_path = os.path.join(
        schema_project_path, "releases", "versions",
        f"{schema_file_name}_{version}.sql")

    if not os.path.exists(version_file_path):
        raise ValueError(
            f"The file {version_file_path} does not exist exists. Please make "
            f"sure to prepare the release {version} first.")

    # Get the version script file section
    target_version_sections = get_file_sections(version_file_path)

    version_as_ints = lib.core.convert_version_str_to_list(version)

    # Ensure there are no missing updates between 2 released versions
    updatable_versions = get_updatable_versions(schema_project_path)

    # Get the list of released version and build list of updatable versions
    released_versions = get_released_versions(schema_project_path)
    i = 0
    updatable_versions = []
    updatable_versions_sections = {}
    while i + 1 < len(released_versions) and released_versions[i] < version_as_ints:
        version_from = '%d.%d.%d' % tuple(released_versions[i])
        version_to = '%d.%d.%d' % tuple(released_versions[i + 1])

        updatable_versions.append(version_from)
        updatable_versions_sections[version_from] = {
            "version_from": released_versions[i],
            "version_to": released_versions[i + 1],
            "sections": get_file_sections(os.path.join(
                schema_project_path, "releases", "updates",
                f"{schema_file_name}_{version_from}_to_{version_to}.sql")),
        }
        i += 1

    if len(released_versions) == 0:
        raise ValueError(
            "Please prepare a version for release before generating a deployment script.")

    # If there is exactly one released versions yet, use the version SQL script as deployment script
    if len(released_versions) == 1:
        shutil.copyfile(version_file_path, deployment_file_path)
        return deployment_file_path

    # Prepare the schema version template file
    template_folder = os.path.join(Path(__file__).parent.parent, "templates")
    schema_deployment_template_file_path = os.path.join(
        template_folder, "scripts", "schema_deployment_a.b.c.sql")
    if not os.path.exists(schema_deployment_template_file_path):
        raise ValueError(
            f"The schema release version template file `{schema_deployment_template_file_path}` does not exist.")
    with open(schema_deployment_template_file_path, "r") as f:
        # Remove copyright line and replace placeholders
        schema_deployment_script = Template("".join(f.readlines()[1:]))
    schema_deployment_script = schema_deployment_script.safe_substitute({
        "license": get_license_text(project_settings=project_settings),
        "schema_name": schema_name,
        "version_target": version,
        "version_comma_str": ", ".join(str(number) for number in version_as_ints),
        "section_130_creation_of_helpers":
            target_version_sections.get("130", {}).get("sql_content", "").strip("\n"),
        "section_140_non_idempotent_schema_objects": indent(
            target_version_sections.get("140", {}).get("sql_content", "").strip("\n"), "    "),
        "section_150_idempotent_schema_objects":
            target_version_sections.get("150", {}).get("sql_content", "").strip("\n"),
        "section_170_authorization": indent(
            target_version_sections.get("170", {}).get("sql_content", "").strip("\n"), "    "),
        "section_190_removal_of_helpers":
            target_version_sections.get("190", {}).get("sql_content", "").strip("\n"),
        "updatable_versions": ", ".join(f'"{v}"' for v in updatable_versions),
    })

    matches = re.finditer(
        MSM_LOOP_UPDATABLE_VERSIONS_REGEX, schema_deployment_script, re.MULTILINE | re.DOTALL)
    # for match_id, match in enumerate(matches, start=1):
    #     print("Match {match_id} was found at {start}-{end}: {match}".format(
    #         match_id=match_id, start=match.start(), end=match.end(), match=match.group()))
    #     for group_id in range(0, len(match.groups())):
    #         group_id = group_id + 1
    #         print("Group {group_id} found at {start}-{end}: {group}".format(group_id=group_id,
    #               start=match.start(group_id), end=match.end(group_id), group=match.group(group_id)))

    for match in reversed(list(matches)):
        loop_content_template = match.group(3)
        loop_content = ""
        needs_indent = match.group(2)
        if needs_indent is None or needs_indent == "":
            needs_indent = 0
        else:
            needs_indent = int(needs_indent)

        for version_from in updatable_versions_sections:
            loop_version_content = loop_content_template
            sections_to_replace = reversed(list(
                re.finditer(MSM_SECTION_PLACEHOLDER_REGEX, loop_content_template)))
            for s in sections_to_replace:
                section_id = s.group(1)
                sql_content = updatable_versions_sections[version_from]["sections"].get(
                    section_id, {}).get("sql_content", "").strip("\n")
                sql_content_indent = indent(
                    sql_content, " " * needs_indent) if needs_indent else sql_content
                sql_content_indent += "\n"

                # If the sql_content has no SQL statements, do not insert
                if sql_content_has_no_statement(sql_content_indent):
                    sql_content_indent = ""

                loop_version_content = (
                    loop_content_template[:s.start()]
                    + sql_content_indent
                    + loop_content_template[s.end():])

            loop_content += Template(loop_version_content).safe_substitute({
                "version_from": '%d.%d.%d' % tuple(updatable_versions_sections[version_from]["version_from"]),
                "version_to": '%d.%d.%d' % tuple(updatable_versions_sections[version_from]["version_to"]),
            })

        # Check if the loop_content is empty (TODO: check if it contains no statements) and if so
        # insert a placeholder
        if len(updatable_versions_sections) > 0 and loop_content == "" and needs_indent > 0:
            loop_content = " " * needs_indent + "DO NONE;\n"

        if sql_content_has_no_statement(loop_content):
            loop_content = ""

        schema_deployment_script = (
            schema_deployment_script[:match.start()]
            + loop_content
            + schema_deployment_script[match.end():])

    # Remove empty sections
    schema_deployment_script_sections = get_script_sections(
        schema_deployment_script)
    remove_empty_sections(
        sections=schema_deployment_script_sections,
        keep_even_if_empty=["license", "003"])

    write_sections_to_file(
        file_path=deployment_file_path,
        sections=schema_deployment_script_sections,
        overwrite_existing=overwrite_existing)

    # with open(deployment_file_path, "w") as f:
    #     f.write(schema_deployment_script)

    return deployment_file_path


def get_available_licenses() -> list[str]:
    """Returns the list of available licenses

    Returns:
        The available licenses as a list of strings
    """
    template_folder = os.path.join(Path(__file__).parent.parent, "templates")
    license_path = os.path.join(template_folder, "license")

    files = []
    for file in next(os.walk(license_path))[2]:
        files.append(file[:-4])
    files.sort()

    return files


def substitute_source_statements_with_content(script: str, source_absolute_file_path: str):
    """Returns the script with substituted source statements

    Args:
        script: The script content as string
        source_absolute_file_path: The absolute path of the referenced script files

    Returns:
        The script with substituted source statements
    """

    matches = re.finditer(
        MSM_SECTION_SOURCE_REGEX, script, re.MULTILINE | re.DOTALL)
    # for match_id, match in enumerate(matches, start=1):
    #     print("Match {match_id} was found at {start}-{end}: {match}".format(
    #         match_id=match_id, start=match.start(), end=match.end(), match=match.group()))
    #     for group_id in range(0, len(match.groups())):
    #         group_id = group_id + 1
    #         print("Group {group_id} found at {start}-{end}: {group}".format(group_id=group_id,
    #               start=match.start(group_id), end=match.end(group_id), group=match.group(group_id)))

    for match in reversed(list(matches)):
        source_indention = len(match.group(1))
        source_file_path = match.group(2)
        source_slicing = match.group(3)

        # If a relative file path is defined in the SOURCE statement, convert it to an absolute path
        if not source_file_path.startswith("/"):
            source_file_path = os.path.normpath(os.path.join(
                source_absolute_file_path, source_file_path))

        if not os.path.exists(source_file_path):
            raise ValueError(
                f"The give SOURCE file path `{source_file_path}` could not be resolved.")

        with open(source_file_path, "r") as f:
            source_file_content = f.read()

            # If slicing was defined for the source content, only get the relevant substring
            if source_slicing is not None:
                if source_slicing.startswith(":"):
                    source_file_content = source_file_content[:int(
                        source_slicing[1:])]
                else:
                    slices = source_slicing.split(":")
                    source_file_content = source_file_content[int(
                        slices[0]):int(slices[1]) if slices[1] else None]

        if source_indention > 0:
            source_file_content = indent(source_file_content, source_indention * " ")

        script = (
            script[:match.start()]
            + source_file_content
            + script[match.end():])

    return script


def get_schema_name(schema_project_path: str):
    """Gets the schema name from the project settings

    Args:
        schema_project_path (str): The path to the schema project.

    Returns:
        The schema name
    """
    project_settings = get_project_settings(schema_project_path)
    schema_name = project_settings.get("schemaName", None)
    if schema_name is None:
        raise ValueError(
            "The schema name could not be read from the project settings.")

    return schema_name


def get_schema_exists(
        session: object, schema_project_path: str = None,
        schema_name: str = None) -> bool:
    """Checks whether the given schema exists

    Either the schema_project_path or the schema_name can be given.

    Args:
        session (object): The database session to use.
        schema_project_path (str): The path to the schema project.
        schema_name (str): The name of the schema

    Returns:
        True if the schema exists
    """
    if schema_name is None:
        schema_name = get_schema_name(schema_project_path)

    return (lib.core.MsmDbExec(
        'SELECT COUNT(*) as schema_count FROM information_schema.SCHEMATA '
        'WHERE SCHEMA_NAME = ?'
    ).exec(session, [schema_name]).first["schema_count"] == 1)


def get_schema_is_managed(session: object, schema_project_path: str = None,
                          schema_name: str = None) -> bool:
    """Checks whether the given schema is managed

    Either the schema_project_path or the schema_name can be given.

    Args:
        session (object): The database session to use.
        schema_project_path (str): The path to the schema project.
        schema_name (str): The name of the schema

    Returns:
        True if the schema is managed by MSM
    """
    if schema_name is None:
        schema_name = get_schema_name(schema_project_path)

    return (lib.core.MsmDbExec(
            'SELECT COUNT(*) as table_count FROM information_schema.TABLES '
            'WHERE TABLE_SCHEMA = ? AND TABLE_NAME = ? AND TABLE_TYPE = "VIEW"'
            ).exec(session, [schema_name, "msm_schema_version"])
            .first["table_count"] == 1)


def get_schema_version(session: object, schema_project_path: str = None,
                       schema_name: str = None) -> str | None:
    """Returns the current version of the schema

    Either the schema_project_path or the schema_name can be given.

    Args:
        session (object): The database session to use.
        schema_project_path (str): The path to the schema project.
        schema_name (str): The name of the schema

    Returns:
        True if the schema is managed by msm
    """
    if schema_name is None:
        schema_name = get_schema_name(schema_project_path)

    if get_schema_exists(session=session, schema_name=schema_name):
        schema_version = lib.core.MsmDbExec(
            "SELECT CONCAT(major, '.', minor, '.', patch) AS version "
            f"FROM {lib.core.quote_ident(schema_name)}.`msm_schema_version`"
        ).exec(session).first["version"]
    else:
        schema_version = None

    return schema_version


def deploy_schema(
        session: object, schema_project_path: str, version: str = None,
        backup_directory: str = None) -> str:
    """Deploys the database schema

    Deploys the given version of the database schema. If no version is given,
    the latest available version will be deployed.

    If there is an existing schema version that will be upgraded, a dump of
    that schema is created in order to be able to roll back.

    A log will be written during the update.

    Args:
        session (object): The database session to use.
        schema_project_path (str): The path to the schema project.
        version (str): The version to deploy.
        backup_directory (str): The directory to be used for backups

    Returns:
        None
    """
    project_settings = get_project_settings(schema_project_path)
    schema_name = project_settings.get("schemaName", None)
    if schema_name is None:
        err_msg = (
            f"The project settings of `{schema_project_path}` could not be "
            "read.")
        lib.core.write_to_msm_schema_update_log("ERROR", err_msg)
        raise ValueError(err_msg)
    schema_file_name = project_settings.get("schemaFileName", None)

    released_versions = get_released_versions(
        schema_project_path=schema_project_path)
    deployment_script_versions = get_deployment_script_versions(
        schema_project_path=schema_project_path)

    # Check if there are actually any released versions of the schema
    if len(released_versions) == 0:
        err_msg = (
            f"There are no versions of the schema `{schema_name}` that have "
            "been released yet.")
        lib.core.write_to_msm_schema_update_log("ERROR", err_msg)
        raise Exception(err_msg)

    # Check if there is a difference in released versions and deployment scripts
    if released_versions != deployment_script_versions:
        err_msg = (
            "Deployment script(s) missing. Please generate deployment "
            "scripts for all released versions first.")
        lib.core.write_to_msm_schema_update_log("ERROR", err_msg)
        raise Exception(err_msg)

    # If a specific version is requested, ensure that there is a deployment
    # script for this version
    if version is None:
        version = '%d.%d.%d' % tuple(deployment_script_versions[-1])
    elif version not in map(
            lambda v: '%d.%d.%d' % tuple(v), deployment_script_versions):
        err_msg = (f"Deployment or update of database schema `{schema_name}` using "
                   f"version {version} requested but there is no deployment script "
                   "available for this version.")
        lib.core.write_to_msm_schema_update_log("ERROR", err_msg)
        raise ValueError(err_msg)

    # Check if the schema already exists
    schema_exists = get_schema_exists(session=session, schema_name=schema_name)

    # Check if the schema is actually managed by MSM, and if so, get the version
    schema_managed = False
    schema_version = None
    if schema_exists:
        schema_managed = get_schema_is_managed(
            session=session, schema_name=schema_name)

        if schema_managed:
            schema_version = get_schema_version(
                session=session, schema_name=schema_name)

    if schema_exists and not schema_managed:
        err_msg = (
            f"Deployment or update of database schema `{schema_name}` using "
            f"version {version} requested but the schema is not managed by "
            "MSM.")
        lib.core.write_to_msm_schema_update_log("ERROR", err_msg)
        raise Exception(err_msg)

    # If the requested version already matches the version, exit since there is
    # nothing to do
    if schema_version is not None and schema_version == version:
        info_msg = (
            f"Deployment or update of database schema `{schema_name}` using "
            f"version {version} requested but the schema is already on the "
            "requested version. No changes performed."
        )
        lib.core.write_to_msm_schema_update_log("INFO", info_msg)
        return info_msg

    # Check if the current version of the schema is in the list of versions
    # that can be upgraded by the deployment scripts
    if schema_version is not None:
        updatable_versions = get_updatable_versions(schema_project_path)
        updatable_versions_str = map(
            lambda v: '%d.%d.%d' % tuple(v), updatable_versions)

        if (len(updatable_versions) > 0 and
                lib.core.convert_version_str_to_list(schema_version) >
                updatable_versions[-1]):
            info_msg = (
                f"The database schema `{schema_name}` is on a newer version "
                f"{schema_version} than shipped with this project (version "
                f"{'%d.%d.%d' % tuple(updatable_versions[-1])}). No changes "
                "performed."
            )
            lib.core.write_to_msm_schema_update_log("INFO", info_msg)
            return info_msg

        if not schema_version in updatable_versions_str:
            err_msg = (
                f"Update of database schema `{schema_name}` to version "
                f"{version} requested but the version {schema_version} cannot "
                "be updated.")
            lib.core.write_to_msm_schema_update_log("ERROR", err_msg)
            raise Exception(err_msg)

    # Log start of the
    if not schema_exists:
        lib.core.write_to_msm_schema_update_log(
            "INFO",
            f"Starting deployment of database schema `{schema_name}` using "
            f"version {version} ...")
    else:
        lib.core.write_to_msm_schema_update_log(
            "INFO",
            f"Starting update of database schema `{schema_name}` version "
            f"{schema_version} to version {version} ...")

    # Perform dump if the schema exists
    backup_available = False
    if schema_exists:
        lib.core.write_to_msm_schema_update_log(
            "INFO",
            f"Preparing dump of `{schema_name}` version "
            f"{schema_version} in order to be roll back in case of an error.")

        if backup_directory is None:
            backup_directory = os.path.join(
                lib.core.get_msm_plugin_data_path(),
                "backups", f"{schema_file_name}_backup_{schema_version}")
            # If that directory already exists, keep appending counter until
            # a new directory is found
            i = 2
            candidate = backup_directory
            while os.path.exists(candidate):
                candidate = f"{backup_directory}_{i}"
                i += 1
            backup_directory = candidate

        os.makedirs(backup_directory, exist_ok=True)

        # Set the mysqlsh session to the one that was given
        if "shell.Object" in str(type(session)):
            mysqlsh.globals.shell.set_session(session)
        else:
            mysqlsh.globals.shell.set_session(session.session)

        # Ensure that the dump can be read back in case of a failure by setting
        # local_infile to 1
        # cSpell:ignore infile
        row = lib.core.MsmDbExec("SELECT @@local_infile as local_infile").exec(session).first
        original_local_infile = (row and int(row["local_infile"]) == 1)
        if not original_local_infile:
            try:
                lib.core.write_to_msm_schema_update_log(
                    "INFO",
                    "Enabling local_infile option in order to be able to load "
                    "back the schema dump in case of an update error...")
                lib.core.MsmDbExec("SET GLOBAL local_infile=1").exec(session)
            except:
                err_msg = (
                    "Failed to enable the local_infile option. Please execute "
                    "SET PERSIST GLOBAL local_infile=1; on the MySQL Server.")
                lib.core.write_to_msm_schema_update_log("ERROR", err_msg)

                raise Exception(err_msg)

        lib.core.write_to_msm_schema_update_log(
            "INFO",
            f"Creating dump of `{schema_name}` version {schema_version} ...")
        mysqlsh.globals.util.dump_schemas(
            [schema_name],
            f"file://{backup_directory}",
            {
                "skipUpgradeChecks": True,
                "showProgress": False,
            })

        backup_available = True

    # Run deployment script
    try:
        lib.core.execute_msm_sql_script(
            session=session,
            sql_file_path=os.path.join(
                schema_project_path, "releases", "deployment",
                f"{schema_file_name}_deployment_{version}.sql"))

        if not schema_exists:
            info_msg = (
                f"Deployment of `{schema_name}` version "
                f"{version} completed successfully.")
        else:
            if not original_local_infile:
                lib.core.MsmDbExec("SET GLOBAL local_infile=0").exec(session)
                lib.core.write_to_msm_schema_update_log(
                    "INFO",
                    "Restored local_infile option.")

            info_msg = (
                f"Completed the update of `{schema_name}` version "
                f"{schema_version} to {version} successfully.")

        lib.core.write_to_msm_schema_update_log("INFO", info_msg)

        # Remove the backup directory as it is no longer needed
        if backup_available:
            shutil.rmtree(backup_directory)

        return info_msg
    except Exception as e:
        # Drop the schema after failed update
        try:
            lib.core.MsmDbExec(
                f"DROP SCHEMA IF EXISTS {lib.core.quote_ident(schema_name)}"
            ).exec(session)
        except:
            pass

        # Restore the backup if available
        if backup_available:
            try:
                mysqlsh.globals.util.load_dump(
                    f"file://{backup_directory}",
                    {
                        "showMetadata": False,
                        "showProgress": False,
                        "ignoreVersion": True,
                    })

                if not original_local_infile:
                    lib.core.MsmDbExec("SET GLOBAL local_infile=0").exec(session)
                    lib.core.write_to_msm_schema_update_log(
                        "INFO",
                        "Restored local_infile option.")

                # Remove the backup directory as it is no longer needed
                shutil.rmtree(backup_directory)
            except Exception as e_dump_load:
                err_str = (
                    "An error occurred while updating the database schema "
                    f"`{schema_name}` to version {version}. The schema could "
                    f"not be restored back to version {schema_version}. {e} "
                    f"{e_dump_load}")
                lib.core.write_to_msm_schema_update_log("ERROR", err_str)

                raise Exception(err_str)

            err_str = (
                "An error occurred while updating the database schema "
                f"`{schema_name}` to version {version}. The schema has been "
                f"restored back to version {schema_version}. {e}"
            )
            lib.core.write_to_msm_schema_update_log("ERROR", err_str)

            raise Exception(err_str)
        else:
            if not schema_exists:
                err_str = (
                    f"Deploying the database schema `{schema_name}` failed. {e}"
                )
                lib.core.write_to_msm_schema_update_log("ERROR", err_str)

                raise Exception(err_str)
            else:
                err_str = (
                    "An error occurred while updating the database schema "
                    f"`{schema_name}` to version {version}. {e}"
                )
                lib.core.write_to_msm_schema_update_log("ERROR", err_str)

                raise Exception(err_str)


def execute_msm_sql_script(session, sql_script: str = None,
                           script_name: str = None, sql_file_path: str = None):
    """
    Execute a SQL script on the MySQL Server and log the progress in the
    MSM log file.

    The SQL script can be provided as a string, or as a path to a file.

    Args:
        session (object): The MySQL session to use.
        sql_script (str): The SQL script to execute.
        script_name (str): The name of the SQL script.
        sql_file_path (str): The path to the SQL script file.

    Returns:
        None
    """

    lib.core.execute_msm_sql_script(
        session, sql_script=sql_script,
        script_name=script_name, sql_file_path=sql_file_path)