refactor: migrate from class commands to scripts commands

2026-05-17 11:19:30 -03:00
parent 9ffefafad0
commit 0ac04aee84
10 changed files with 775 additions and 812 deletions
@@ -1,16 +1,5 @@
-from argparse import Namespace, ArgumentParser
-from abc import ABC, abstractmethod
-from os.path import isfile
-import rich
-from rich.prompt import Prompt, Confirm
-
 from git_flow import (
    BRANCH_TYPES,
-    COLOR_BLUE,
-    COLOR_GREEN,
-    COLOR_RED,
-    COLOR_RESET,
-    COLOR_YELLOW,
    REPOSITORY_TOKEN_FILENAME,
    GitFlowError,
 )
@@ -18,187 +7,160 @@ from git_flow.git import FLOWCONFIG_FILE, Git
 from git_flow.remote.base import RemoteAPI
 from git_flow.remote.bitbucket import BitbucketRemoteAPI
 from git_flow.remote.github import GithubRemoteAPI
+from os.path import isfile
+from rich.console import Console
+from rich.prompt import Prompt, Confirm
+from rich.panel import Panel

 TYPE_SUCCESS = 0
 TYPE_WARNING = 1
 TYPE_ERROR = 2
 TYPE_INFO = 3

+console = Console(highlight=False)
+flowconfig = Git.get_config(FLOWCONFIG_FILE) if isfile(FLOWCONFIG_FILE) else {}

-class Command(ABC):
-    flowconfig: dict[str, str]

-    @abstractmethod
-    def name(self) -> str:
-        pass
+def success(msg: str):
+    console.print(f":white_check_mark: [green]{msg}[/green]")

-    @abstractmethod
-    def description(self) -> str:
-        pass

-    @abstractmethod
-    def run(self, args: Namespace = Namespace()):
-        pass
+def warning(msg: str):
+    console.print(f":warning: [yellow]{msg}[/yellow]")

