diff --git a/src/git_flow/command/base.py b/src/git_flow/command/base.py index b3ef583..398cd1d 100644 --- a/src/git_flow/command/base.py +++ b/src/git_flow/command/base.py @@ -3,6 +3,7 @@ from git_flow import ( REPOSITORY_TOKEN_FILENAME, FLOWCONFIG_FILENAME, FLOWCONFIG_VERSION, + COMMIT_TYPES, GitFlowError, ) from git_flow.git import Git @@ -18,9 +19,11 @@ import rich import rich.panel import questionary - flowconfig = Git.get_config(FLOWCONFIG_FILENAME) if isfile(FLOWCONFIG_FILENAME) else {} -environments = flowconfig["flow.branches"].split(",") if "flow.branches" in flowconfig else [] +environments = ( + flowconfig["flow.branches"].split(",") if "flow.branches" in flowconfig else [] +) + def ensure_initialized(): initialized = flowconfig["flow.initialized"] if flowconfig else None @@ -48,7 +51,7 @@ def ensure_right_branch(): if branch == "HEAD": raise GitFlowError("No se encuentra parado sobre una rama.") - elif io_confirm(f"Se encuentra sobre la rama '{branch}'. ¿Es correcto?"): + elif io_confirm(f"Rama actual: {branch}. ¿Es correcto?"): return branch else: raise GitFlowError("Ejecución cancelada.") @@ -103,12 +106,12 @@ def ensure_clean_worktree(has_remote: bool): if not_empty: if not has_remote: io_warning("Existen cambios en tu entorno de trabajo sin commitear.") + + if not io_confirm("¿Desea continuar?", False): + raise GitFlowError("Ejecución abortada") else: raise GitFlowError("No se puede continuar con cambios pendientes.") - if not io_confirm("¿Desea continuar?", False): - raise GitFlowError("Ejecución abortada") - def get_remote_api(token: str) -> RemoteAPI: if "flow.remote" not in flowconfig: @@ -118,7 +121,8 @@ def get_remote_api(token: str) -> RemoteAPI: [schema, host, repository] = RemoteAPI.parse(flowconfig["flow.remote"]) if remote_type == "gitea": - return GiteaRemoteAPI(("https://" if schema == "ssh://" else schema) + host, repository, token) + remote_schema = "https://" if schema == "ssh://" else schema + return GiteaRemoteAPI(remote_schema + host, repository, token) elif remote_type == "bitbucket": return BitbucketRemoteAPI(repository, token) elif remote_type == "github": @@ -133,40 +137,85 @@ def is_valid_ticket(ticket: str) -> bool: if len(components) != 2: return False - + ticket_project = components[0] ticket_number = components[1] - return ticket_project.isalpha() and ticket_project.isupper() and ticket_number.isnumeric() + return ( + ticket_project.isalpha() + and ticket_project.isupper() + and ticket_number.isnumeric() + ) + + +def get_commit_message(commit_types: list[str] = COMMIT_TYPES) -> str: + commit_type = io_choice("Tipo de commit", commit_types) + commit_message = io_prompt( + "Mensaje de commit", + validator=lambda s: 0 < len(s) and len(s) < 100, + instruction="100 caracteres máximo", + ) + return commit_type + ": " + commit_message + def io_error(message: str, title: Optional[str] = None): - _io_status(message, "❌ Error: ", "red", title) + _io_status(message, "✖ Error: ", "red", title) + def io_warning(message: str, title: Optional[str] = None): - _io_status(message, "⚠️ Warning: ", "yellow", title) + _io_status(message, "⚠ Warning: ", "yellow", title) + def io_info(message: str, title: Optional[str] = None): - _io_status(message, "ℹ️ Info: ", "blue", title) + _io_status(message, "ℹ Info: ", "blue", title) + def io_success(message: str, title: Optional[str] = None): - _io_status(message, "✅ Success: ", "green", title) + _io_status(message, "✔ Success: ", "green", title) + def _io_status(message: str, prefix: str, style: str, title: Optional[str] = None): if title: - rich.print(rich.panel.Panel(message, title=prefix + title, style=style, expand=False, title_align="left")) + rich.print( + rich.panel.Panel( + message, + title=prefix + title, + style=style, + expand=False, + title_align="left", + ) + ) else: rich.print(f"[{style}]{prefix}{message}[/{style}]") - + + def io_confirm(question: str, default: bool = True) -> bool: - return questionary.confirm(question, default=default, auto_enter=False).ask() + return questionary.confirm(question, default=default, auto_enter=False).unsafe_ask() + def io_choice(prompt: str, options: list[str]) -> str: - return questionary.select(prompt, options, instruction="Usar flechas").ask() + return questionary.select( + prompt, + options, + use_search_filter=True, + use_jk_keys=False, + instruction="Tipear o usar flechas", + ).unsafe_ask() -def io_prompt(message: str, default: str = "", persistent: bool = False, strip: bool = False, validator = None) -> str: + +def io_prompt( + message: str, + default: str = "", + persistent: bool = False, + strip: bool = False, + validator=None, + instruction: str | None = None, +) -> str: if persistent and not validator: validator = lambda s: len(s.strip()) > 0 - value = questionary.text(message, default, validate=validator).ask() + value = questionary.text( + message, default, validate=validator, instruction=instruction + ).unsafe_ask() - return value.strip() if strip else value \ No newline at end of file + return value.strip() if strip else value diff --git a/src/git_flow/command/branch.py b/src/git_flow/command/branch.py index ba1e9e3..1c78dc3 100644 --- a/src/git_flow/command/branch.py +++ b/src/git_flow/command/branch.py @@ -1,16 +1,20 @@ +from typing import Annotated + +import typer +from rich.console import Console + +import git_flow.command.base as base from git_flow import ( TRASH_BRANCH_PREFIX, WIP_BRANCH_PREFIX, GitFlowError, ) -from git_flow.command.base import * from git_flow.git import Git -from typing import Annotated -import typer BRANCH_FORMAT = "%(refname:short)" app = typer.Typer() +console = Console() @app.command() @@ -38,28 +42,28 @@ def branch( ] = False, ): """Lista ramas del repositorio, agrupandolas por entorno objetivo""" - ensure_initialized() + base.ensure_initialized() current_branch = Git.get_current_branch() - if current_branch in environments: + if current_branch in base.environments: current_environment = current_branch else: - current_environment, _ = get_branch_env_and_type(current_branch) + current_environment, _ = base.get_branch_env_and_type(current_branch) if trash: show_branches(TRASH_BRANCH_PREFIX, current_branch) elif wip: show_branches(WIP_BRANCH_PREFIX, current_branch) elif all: - show_envs(environments, current_branch) - show_all_branches(environments, current_branch) + show_envs(base.environments, current_branch) + show_all_branches(base.environments, current_branch) elif environment is None: - show_envs(environments, current_branch) - show_env_branches(current_environment, environments, current_branch) - elif environment in environments: - show_envs(environments, current_branch) - show_env_branches(environment, environments, current_branch) + show_envs(base.environments, current_branch) + show_env_branches(current_environment, base.environments, current_branch) + elif environment in base.environments: + show_envs(base.environments, current_branch) + show_env_branches(environment, base.environments, current_branch) else: raise GitFlowError(f"'{environment}' no es un entorno válido.") @@ -78,7 +82,9 @@ def show_branches(branch_prefix: str, current_branch: str): def get_branches(branch_prefix: str): - return Git("for-each-ref", "refs/heads/" + branch_prefix, format=BRANCH_FORMAT).lines() + return Git( + "for-each-ref", "refs/heads/" + branch_prefix, format=BRANCH_FORMAT + ).lines() def show_all_branches(environments: list[str], current_branch: str): diff --git a/src/git_flow/command/commit.py b/src/git_flow/command/commit.py index 79ae2c6..a7834d6 100644 --- a/src/git_flow/command/commit.py +++ b/src/git_flow/command/commit.py @@ -1,8 +1,10 @@ -import typer from datetime import datetime -from git_flow import COMMIT_TYPES, TRASH_BRANCH_PREFIX, WIP_BRANCH_PREFIX, GitFlowError -from git_flow.git import Git + +import typer + import git_flow.command.base as base +from git_flow import TRASH_BRANCH_PREFIX, WIP_BRANCH_PREFIX, GitFlowError +from git_flow.git import Git app = typer.Typer() @@ -20,11 +22,8 @@ def commit(): else: raise GitFlowError("Debe agregar algún cambio al indice para continuar.") - commit_type = base.io_choice("Tipo de commit", COMMIT_TYPES) - commit_message = base.io_prompt( - "Mensaje (máximo recomendado: 100 caracteres)", persistent=True - ) - message = commit_type + ": " + commit_message + message = base.get_commit_message() + commit_type = message[: message.index(":")] if branch.startswith(WIP_BRANCH_PREFIX): original_branch = branch.removeprefix(WIP_BRANCH_PREFIX) diff --git a/src/git_flow/command/init.py b/src/git_flow/command/init.py index f60b450..24db201 100644 --- a/src/git_flow/command/init.py +++ b/src/git_flow/command/init.py @@ -1,8 +1,8 @@ import typer +import git_flow.command.base as base from git_flow import FLOWCONFIG_FILENAME, FLOWCONFIG_VERSION, GitFlowError from git_flow.git import Git -from git_flow.command.base import * app = typer.Typer() @@ -15,7 +15,7 @@ def init(): _ensure_not_already_initialized() branches = _setup_flow_branches() - (remote, remote_type) = _setup_flow_remote() + remote, remote_type = _setup_flow_remote() flowconfig = { "flow.version": str(FLOWCONFIG_VERSION), @@ -36,32 +36,34 @@ def init(): def _ensure_is_repository(): if not Git.is_repository(): - warning("El directorio actual no es un repositorio.") + base.io_warning("El directorio actual no es un repositorio.") - if not confirm("¿Desea inicializarlo?"): + if not base.io_confirm("¿Desea inicializarlo?"): raise GitFlowError("No se puede continuar sin inicializar el repositorio") Git("init").exec() def _ensure_not_already_initialized(): - if not flowconfig: + if not base.flowconfig: return - elif flowconfig.get("flow.initialized") == "true": + elif base.flowconfig.get("flow.initialized") == "true": raise GitFlowError("El repositorio ya fue inicializado para usar git-flow") else: raise GitFlowError("Valor de 'flow.initialized' es inválido") def _setup_flow_branches(): - panel( - "Configuración de entornos", + base.io_info( """Ingrese las ramas que representan los entornos de deploy del proyecto en orden creciente de cercanía al entorno productivo, y separados por coma. Por ejemplo: "dev, test, prod".""", + title="Configuración de entornos", ) - branches = list(map(lambda b: b.strip(), prompt("Ramas", "main").split(","))) + branches = list( + map(lambda b: b.strip(), base.io_prompt("Ramas", "main").split(",")) + ) if not all(branches): raise GitFlowError("No puede ingresar una rama vacia") @@ -70,21 +72,23 @@ coma. Por ejemplo: "dev, test, prod".""", def _setup_flow_remote(): - panel( - "Configuración de remoto", - "Puede configurar un repositorio remoto para generar PRs automáticamente", + base.io_info( + """Configurar un repositorio remoto le permite generar PRs automáticamente para el +entorno correspondiente. Deberá seleccionar uno de los remotos actuales, o crear +uno nuevo.""", + title="Configuración de remoto", ) remote = None remote_type = None - if confirm("¿Configurar repositorio remoto?"): + if base.io_confirm("¿Configurar repositorio remoto?"): remotes = Git("remote").lines(print="Listando remotos disponibles") if not remotes: - info("No tiene ningún repositorio remoto, se creará uno.") + base.io_info("No tiene ningún repositorio remoto, se creará uno.") - url = prompt( + url = base.io_prompt( "Ingrese la URL del repositorio (e.g. https://github.com/user/repo.git)", persistent=True, ) @@ -93,10 +97,12 @@ def _setup_flow_remote(): elif len(remotes) == 1: remote = remotes[0] else: - info("Tiene más de un remoto, seleccione el que va a utilizar.") - remote = choice("Remoto", remotes) + base.io_info("Tiene más de un remoto, seleccione el que va a utilizar.") + remote = base.io_choice("Remoto", remotes) - remote_type = choice("Tipo de remoto:", ["bitbucket", "github", "gitea"]) + remote_type = base.io_choice( + "Tipo de remoto:", ["bitbucket", "github", "gitea"] + ) return (remote, remote_type) @@ -105,10 +111,10 @@ def _ensure_all_flow_branches_exist(branches: list[str], remote: str | None): existing_branches = Git.get_branches() if not all(map(lambda b: b in existing_branches, branches)): - warning("Algunas de las ramas de entornos expecificadas no existen.") - info("Debe indicar sobre que rama se crearán las ramas de entorno.") + base.io_warning("Algunas de las ramas de entornos expecificadas no existen.") + base.io_info("Debe indicar sobre que rama se crearán las ramas de entorno.") - target_branch = choice("Rama", existing_branches) + target_branch = base.io_choice("Rama", existing_branches) for branch in branches: if branch not in existing_branches: diff --git a/src/git_flow/command/merge.py b/src/git_flow/command/merge.py index 74d3f3c..54e19e6 100644 --- a/src/git_flow/command/merge.py +++ b/src/git_flow/command/merge.py @@ -1,7 +1,8 @@ import typer + +import git_flow.command.base as base from git_flow import GitFlowError from git_flow.changelog import Changelog -import git_flow.command.base as base from git_flow.git import Git app = typer.Typer() @@ -87,8 +88,12 @@ def check_merge_conflicts(target: str): Git("merge", abort=True).exec(print="Abortando merge de prueba", check=False) if merge_conflicts: - base.io_error(f"La rama actual tiene conflictos con {target}.\n" - f"Se recomienda mergear {target} a la rama actual, resolver los conflictos localmente, y ejecutar nuevamente este comando", "Conflictos de merge") + base.io_error( + f"La rama actual tiene conflictos con {target}.\n" + f"Se recomienda mergear {target} a la rama actual, resolver los conflictos localmente,\n" + "y ejecutar nuevamente este comando", + title="Conflictos de merge", + ) raise GitFlowError("Ejecución abortada.") else: base.io_success("No se detectaron merge conflicts.") @@ -109,7 +114,7 @@ def create_pull_request(token: str, branch: str, target: str): target, title, changelog.generate_content(target, branch), - branch not in base.environments + branch not in base.environments, ) base.io_success(message, "Rama creada") @@ -118,12 +123,13 @@ def create_pull_request(token: str, branch: str, target: str): Git("switch", target).exec(print="Cambiando a rama objetivo") Git("pull").exec(print="Obteniendo cambios") + def _get_pr_title_from_branch(branch: str) -> str: if branch in base.environments: target = base.environments[base.environments.index(branch) + 1] return f"Sincronización de entorno {branch} a {target}" - components = branch.split("/") # / or release/// + components = branch.split("/") # / or release/// branch_type = components[-2] branch_desc = components[-1] @@ -136,11 +142,16 @@ def _get_pr_title_from_branch(branch: str) -> str: maybe_ticket = branch_desc[:second_dash] if base.is_valid_ticket(maybe_ticket): branch_ticket = maybe_ticket - branch_desc = branch_desc[second_dash+1:] + branch_desc = branch_desc[second_dash + 1 :] - return " ".join(filter(None, [ - f"[Pasaje a {components[1]}]" if len(components) == 4 else None, - branch_type + ":", - branch_ticket, - branch_desc.replace("-", " ") - ])) + return " ".join( + filter( + None, + [ + f"[Pasaje a {components[1]}]" if len(components) == 4 else None, + branch_type + ":", + branch_ticket, + branch_desc.replace("-", " "), + ], + ) + ) diff --git a/src/git_flow/command/new.py b/src/git_flow/command/new.py index eef826a..0dee72b 100644 --- a/src/git_flow/command/new.py +++ b/src/git_flow/command/new.py @@ -1,9 +1,11 @@ -import typer -from git_flow import BRANCH_TYPES -import git_flow.command.base as base -from git_flow.git import Git from typing import Optional +import typer + +import git_flow.command.base as base +from git_flow import BRANCH_TYPES +from git_flow.git import Git + app = typer.Typer() @@ -17,8 +19,11 @@ def new(): if branch not in base.environments: base.io_warning("La rama actual no corresponde a un entorno configurado.") - base.io_info("Se creará una nueva rama de trabajo sobre la rama actual.\n" - "Debe seleccionar el tipo de cambio a realizar.", title="Tipo de cambio") + base.io_info( + "Se creará una nueva rama de trabajo sobre la rama actual.\n" + "Debe seleccionar el tipo de cambio a realizar.", + title="Tipo de cambio", + ) branch_type = base.io_choice("Tipos de cambio", BRANCH_TYPES) ticket = _get_optional_ticket() @@ -30,15 +35,20 @@ def new(): if base.io_confirm(f"¿Crear rama '{new_branch}' sobre la rama actual?"): Git("switch", new_branch, create=True).exec(print="Creando nueva rama") + def _get_optional_ticket() -> Optional[str]: return base.io_prompt( "Ticket que respalda el cambio (en formato ABC-123)", - validator=lambda x: len(x.strip()) == 0 or base.is_valid_ticket(x) + validator=lambda x: len(x.strip()) == 0 or base.is_valid_ticket(x), ) + def _get_unique_branch_name(branch_type: str, ticket: Optional[str]) -> str: - base.io_info(f"Ingrese las palabras clave que describan el cambio de tipo '{branch_type}'.\n" - "Las mismas se utilizaran para generar el nombre de la rama.", title="Nombre de rama") + base.io_info( + f"Ingrese las palabras clave que describan el cambio de tipo '{branch_type}'.\n" + "Las mismas se utilizaran para generar el nombre de la rama.", + title="Nombre de rama", + ) branch = None branches = Git.get_branches() @@ -46,7 +56,9 @@ def _get_unique_branch_name(branch_type: str, ticket: Optional[str]) -> str: keywords = base.io_prompt("Palabras clave", persistent=True) keywords = filter(lambda k: len(k) > 0, keywords.split(" ")) - branch = branch_type + "/" + (ticket + "-" if ticket else "") + "-".join(keywords) + branch = ( + branch_type + "/" + (ticket + "-" if ticket else "") + "-".join(keywords) + ) if branch in branches: base.io_error( diff --git a/src/git_flow/command/release.py b/src/git_flow/command/release.py index f0402b8..367d18b 100644 --- a/src/git_flow/command/release.py +++ b/src/git_flow/command/release.py @@ -1,38 +1,40 @@ -from git_flow import COMMIT_TYPES, GitFlowError -from git_flow.command.base import * -from git_flow.git import Git from typing import Optional + import typer +import git_flow.command.base as base +from git_flow import COMMIT_TYPES, GitFlowError +from git_flow.git import Git + app = typer.Typer() @app.command() def release(group: Optional[str] = None): """Realiza el merge de la rama, o crea el PR para hacerlo si tiene un remoto configurado""" - ensure_initialized() + base.ensure_initialized() - branch = ensure_right_branch() - env, _ = get_branch_env_and_type(branch) + branch = base.ensure_right_branch() + env, _ = base.get_branch_env_and_type(branch) - if env == environments[-1]: + if env == base.environments[-1]: raise GitFlowError( "No se puede hacer release de una rama en el ultimo entorno." ) - has_remote = "flow.remote" in flowconfig + has_remote = "flow.remote" in base.flowconfig - ensure_clean_worktree(has_remote) + base.ensure_clean_worktree(has_remote) - next_env = environments[environments.index(env) + 1] + next_env = base.environments[base.environments.index(env) + 1] next_branch = ( branch.replace(f"/{env}/", f"/{next_env}/") if branch.startswith("release/") else f"release/{next_env}/{branch}" ) - base = Git.get_first_fork_point(branch, env) - commits = Git("log", base + "..", format="%s").lines() + fork_point = Git.get_first_fork_point(branch, env) + commits = Git("log", fork_point + "..", format="%s").lines() if not commits: raise GitFlowError("No hay cambios a mergear.") @@ -45,9 +47,9 @@ def release(group: Optional[str] = None): grouping = group is not None if not group: - if confirm("¿Desea agrupar este release con otra rama?", False): + if base.io_confirm("¿Desea agrupar este release con otra rama?", False): grouping = True - group = choice( + group = base.io_choice( "Grupo release: ", Git.get_branches("release/" + next_env + "/") ) else: @@ -58,18 +60,13 @@ def release(group: Optional[str] = None): Git("pull").exec(print="Sincronizando cambios la rama objetivo") Git("switch", "-").exec(print="Volviendo a la rama original") - if len(commits) > 1 and confirm( + if len(commits) > 1 and base.io_confirm( f"¿Desea reemplazar los {len(commits)} commits de la rama por uno solo?", False ): - info("Debe ingresar el mensaje del commit a crear.") + base.io_info("Debe ingresar el mensaje del commit a crear.") + message = base.get_commit_message(COMMIT_TYPES[:-1]) - commit_type = choice("Tipo de commit", COMMIT_TYPES[:-1]) - commit_message = prompt( - "Mensaje (máximo recomendado: 100 caracteres)", persistent=True - ) - message = commit_type + ": " + commit_message - - Git("switch", next_branch, base, create=True).exec( + Git("switch", next_branch, fork_point, create=True).exec( print="Creando rama release en base" ) Git("merge", branch, squash=True).exec(print="Squasheando commits en uno solo") @@ -77,7 +74,7 @@ def release(group: Optional[str] = None): else: Git("switch", next_branch, create=True).exec(print="Creando rama release") - status = Git("rebase", base, next_branch, onto=group).code( + status = Git("rebase", fork_point, next_branch, onto=group).code( print="Moviendo cambios hacia el siguiente ambiente" ) diff --git a/src/git_flow/command/tag.py b/src/git_flow/command/tag.py index f7e4b09..0100c42 100644 --- a/src/git_flow/command/tag.py +++ b/src/git_flow/command/tag.py @@ -1,3 +1,8 @@ +from typing import Optional + +import typer + +import git_flow.command.base as base from git_flow import ( COMMIT_TYPE_INCREMENT, SEMVER_MAJOR, @@ -7,10 +12,7 @@ from git_flow import ( GitFlowError, ) from git_flow.changelog import Changelog -from git_flow.command.base import * from git_flow.git import Git -from typing import Optional -import typer app = typer.Typer() @@ -18,22 +20,22 @@ app = typer.Typer() @app.command() def tag(token: Optional[str] = None): """Crea un nuevo tag para el último merge.""" - ensure_initialized() + base.ensure_initialized() target = Git.get_current_branch() - if target not in environments: + if target not in base.environments: raise GitFlowError("Solo se pueden taggear commits en ramas principales.") branch = Git( "show", get_last_merge_commit() + "^2", patch=False, format="%h" ).firstline() - base = Git.get_first_fork_point(branch, target) - commits = Git("log", base + ".." + branch, format="%s").lines() + fork_point = Git.get_first_fork_point(branch, target) + commits = Git("log", fork_point + ".." + branch, format="%s").lines() next_tag = get_next_tag_from_commits(commits, target) if not next_tag: - info( + base.io_info( "Los cambios realizados no implican un salto de versión, se mantiene la anterior." ) return @@ -45,21 +47,21 @@ def tag(token: Optional[str] = None): changelog = Changelog() - changelog.update(next_tag, base, branch) + changelog.update(next_tag, fork_point, branch) Git("add", Changelog.FILENAME).exec(print="Agregando nuevo CHANGELOG.md") Git("commit", message=Changelog.COMMIT_MESSAGE).exec(print="Creando commit") tag_commit = Git("show", "HEAD", patch=False, format="%h").firstline() if ci: Git("push").exec(print="Subiendo commit al remoto para taggearlo") - success(get_remote_api(token).create_tag(next_tag, tag_commit)) + base.io_success(base.get_remote_api(token).create_tag(next_tag, tag_commit)) Git("tag", next_tag, tag_commit).exec(print="Creando tag localmente") def get_token_and_ci(token: Optional[str]): try: - return (ensure_repository_token(), False) + return (base.ensure_repository_token(), False) except GitFlowError: return (token, True)