git-flow/src/git_flow/git.py

import subprocess

from git_flow.io import *


FLOWCONFIG_FILE = ".flowconfig"

CHANGELOG_FILE = "CHANGELOG.md"

BUMP_VERSION = "chore: bump version and update CHANGELOG.md [skip ci]"


class Git:
    class Status:
        worktree: str
        index: str
        file: str

        def __init__(self, line: str) -> None:
            self.worktree = line[0]
            self.index = line[1]
            self.file = line[3:].strip("\n")

    command: list[str]

    @staticmethod
    def get_config(file: str | None = None):
        config: dict[str, str] = {}
        lines = Git("config", file=file, list=True).lines()

        for line in lines:
            pos = line.index("=")
            key = line[:pos]
            value = line[pos + 1 :]
            config[key] = value

        return config

    @staticmethod
    def set_config(config: dict[str, str], file: str | None = None):
        for key, value in config:
            Git("config", key, value, file=file).exec()

    @staticmethod
    def flow_config(*args, **kwargs):
        return Git("config", *args, file=FLOWCONFIG_FILE, **kwargs)

    @staticmethod
    def get_current_branch():
        return Git("rev-parse", "HEAD", abbrev_ref=True).firstline()

    @staticmethod
    def get_branches():
        return Git._get_references("heads")

    @staticmethod
    def status():
        return list(map(Git.Status, Git("status", porcelain=True).lines(False)))

    @staticmethod
    def get_current_tag():
        return Git("describe", abbrev="0", tags=True).firstline(check=False)

    @staticmethod
    def get_first_fork_point(branch: str, env: str):
        boundary_commits = list(
            filter(
                lambda c: c.startswith("-"),  # boundary commits
                Git(
                    "rev-list", env + "..." + branch, topo_order=True, boundary=True
                ).lines(),
            )
        )

        if not boundary_commits:
            raise RuntimeError("No se encontro un commit base para realizar el rebase.")

        return boundary_commits[-1].removeprefix("-")

    @staticmethod
    def _get_references(kind: str):
        prefix = "refs/" + kind + "/"
        return list(map(
            lambda r: r.removeprefix(prefix),
            Git("for-each-ref", prefix + "*", format="%(refname)").lines(),
        ))

    @staticmethod
    def is_repository() -> bool:
        return Git("rev-parse", is_inside_work_tree=True).code() == 0

    def __init__(self, subcommand: str, *args: str, **kwargs: str | int | bool | list[str] | None) -> None:
        self.command = ["git", subcommand]

        for option, value in kwargs.items():
            option = option.replace("_", "-")

            if isinstance(value, bool):
                if len(option) == 1:
                    self.command.append("-" + option)
                else:
                    prefix = "--" if value else "--no-"
                    self.command.append(prefix + option)
            elif value is None:
                continue
            elif isinstance(value, list):
                for v in value:
                    self.command.append("--" + option + "=" + v)
            else:
                value = value if isinstance(value, str) else str(value)

                if len(option) == 1:
                    self.command.append("-" + option)
                    self.command.append(value)
                else:
                    self.command.append("--" + option + "=" + value)

        self.command += args

    def lines(self, strip: bool = True, **kwargs):
        process = self._get(**kwargs)
        lines = process.stdout.splitlines()

        return lines if not strip else list(map(lambda l: l.strip(), lines))

    def firstline(self, **kwargs: bool):
        lines = self.lines(**kwargs)

        return lines[0] if lines else ""

    def code(self, **kwargs):
        return self._run(check=False, **kwargs).returncode

    def exec(self, **kwargs):
        self._run(**kwargs)

    def _get(self, print: str|None = None, check: bool = True):
        process = subprocess.run(self.command, capture_output=True, text=True)

        if print is not None:
            self._print_process(process.stdout, process.stderr, print)

        if check:
            process.check_returncode()

        return process


    def _run(self, print: str|None = None, check: bool = True):
        process = subprocess.run(self.command, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)

        if print is not None:
            self._print_process("", "", title=print)

        if check:
            process.check_returncode()

        return process

    def _print_process(self, stdout: str, stderr: str, title: str):
        command = []

        for arg in self.command:
            if arg.startswith("--") and "=" in arg:
                pos = arg.index("=")
                arg = arg[: pos + 1] + '"' + arg[pos + 1 :] + '"'
            elif " " in arg:
                arg = '"' + arg + '"'

            command.append(arg)

        title = title or "Ejecutando"
        print(COLOR_YELLOW + "| " + title + COLOR_RESET)
        print(COLOR_YELLOW + "| $ " + " ".join(command) + COLOR_RESET)

        for line in stdout.splitlines():
            print(COLOR_YELLOW + "| " + COLOR_RESET + "[out] " + line)

        for line in stderr.splitlines():
            print(COLOR_YELLOW + "| " + COLOR_RED + "[err] " + line + COLOR_RESET)