Modify MySQL config files on MacOS as installed by Homebrew to alllow simultaneous 5.7 and 8.0 installations

macos_modify_mysql_configs.py

import configparser
import plistlib
import shutil
import subprocess
from enum import Enum
from pathlib import Path
from string import Template
from typing import NamedTuple, Optional

EXEC_START_TMPL = Template("/opt/homebrew/opt/mysql@${mysql_ver}/bin/mysqld_safe")
DEFAULTS_FILE_TMPL = Template(
    "--defaults-file=/opt/homebrew/etc/my.cnf.d/mysql${mysql_ver}.cnf"
)
DATADIR_TMPL = Template("--datadir=/opt/homebrew/var/mysql$mysql_ver")
CELLAR_PATH = Path("/opt/homebrew/Cellar/")


class ConfigFileType(str, Enum):
    PLIST = "PLIST"
    SERVICE = "SERVICE"


class ConfigFileVersion(str, Enum):
    FIVE_SEVEN = "5.7"
    EIGHT_ZERO = "8.0"


class ConfigFile(NamedTuple):
    path: Path
    type: ConfigFileType
    version: ConfigFileVersion


def make_opts_list(version_prefix: ConfigFileVersion) -> list[str]:
    return [
        DEFAULTS_FILE_TMPL.substitute(mysql_ver=version_prefix.value.replace(".", "")),
        DATADIR_TMPL.substitute(mysql_ver=version_prefix.value.replace(".", "")),
    ]


def get_version_path(version_prefix: ConfigFileVersion) -> Optional[Path]:
    for _, dirname, _ in CELLAR_PATH.walk(top_down=True):
        if len(dirname) == 1 and dirname[0].startswith(f"mysql@{version_prefix.value}"):
            try:
                return next((CELLAR_PATH / Path(dirname[0])).iterdir())
            except StopIteration:
                continue
    return None


def get_file_paths(version_prefix: ConfigFileVersion) -> Optional[list[ConfigFile]]:
    ret: list[ConfigFile] = []
    if not (dir_path := get_version_path(version_prefix)):
        return None
    try:
        if plist_path := next(dir_path.glob("*.plist")):
            ret.append(
                ConfigFile(
                    path=plist_path, type=ConfigFileType.PLIST, version=version_prefix
                )
            )
    except StopIteration:
        pass
    try:
        if svc_path := next(dir_path.glob("*.service")):
            ret.append(
                ConfigFile(
                    path=svc_path, type=ConfigFileType.SERVICE, version=version_prefix
                )
            )
    except StopIteration:
        pass

    return ret


def update_plist(plist_path: Path, version_prefix: ConfigFileVersion) -> bool:
    with open(plist_path, "rb") as f:
        plist = plistlib.load(f)
    try:
        del plist["ProgramArguments"][
            plist["ProgramArguments"].index(DATADIR_TMPL.substitute(mysql_ver=""))
        ]
    except KeyError as e:
        print(e)
        return False
    except ValueError:
        pass
    _file_version_prefix = version_prefix.value.replace(".", "")
    for opt in make_opts_list(version_prefix=version_prefix):
        if opt not in plist["ProgramArguments"]:
            plist["ProgramArguments"].append(opt)
    if not plist["WorkingDirectory"].endswith(_file_version_prefix):
        plist["WorkingDirectory"] = f"{plist['WorkingDirectory']}{_file_version_prefix}"
    with open(plist_path, "wb") as f:
        return f.write(plistlib.dumps(plist)) > 0


def update_svc(svc_path: Path, version_prefix: ConfigFileVersion) -> bool:
    config = configparser.ConfigParser()
    config.optionxform = str  # type: ignore[assignment]
    config.read(svc_path)
    _file_version_prefix = version_prefix.value.replace(".", "")
    if "Service" not in config.sections():
        return False
    opts_list = make_opts_list(version_prefix)
    opts_string = " ".join(opts_list).replace("=", r"\=")
    config["Service"]["ExecStart"] = (
        f"{EXEC_START_TMPL.substitute(mysql_ver=version_prefix.value)} {opts_string}"
    )
    if not config["Service"]["WorkingDirectory"].endswith(_file_version_prefix):
        config["Service"]["WorkingDirectory"] = (
            f"{config['Service']['WorkingDirectory']}{_file_version_prefix}"
        )
    with open(svc_path, "w") as f:
        config.write(f, space_around_delimiters=False)

    return True


if __name__ == "__main__":
    if (
        (_confirm := input("Stop MySQL 5.7 and 8.0 locally? (y/n) "))
        .lower()
        .startswith("n")
    ):
        print("INFO: user declined to continue, exiting")
        raise SystemExit(0)
    try:
        for mysql_ver in ("5.7", "8.0"):
            print(f"INFO: stopping mysql@{mysql_ver}")
            subprocess.run(
                ["brew", "services", "stop", f"mysql@{mysql_ver}"],
                check=True,
                stdout=subprocess.DEVNULL,
            )
            print(f"INFO: mysql@{mysql_ver} stopped")
    except subprocess.CalledProcessError:
        print(f"ERROR: failed to stop local MySQL {mysql_ver}")
        raise SystemExit(1)
    config_files: list[ConfigFile] = []
    for ver in (ConfigFileVersion.FIVE_SEVEN, ConfigFileVersion.EIGHT_ZERO):
        if (_files := get_file_paths(version_prefix=ver)) is None:
            print("ERROR: expected files not found, exiting")
            raise SystemExit(1)
        config_files.extend(_files)
    for file in config_files:
        shutil.copy2(file.path, file.path.with_suffix(".bak"))
        if file.type is ConfigFileType.PLIST:
            if not update_plist(plist_path=file.path, version_prefix=file.version):
                print(f"ERROR: failed to update .plist file at {file.path}")
                print("INFO: restoring backup of file, then exiting")
                shutil.move(
                    file.path.with_suffix(".bak"),
                    file.path.with_suffix(".plist"),
                )
                raise SystemExit(1)
        elif file.type is ConfigFileType.SERVICE:
            if not update_svc(svc_path=file.path, version_prefix=file.version):
                print(f"ERROR: failed to update .service file at {file.path}")
                print("INFO: restoring backup of file, then exiting")
                shutil.move(
                    file.path.with_suffix(".bak"),
                    file.path.with_suffix(".service"),
                )
                raise SystemExit(1)
    print("INFO: successfully modified files, restarting MySQL 5.7 and 8.0")
    try:
        for mysql_ver in ("5.7", "8.0"):
            print(f"INFO: starting mysql@{mysql_ver}")
            subprocess.run(
                ["brew", "services", "run", f"mysql@{mysql_ver}"],
                check=True,
                stdout=subprocess.DEVNULL,
            )
            print(f"INFO: mysql@{mysql_ver} running")
    except subprocess.CalledProcessError:
        print(f"ERROR: failed to start MySQL {mysql_ver}")
        raise SystemExit(1)

    if (_confirm := input("Delete backup files? (y/n) ")).lower().startswith("y"):
        for file in config_files:
            file.path.with_suffix(".bak").unlink(missing_ok=True)
        print("INFO: all backup files deleted")
    else:
        print("INFO: all backup files maintained - manually clean them up if desired")

    print("\nBye!")