jt/git-flow
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: migrar comando merge para que utilice questionary

Browse Source
This commit is contained in:
jt
2026-07-14 00:28:20 -03:00
parent 320d40a211
commit 0782cfb041
2 changed files with 34 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/git_flow/command/base.py
+3
View File
@@ -148,6 +148,9 @@ def io_warning(message: str, title: Optional[str] = None):
def io_info(message: str, title: Optional[str] = None): 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)
def _io_status(message: str, prefix: str, style: str, title: Optional[str] = None): def _io_status(message: str, prefix: str, style: str, title: Optional[str] = None):
if title: 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"))
src/git_flow/command/merge.py
+31 -33
View File
@@ -1,7 +1,7 @@
import typer import typer
from git_flow import GitFlowError from git_flow import GitFlowError
from git_flow.changelog import Changelog from git_flow.changelog import Changelog
from git_flow.command.base import * import git_flow.command.base as base
from git_flow.git import Git from git_flow.git import Git
app = typer.Typer() app = typer.Typer()
@@ -10,36 +10,36 @@ app = typer.Typer()
@app.command() @app.command()
def merge(): def merge():
"""Realiza el merge de la rama, o crea el PR para hacerlo si tiene un remoto configurado""" """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() branch = base.ensure_right_branch()
if branch in environments: if branch in base.environments:
if branch == environments[-1]: if branch == base.environments[-1]:
raise GitFlowError("No se puede hacer un merge del último entorno.") raise GitFlowError("No se puede hacer un merge del último entorno.")
panel( base.io_info(
"Merge de entornos",
"""Está por hacer un merge de 2 entornos que puede implicar muchos cambios en el proyecto.""", """Está por hacer un merge de 2 entornos que puede implicar muchos cambios en el proyecto.""",
"Merge de entornos",
) )
if not confirm("¿Desea continuar?"): if not base.io_confirm("¿Desea continuar?"):
raise GitFlowError("Ejecución abortada") raise GitFlowError("Ejecución abortada")
target = environments[environments.index(branch) + 1] target = base.environments[base.environments.index(branch) + 1]
else: else:
target, _ = get_branch_env_and_type(branch) target, _ = base.get_branch_env_and_type(branch)
if "flow.remote" in flowconfig: if "flow.remote" in base.flowconfig:
remote = flowconfig["flow.remote"] remote = base.flowconfig["flow.remote"]
token = ensure_repository_token() token = base.ensure_repository_token()
run_remote(remote, token, branch, target) run_remote(remote, token, branch, target)
else: else:
run_local(branch, target) run_local(branch, target)
def run_remote(remote: str, token: str, branch: str, target: str): def run_remote(remote: str, token: str, branch: str, target: str):
ensure_clean_worktree(True) base.ensure_clean_worktree(True)
Git("switch", target).exec(print="Cambiando a rama destino") Git("switch", target).exec(print="Cambiando a rama destino")
Git("pull").exec(print="Obteniendo ultimos cambios del remoto") Git("pull").exec(print="Obteniendo ultimos cambios del remoto")
@@ -48,8 +48,8 @@ def run_remote(remote: str, token: str, branch: str, target: str):
check_merge_conflicts(target) check_merge_conflicts(target)
show_commits_to_merge(target) show_commits_to_merge(target)
if confirm(f"¿Crear PR de '{branch}' a '{target}'?"): if base.io_confirm(f"¿Crear PR de '{branch}' a '{target}'?"):
if branch not in environments: if branch not in base.environments:
Git("push", remote, branch, set_upstream=True).exec( Git("push", remote, branch, set_upstream=True).exec(
print="Subiendo rama al remoto para crear PR" print="Subiendo rama al remoto para crear PR"
) )
@@ -58,11 +58,11 @@ def run_remote(remote: str, token: str, branch: str, target: str):
def run_local(branch: str, target: str): def run_local(branch: str, target: str):
ensure_clean_worktree(False) base.ensure_clean_worktree(False)
check_merge_conflicts(target) check_merge_conflicts(target)
show_commits_to_merge(target) show_commits_to_merge(target)
if confirm(f"¿Mergear rama '{branch}' a '{target}'?"): if base.io_confirm(f"¿Mergear rama '{branch}' a '{target}'?"):
Git("switch", target).exec(print="Cambiando a rama destino") Git("switch", target).exec(print="Cambiando a rama destino")
Git("merge", branch, ff=False).exec(print="Mergeando") Git("merge", branch, ff=False).exec(print="Mergeando")
@@ -87,13 +87,11 @@ def check_merge_conflicts(target: str):
Git("merge", abort=True).exec(print="Abortando merge de prueba", check=False) Git("merge", abort=True).exec(print="Abortando merge de prueba", check=False)
if merge_conflicts: if merge_conflicts:
error(f"La rama actual tiene conflictos con {target}") base.io_error(f"La rama actual tiene conflictos con {target}.\n"
info( f"Se recomienda mergear {target} a la rama actual, resolver los conflictos localmente, y ejecutar nuevamente este comando", "Conflictos de merge")
f"""Se recomienda mergear {target} a la rama actual, resolver los conflictos localmente, y ejecutar nuevamente este comando"""
)
raise GitFlowError("Ejecución abortada.") raise GitFlowError("Ejecución abortada.")
else: else:
success("No se detectaron merge conflicts.") base.io_success("No se detectaron merge conflicts.")
def create_pull_request(token: str, branch: str, target: str): def create_pull_request(token: str, branch: str, target: str):
@@ -101,28 +99,28 @@ def create_pull_request(token: str, branch: str, target: str):
commits = Git("log", target + "..", format="%s").lines() commits = Git("log", target + "..", format="%s").lines()
title = commits[0] if len(commits) == 1 else _get_pr_title_from_branch(branch) title = commits[0] if len(commits) == 1 else _get_pr_title_from_branch(branch)
panel("Título del PR", title) base.panel("Título del PR", title)
if confirm("¿Desea cambiar el título del PR?", False): if base.io_confirm("¿Desea cambiar el título del PR?", False):
title = prompt("Título del PR") title = base.io_prompt("Título del PR", persistent=True)
message = get_remote_api(token).create_pull_request( message = base.get_remote_api(token).create_pull_request(
branch, branch,
target, target,
title, title,
changelog.generate_content(target, branch), changelog.generate_content(target, branch),
branch not in environments branch not in base.environments
) )
success(message) base.io_success(message, "Rama creada")
if confirm("¿Desea cambiar a la rama objetivo y bajar los cambios?"): if base.io_confirm("¿Desea cambiar a la rama objetivo y bajar los cambios?"):
Git("switch", target).exec(print="Cambiando a rama objetivo") Git("switch", target).exec(print="Cambiando a rama objetivo")
Git("pull").exec(print="Obteniendo cambios") Git("pull").exec(print="Obteniendo cambios")
def _get_pr_title_from_branch(branch: str) -> str: def _get_pr_title_from_branch(branch: str) -> str:
if branch in environments: if branch in base.environments:
target = environments[environments.index(branch) + 1] target = base.environments[base.environments.index(branch) + 1]
return f"Sincronización de entorno {branch} a {target}" return f"Sincronización de entorno {branch} a {target}"
components = branch.split("/") # <type>/<desc> or release/<env>/<type>/<desc> components = branch.split("/") # <type>/<desc> or release/<env>/<type>/<desc>
@@ -136,7 +134,7 @@ def _get_pr_title_from_branch(branch: str) -> str:
if first_dash >= 0 and second_dash >= 0: if first_dash >= 0 and second_dash >= 0:
maybe_ticket = branch_desc[:second_dash] maybe_ticket = branch_desc[:second_dash]
if is_valid_ticket(maybe_ticket): if base.is_valid_ticket(maybe_ticket):
branch_ticket = maybe_ticket branch_ticket = maybe_ticket
branch_desc = branch_desc[second_dash+1:] branch_desc = branch_desc[second_dash+1:]