Compare commits

...
9 Commits
10 changed files with 98 additions and 38 deletions
+11 -1
View File
@@ -22,12 +22,22 @@ jobs:
id: version
run: |
pip install git-flow-envs=="${{ vars.GITFLOWENVS_VERSION }}"
git-flow tag --token="${{ secrets.GITEA_TOKEN }}"
# El código 3 indica que no hace falta un nuevo tag: se saltea la publicación
status=0
git-flow tag --token="${{ secrets.GITEA_TOKEN }}" || status=$?
if [ "$status" -eq 3 ]; then
echo "skip=true" >> "$GITHUB_OUTPUT"
elif [ "$status" -ne 0 ]; then
exit "$status"
fi
- name: Install uv
if: steps.version.outputs.skip != 'true'
uses: astral-sh/setup-uv@v8.3.2
- name: Build and publish
if: steps.version.outputs.skip != 'true'
run: |
uv build
uv publish --token="${{ secrets.PYPI_TOKEN }}"
+1 -1
View File
@@ -53,7 +53,7 @@ más información):
- `init`: Inicializa el repositorio para utilizar git-flow (ramas principales y remoto).
- `new`: Crea una nueva rama siguiendo CB.
- `commit`: Crea un commit siguiendo CC.
- `commit`: Crea un commit siguiendo CC. No permite commitear directamente sobre un entorno.
- `merge`: Mergea la rama actual a uno de los entornos.
- `tag`: Crea un tag sobre el ultimo merge siguiendo SemVer (ver [Tag](#tag)).
- `release`: Prepara la rama actual para pasarse al siguiente entorno configurado.
-14
View File
@@ -1,14 +0,0 @@
image: python:3.10
pipelines:
branches:
'main':
- step:
name: "Tag and publish package"
caches:
- pip
script:
- pip install uv
- uv run git-flow tag --token=$BEARER
- uv build
- uv publish --token=$PYPI_TOKEN
+6 -6
View File
@@ -5,6 +5,7 @@ from os.path import isfile
import questionary
import rich
import rich.panel
from rich import box
from rich.table import Table
from git_flow import (
@@ -51,15 +52,14 @@ def ensure_initialized():
)
def ensure_right_branch():
def ensure_on_branch():
branch = Git.get_current_branch()
if branch == "HEAD":
raise GitFlowError("No se encuentra parado sobre una rama.")
elif io_confirm(f"Rama actual: {branch}. ¿Es correcto?"):
return branch
else:
raise GitFlowError("Ejecución cancelada.")
io_info(f"Rama actual: {branch}")
return branch
def get_branch_env_and_type(branch: str) -> tuple[str, str]:
@@ -201,7 +201,7 @@ def show_commits(title: str, revision_range: str):
if not lines:
raise GitFlowError("No hay cambios a mergear.")
table = Table(title=title, title_justify="left", title_style="bold")
table = Table(title=title, title_justify="left", title_style="bold", box=box.SQUARE)
table.add_column("Fecha", style="dim", no_wrap=True)
table.add_column("Tipo", style="cyan", no_wrap=True)
table.add_column("Mensaje")
+60 -7
View File
@@ -1,10 +1,13 @@
from datetime import datetime
import typer
from rich import box
from rich.table import Table
import git_flow.command.base as base
from git_flow import TRASH_BRANCH_PREFIX, WIP_BRANCH_PREFIX, GitFlowError
from git_flow.command import base
from git_flow.git import Git
from git_flow.io import console
app = typer.Typer()
@@ -13,7 +16,16 @@ app = typer.Typer()
def commit():
"""Crea un commit siguiendo el formato de Conventional Commits"""
base.ensure_initialized()
branch = base.ensure_right_branch()
branch = base.ensure_on_branch()
# Una rama WIP termina squasheando en su rama original, que tampoco puede ser un entorno
target_branch = branch.removeprefix(WIP_BRANCH_PREFIX)
if target_branch in base.environments:
raise GitFlowError(
f"No se puede commitear directamente sobre el entorno '{target_branch}'. "
"Debe crear una rama de trabajo con el comando `new`."
)
if not _has_files_staged():
base.io_warning("No hay cambios en el indice para commitear.")
@@ -22,6 +34,11 @@ def commit():
else:
raise GitFlowError("Debe agregar algún cambio al indice para continuar.")
_show_index()
if not base.io_confirm("¿Desea commitear estos cambios?"):
raise GitFlowError("Ejecución cancelada.")
message = base.get_commit_message()
commit_type = message[: message.index(":")]
@@ -54,10 +71,46 @@ def _has_files_staged():
if not status:
raise GitFlowError("No hay cambios para commitear.")
files_in_index = []
return any(_is_staged(s) for s in status)
for s in status:
if s.worktree != "?" and s.worktree != " ":
files_in_index.append(s.file)
return len(files_in_index) > 0
def _is_staged(s: Git.Status) -> bool:
return s.index != "?" and s.index != " "
INDEX_STATES = {
"M": "modificado",
"T": "cambio de tipo",
"A": "agregado",
"D": "eliminado",
"R": "renombrado",
"C": "copiado",
"U": "en conflicto",
}
def _show_index():
"""Muestra en una tabla los archivos del indice (los que entran en el commit), e informa
cuántos archivos quedan con cambios fuera del indice."""
status = Git.status()
staged = [s for s in status if _is_staged(s)]
not_staged = [s for s in status if s.worktree != " "]
table = Table(
title="Cambios a commitear",
title_justify="left",
title_style="bold",
box=box.SQUARE,
)
table.add_column("Estado", style="cyan", no_wrap=True)
table.add_column("Archivo")
for s in staged:
table.add_row(INDEX_STATES.get(s.index, s.index), s.file)
console.print(table)
if not_staged:
base.io_info(
f"{len(not_staged)} archivo(s) con cambios fuera del indice no se incluirán en el commit."
)
+4 -2
View File
@@ -15,7 +15,7 @@ def merge():
"""Realiza el merge de la rama, o crea el PR para hacerlo si tiene un remoto configurado"""
base.ensure_initialized()
branch = base.ensure_right_branch()
branch = base.ensure_on_branch()
if branch in base.environments:
if branch == base.environments[-1]:
@@ -79,7 +79,9 @@ def check_merge_conflicts(target: str):
print="Realizando merge de prueba para verificar conflictos"
)
Git("merge", abort=True).exec(print="Abortando merge de prueba", check=False)
# Si el merge no llegó a realizarse (por ejemplo, "Already up to date") no hay nada que abortar
if Git.is_merging():
Git("merge", abort=True).exec(print="Abortando merge de prueba")
if merge_conflicts:
base.io_error(
+1 -1
View File
@@ -14,7 +14,7 @@ def new():
"""Crea una nueva rama siguiendo Conventional Branches"""
base.ensure_initialized()
branch = base.ensure_right_branch()
branch = base.ensure_on_branch()
if branch not in base.environments:
base.io_warning("La rama actual no corresponde a un entorno configurado.")
+1 -1
View File
@@ -12,7 +12,7 @@ def release():
"""Realiza el merge de la rama, o crea el PR para hacerlo si tiene un remoto configurado"""
base.ensure_initialized()
branch = base.ensure_right_branch()
branch = base.ensure_on_branch()
env, _ = base.get_branch_env_and_type(branch)
if env == base.environments[-1]:
+5 -2
View File
@@ -14,6 +14,9 @@ from git_flow.git import Git
app = typer.Typer()
# Código de salida cuando no hace falta crear un tag, para que el CI pueda saltear la publicación
EXIT_NO_TAG_NEEDED = 3
@app.command()
def tag(target: str | None = None, token: str | None = None, update_changelog: bool = False):
@@ -35,7 +38,7 @@ def tag(target: str | None = None, token: str | None = None, update_changelog: b
merge_commit_tag = Git("tag", "--points-at", merge_commit).firstline()
if merge_commit_tag:
base.io_info(f"El último merge ya tiene un tag: {merge_commit_tag}")
return
raise typer.Exit(code=EXIT_NO_TAG_NEEDED)
token, ci = get_token_and_ci(token)
@@ -97,7 +100,7 @@ def tag(target: str | None = None, token: str | None = None, update_changelog: b
"Los cambios realizados no implican un salto de versión, se mantiene la anterior."
)
raise typer.Exit(code=1)
raise typer.Exit(code=EXIT_NO_TAG_NEEDED)
if update_changelog:
changelog = Changelog()
+9 -3
View File
@@ -26,8 +26,9 @@ class Git:
file: str
def __init__(self, line: str) -> None:
self.worktree = line[0]
self.index = line[1]
# Formato porcelain "XY <file>": X es el estado en el indice, Y en el worktree
self.index = line[0]
self.worktree = line[1]
self.file = line[3:].strip("\n")
command: list[str]
@@ -62,6 +63,11 @@ class Git:
def get_branches(prefix: str = ""):
return Git._get_references("heads/" + prefix)
@staticmethod
def is_merging() -> bool:
"""Indica si hay un merge en curso (existe MERGE_HEAD), con o sin conflictos."""
return Git("rev-parse", "MERGE_HEAD", quiet=True, verify=True).code() == 0
@staticmethod
def status():
return list(map(Git.Status, Git("status", porcelain=True).lines(False)))
@@ -170,7 +176,7 @@ class Git:
process = self._get(**kwargs)
lines = process.stdout.splitlines()
return lines if not strip else list(map(lambda l: l.strip(), lines))
return lines if not strip else [l.strip() for l in lines]
def firstline(self, **kwargs):
lines = self.lines(**kwargs)