-    def init(self):
-        self.flowconfig = (
-            Git.get_config(FLOWCONFIG_FILE) if isfile(FLOWCONFIG_FILE) else {}
+
+def error(msg: str):
+    console.print(f":x: [red]{msg}[/red]")
+
+
+def info(msg: str):
+    console.print(f":information: [blue]{msg}[/blue]")
+
+
+def panel(title: str, text: str):
+    console.print(Panel(text, title=title, expand=False, title_align="left"))
+
+
+def prompt(
+    prompt: str,
+    default: str | None = None,
+    persistent: bool = False,
+    strip: bool = True,
+) -> str:
+    onetime = not persistent
+
+    while onetime or persistent:
+        result = Prompt.ask(prompt)
+        result = result.strip() if strip else result
+
+        if result:
+            return result
+        elif default is not None:
+            return default
+        elif persistent:
+            error("Debe ingresar un valor no vacío.")
+        else:
+            return result
+
+
+def confirm(question: str, default: bool = True):
+    return Confirm.ask(question, default=default)
+
+
+def choice(prompt: str, options: list[str]):
+    return Prompt.ask(prompt, choices=options)
+
+
+def ensure_initialized():
+    initialized = flowconfig["flow.initialized"] if flowconfig else None
+
+    if not initialized:
+        raise GitFlowError(
+            "El repositorio no fue inicializado, debe ejecutar el comando `init`."
+        )
+    elif initialized != "true":
+        raise GitFlowError(
+            f"El valor de `flow.initialized` ({initialized}) es inválido."
        )

-    def success(self, msg: str):
-        rich.print(f":white_check_mark: [green]{msg}[/green]")

-    def warning(self, msg: str):
-        rich.print(f":warning: [yellow]{msg}[/yellow]")
+def ensure_right_branch():
+    branch = Git.get_current_branch()

-    def error(self, msg: str):
-        rich.print(f":x: [red]{msg}[/red]")
+    if branch == "HEAD":
+        raise GitFlowError("No se encuentra parado sobre una rama.")
+    elif confirm(f"Se encuentra sobre la rama '{branch}'. ¿Es correcto?"):
+        return branch
+    else:
+        raise GitFlowError("Ejecución cancelada.")

-    def info(self, msg: str):
-        rich.print(f":information: [blue]{msg}[/blue]")

-    def prompt(
-        self,
-        prompt: str,
-        default: str | None = None,
-        persistent: bool = False,
-        strip: bool = True,
-    ) -> str:
-        onetime = not persistent
+def get_branch_env_and_type(branch: str) -> tuple[str, str]:
+    components = branch.split("/")
+    target_branches = flowconfig["flow.branches"].split(",")

-        while onetime or persistent:
-            result = Prompt.ask(prompt)
-            result = result.strip() if strip else result
-
-            if result:
-                return result
-            elif default is not None:
-                return default
-            elif persistent:
-                self.error("Debe ingresar un valor no vacío.")
-            else:
-                return result
-
-    def confirm(self, question: str, default: bool = True):
-        return Confirm.ask(question, default=default)
-
-    def choice(self, prompt: str, options: list[str]):
-        return Prompt.ask(prompt, choices = options)
-
-    def setup_parser(self, parser: ArgumentParser) -> ArgumentParser:
-        return parser
-
-    def ensure_initialized(self):
-        initialized = self.flowconfig["flow.initialized"] if self.flowconfig else None
-
-        if not initialized:
+    if len(components) == 2:
+        if components[0] not in BRANCH_TYPES:
            raise GitFlowError(
-                "El repositorio no fue inicializado, debe ejecutar el comando `init`."
-            )
-        elif initialized != "true":
-            raise GitFlowError(
-                f"El valor de `flow.initialized` ({initialized}) es inválido."
+                f"Branch inválida, el tipo '{components[0]}' no es válido."
            )

-    def ensure_right_branch(self):
-        branch = Git.get_current_branch()
+        return (target_branches[0], components[0])
+    elif len(components) == 4:
+        target_branches = target_branches[1:]

-        if branch == "HEAD":
-            raise GitFlowError("No se encuentra parado sobre una rama.")
-        elif self.confirm(f"Se encuentra sobre la rama '{branch}'. ¿Es correcto?"):
-            return branch
-        else:
-            raise GitFlowError("Ejecución cancelada.")
-
-    def ensure_repository_token(self):
-        if not isfile(REPOSITORY_TOKEN_FILENAME):
+        if (
+            components[0] != "release"
+            or components[1] not in target_branches
+            or components[2] not in BRANCH_TYPES
+        ):
            raise GitFlowError(
-                "No existe un token para el repositorio en " + REPOSITORY_TOKEN_FILENAME
+                "Branch release inválido, debe tener el siguiente formato: "
+                "release/<env>/<type>/<name>, pero es: " + branch
            )

-        with open(REPOSITORY_TOKEN_FILENAME) as f:
-            return f.readline().strip()
+        return (components[1], components[2])
+    else:
+        raise GitFlowError("Branch inválido: " + branch)

-    def ensure_clean_worktree(self, has_remote: bool):
-        status = Git.status()

-        if not status:
-            return
+def ensure_repository_token():
+    if not isfile(REPOSITORY_TOKEN_FILENAME):
+        raise GitFlowError(
+            "No existe un token para el repositorio en " + REPOSITORY_TOKEN_FILENAME
+        )

-        not_empty = any(map(lambda s: s.index != " " or s.worktree != " ", status))
+    with open(REPOSITORY_TOKEN_FILENAME) as f:
+        return f.readline().strip()

-        if not_empty:
-            if not has_remote:
-                self.warning("Existen cambios en tu entorno de trabajo sin commitear.")
-            else:
-                raise GitFlowError("No se puede continuar con cambios pendientes.")

-            if not self.confirm("¿Desea continuar?", False):
-                raise GitFlowError("Ejecución abortada")
+def ensure_clean_worktree(has_remote: bool):
+    status = Git.status()

-    def get_branch_env_and_type(self, branch: str) -> tuple[str, str]:
-        components = branch.split("/")
-        target_branches = self.flowconfig["flow.branches"].split(",")
+    if not status:
+        return

-        if len(components) == 2:
-            if components[0] not in BRANCH_TYPES:
-                raise GitFlowError(
-                    f"Branch inválida, el tipo '{components[0]}' no es válido."
-                )
+    not_empty = any(map(lambda s: s.index != " " or s.worktree != " ", status))

-            return (target_branches[0], components[0])
-        elif len(components) == 4:
-            target_branches = target_branches[1:]
-
-            if (
-                components[0] != "release"
-                or components[1] not in target_branches
-                or components[2] not in BRANCH_TYPES
-            ):
-                raise GitFlowError(
-                    "Branch release inválido, debe tener el siguiente formato: "
-                    "release/<env>/<type>/<name>, pero es: " + branch
-                )
-
-            return (components[1], components[2])
+    if not_empty:
+        if not has_remote:
+            warning("Existen cambios en tu entorno de trabajo sin commitear.")
        else:
-            raise GitFlowError("Branch inválido: " + branch)
+            raise GitFlowError("No se puede continuar con cambios pendientes.")

-    def get_remote_api(self, token: str) -> RemoteAPI:
-        if "flow.remote" not in self.flowconfig:
-            raise GitFlowError("El repositorio no tiene configurado un remoto.")
+        if not confirm("¿Desea continuar?", False):
+            raise GitFlowError("Ejecución abortada")

-        [host, repository] = RemoteAPI.parse(self.flowconfig["flow.remote"])

-        if host == "bitbucket.org":
-            return BitbucketRemoteAPI(repository, token)
-        elif host == "github.com":
-            return GithubRemoteAPI(repository, token)
-        else:
-            raise GitFlowError("El host del repositorio remoto es inválido")
+def get_remote_api(token: str) -> RemoteAPI:
+    if "flow.remote" not in flowconfig:
+        raise GitFlowError("El repositorio no tiene configurado un remoto.")

-    def _print(self, type: int, msg: str):
-        print(self._get_color(type) + self._get_tag(type) + " " + msg + COLOR_RESET)
+    [host, repository] = RemoteAPI.parse(flowconfig["flow.remote"])

-    def _get_color(self, type: int) -> str:
-        if type == TYPE_SUCCESS:
-            return COLOR_GREEN
-        elif type == TYPE_WARNING:
-            return COLOR_YELLOW
-        elif type == TYPE_ERROR:
-            return COLOR_RED
-        elif type == TYPE_INFO:
-            return COLOR_BLUE
-        return ""
-
-    def _get_tag(self, type: int) -> str:
-        if type == TYPE_SUCCESS:
-            return "[success]"
-        elif type == TYPE_WARNING:
-            return "[warning]"
-        elif type == TYPE_ERROR:
-            return "[error]"
-        elif type == TYPE_INFO:
-            return "[info]"
-
-        return ""
+    if host == "bitbucket.org":
+        return BitbucketRemoteAPI(repository, token)
+    elif host == "github.com":
+        return GithubRemoteAPI(repository, token)
+    else:
+        raise GitFlowError("El host del repositorio remoto es inválido")