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

Merged in feature/add-release-command (pull request #33)

Browse Source 
feature: add release command

Approved-by: Jonathan Teran
This commit is contained in:
JonathanGitFlow
2025-11-23 06:53:25 +00:00
committed by jt
parent 27147232fe 43d7e5a057
commit d9015cc5ab
4 changed files with 105 additions and 469 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/git_flow/command/base.py
+23 -2
View File
@@ -1,6 +1,5 @@
from argparse import Namespace, ArgumentParser
from abc import ABC, abstractmethod
import os
from os.path import isfile
from git_flow import (
@@ -45,6 +44,11 @@ class Command(ABC):
def run(self, args: Namespace = Namespace()):
pass
def init(self):
self.flowconfig = (
Git.get_config(FLOWCONFIG_FILE) if isfile(FLOWCONFIG_FILE) else {}
)
def success(self, msg: str):
self._print(TYPE_SUCCESS, msg)
@@ -145,7 +149,7 @@ class Command(ABC):
raise GitFlowError("Ejecución cancelada.")
def ensure_repository_token(self):
if not os.path.isfile(REPOSITORY_TOKEN_FILENAME):
if not isfile(REPOSITORY_TOKEN_FILENAME):
raise GitFlowError(
"No existe un token para el repositorio en " + REPOSITORY_TOKEN_FILENAME
)
@@ -153,6 +157,23 @@ class Command(ABC):
with open(REPOSITORY_TOKEN_FILENAME) as f:
return f.readline().strip()
def ensure_clean_worktree(self, has_remote: bool):
status = Git.status()
if not status:
return
not_empty = any(map(lambda s: s.index != " " or s.worktree != " ", status))
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 get_branch_env_and_type(self, branch: str) -> tuple[str, str]:
components = branch.split("/")
target_branches = self.flowconfig["flow.branches"].split(" ")
src/git_flow/command/merge.py
+2 -19
View File
@@ -27,7 +27,7 @@ class MergeCommand(Command):
self.run_local(base, branch, target)
def run_remote(self, remote: str, token: str, base: str, branch: str, target: str):
self.check_pending_changes(True)
self.ensure_clean_worktree(True)
Git("switch", target).exec(print="Cambiando a rama destino")
Git("pull").exec(print="Obteniendo ultimos cambios del remoto")
@@ -44,7 +44,7 @@ class MergeCommand(Command):
self.create_pull_request(token, base, branch, target)
def run_local(self, base: str, branch: str, target: str):
self.check_pending_changes(False)
self.ensure_clean_worktree(False)
self.check_merge_conflicts(target)
self.show_commits_to_merge(base)
@@ -52,23 +52,6 @@ class MergeCommand(Command):
Git("switch", target).exec(print="Cambiando a rama destino")
Git("merge", branch, ff=False).exec(print="Mergeando")
def check_pending_changes(self, has_remote: bool):
status = Git.status()
if not status:
return
not_empty = any(map(lambda s: s.index != " " or s.worktree != " ", status))
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 show_commits_to_merge(self, base):
commits = Git("log", base + "..", format="%s").lines()
src/git_flow/command/release.py
+67
View File
@@ -0,0 +1,67 @@
from argparse import Namespace
from git_flow import COMMIT_TYPES, GitFlowError
from git_flow.command.base import Command
from git_flow.git import Git
class ReleaseCommand(Command):
def name(self) -> str:
return "release"
def description(self) -> str:
return """Realiza el merge de la rama, o crea el PR para hacerlo si tiene un remoto configurado"""
def run(self, args: Namespace = Namespace()):
self.ensure_initialized()
branch = self.ensure_right_branch()
envs = self.flowconfig["flow.branches"].split(",")
(env, _) = self.get_branch_env_and_type(branch)
if env == envs[-1]:
raise GitFlowError(
"No se puede hacer release de una rama en el ultimo entorno."
)
has_remote = "flow.remote" in self.flowconfig
self.ensure_clean_worktree(has_remote)
next_env = envs[envs.index(env) + 1]
next_branch = (
branch.replace(f"/{env}/", f"/{next_env}/")
if branch.startswith("release/")
else f"release/{next_env}/{branch}"
)
if has_remote:
Git("pull").exec(print="Sincronizando cambios de la rama")
Git("switch", next_env).exec(print="Cambiando al siguiente entorno")
Git("pull").exec(print="Sincronizando cambios del entorno")
Git("switch", "-").exec(print="Volviendo a la rama original")
base = Git.get_first_fork_point(branch, env)
commits = int(Git("rev-list", base + "..", count=True).firstline())
if commits > 1 and self.confirm(
"Desea reemplazar los {commits} commits de la rama por uno solo?"
):
Git("switch", next_branch, base, create=True).exec(
print="Creando rama release en base"
)
Git("merge", branch, squash=True).exec(
print="Squasheando commits en uno solo"
)
commit_type = self.choice("Tipo de commit", COMMIT_TYPES[:-1])
commit_message = self.prompt(
"Mensaje (máximo recomendado: 100 caracteres)", persistent=True
)
message = commit_type + ": " + commit_message
Git("commit", m=message).exec(print="Creando commit único")
else:
Git("switch", next_branch, create=True).exec(print="Creando rama release")
Git("rebase", base, next_branch, onto=next_env).exec(
print="Moviendo cambios hacia el siguiente ambiente"
)
src/git_flow/main.py
+13 -448
View File
@@ -1,27 +1,22 @@
#!/usr/bin/env python3
import os
from os.path import isfile
import sys
import datetime
import requests
import locale
from argparse import Namespace, ArgumentParser
from git_flow import GitFlowError
from git_flow.changelog import Changelog
from git_flow.command.base import Command
from git_flow.command.commit import CommitCommand
from git_flow.command.init import InitCommand
from git_flow.command.merge import MergeCommand
from git_flow.command.new import NewCommand
from git_flow.command.release import ReleaseCommand
from git_flow.command.tag import TagCommand
from git_flow.git import BUMP_VERSION, FLOWCONFIG_FILE, Git
from git_flow.io import *
REPOSITORY_TOKEN_PATH=".repository-token"
LOCALE = 'es_AR.UTF-8'
REPOSITORY_TOKEN_PATH = ".repository-token"
LOCALE = "es_AR.UTF-8"
class GitFlowCommand(Command):
commands: tuple[Command, ...]
@@ -33,14 +28,17 @@ class GitFlowCommand(Command):
return "git-flow"
def description(self) -> str:
return """Git flow es un workflow automatizado siguiendo conventional commits, semver, y
deploys aislados a distintos entornos."""
return """Git flow es una herramienta para automatizar un workflow siguiendo conventional commits, conventional branch y semver."""
def setup_parser(self, parser: ArgumentParser) -> ArgumentParser:
subparsers = parser.add_subparsers(title="comandos", dest="command", required=True)
subparsers = parser.add_subparsers(
title="comandos", dest="command", required=True
)
for command in self.commands:
subparser = subparsers.add_parser(command.name(), description=command.description())
subparser = subparsers.add_parser(
command.name(), description=command.description()
)
command.setup_parser(subparser)
return parser
@@ -55,7 +53,7 @@ class GitFlowCommand(Command):
for command in self.commands:
if command.name() == args.command:
command.flowconfig = Git.get_config(FLOWCONFIG_FILE) if isfile(FLOWCONFIG_FILE) else {}
command.init()
return command.run(args)
@@ -71,6 +69,7 @@ def main():
CommitCommand(),
MergeCommand(),
TagCommand(),
ReleaseCommand(),
)
try:
command.run()
@@ -81,440 +80,6 @@ def main():
return
args = sys.argv
if len(args) < 2:
print_error("missing command")
return
command = args[1]
try:
if command == "tag":
tag_command()
elif command == "release":
release_command()
elif command == "help":
help_command()
else:
help_command(command)
except Exception as e:
print_error(f"Ocurrio un error al ejecutar el comando: {e}")
BRANCH_TYPES = [
"feature",
"refactor",
"bugfix",
"hotfix",
"perf",
"docs",
"typo",
"testing",
"breaking",
]
BRANCH_TYPES_INCREMENT = {
"feature": "minor",
"refactor": "minor",
"bugfix": "patch",
"hotfix": "patch",
"perf": "minor",
"docs": "patch",
"typo": "patch",
"testing": "patch",
"breaking": "major",
}
COMMIT_TYPES = [
"feature",
"bugfix",
"hotfix",
"refactor",
"perf",
"docs",
"typo",
"testing",
"ignore",
"wip"
]
DEFAULT_REMOTE = "origin"
def parse_url(url: str) -> list[str]:
if "@" in url:
# git@github.com:username/repository.git
start = url.find("@") + 1
end = url.find(":")
service = url[start : end]
start = end + 1
end = url.find(".git") if ".git" in url else len(url)
repository = url[start : end]
elif "://" in url:
# https://github.com/username/repository.git
start = url.index("://") + 3
end = url.index("/", start)
service = url[start : end]
start = end + 1
end = url.find(".git") if ".git" in url else len(url)
repository = url[start : end]
else:
raise RuntimeError("Invalid url: " + url)
# Esto permite utilizar un Host ficticio si tenemos 2 cuentas de bitbucket (personal y trabajo)
if service.endswith(".bitbucket.org"):
service = "bitbucket.org"
return [service, repository]
def get_branch_env_and_type(branch: str) -> tuple[str, str]:
components = branch.split("/")
target_branches = Git.flow_config("flow.branches").firstline().split(" ")
if len(components) == 2:
if components[0] not in BRANCH_TYPES:
raise ValueError(f"Branch inválida, el tipo '{components[0]}' no es válido.")
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 ValueError(
"Branch release inválido, debe tener el siguiente formato: "
"release/<env>/<type>/<name>, pero es: " + branch
)
return (components[1], components[2])
else:
raise ValueError("Branch inválido: " + branch)
def tag_command():
ensure_initialized()
ci = len(sys.argv) == 3 # ./git-flow tag <BEARER>
token = sys.argv[2] if ci else None
if token is None:
with open(REPOSITORY_TOKEN_PATH, "r") as file:
token = file.readline().strip()
output = Git("log", first_parent=True, merges=True, max_count=1, format="%H,%s").lines()
if not output:
print_error("No hay merges a taggear")
return
output = output[0]
pos = output.index(',')
commit = output[:pos]
message = output[pos+1:]
merged_branch = None
output = Git("describe", commit, tags=True, exact_match=True).firstline(check=False)
if output:
print_error("El ultimo merge ya tiene tag: " + output)
return
if message.startswith("Merged in "):
merged_branch = message.removeprefix("Merged in ")
merged_branch = merged_branch[:merged_branch.index(' ')]
elif message.startswith("Merge branch '"):
merged_branch = message.removeprefix("Merge branch '")
merged_branch = merged_branch[:merged_branch.index("'")]
else:
print_error("No se pudo obtener el nombre de la rama mergeada")
return
remote = Git.flow_config("flow.remote").firstline()
branch = Git.get_current_branch()
last_version = Git.get_current_tag() or "v1.0.0"
from_major = 1
until_dash = last_version.index("-") if "-" in last_version else None
[major, minor, patch] = map(int, last_version[from_major:until_dash].split("."))
(_, merged_branch_type) = get_branch_env_and_type(merged_branch)
increment = BRANCH_TYPES_INCREMENT[merged_branch_type]
if increment == "major":
major = major + 1
minor = 0
patch = 0
elif increment == "minor":
minor = minor + 1
patch = 0
elif increment == "patch":
patch = patch + 1
last_branch = Git.flow_config("flow.branches").firstline().split(" ")[-1]
suffix = "-" + branch if branch != last_branch else ""
new_version = f"v{major}.{minor}.{patch}" + suffix
if ci:
url = Git("remote", "get-url", remote).firstline()
if url:
[service, repository] = parse_url(url)
create_tag(service, repository, token, new_version, commit)
merged_commit = Git("show", "HEAD^2", patch=False, format="%H").firstline()
if update_changelog(merged_commit, branch):
Git("add", Changelog.FILENAME).exec()
Git("commit", message=BUMP_VERSION).exec()
Git("push").exec()
else:
Git("tag", new_version, commit).exec()
def release_command():
ensure_initialized()
branch = ensure_right_branch()
env_branches = Git.flow_config("flow.branches").firstline().split(" ")
(env, _) = get_branch_env_and_type(branch)
if env == env_branches[-1]:
raise RuntimeError("No se puede hacer release de una rama en el ultimo entorno.")
base = Git.get_first_fork_point(branch, env)
next_env = env_branches[env_branches.index(env) + 1]
next_branch = branch.replace(f"/{env}/", f"/{next_env}/") if branch.startswith("release/") else f"release/{next_env}/{branch}"
Git("switch", next_branch, create=True).exec()
commits = int(Git("rev-list", base + ".." + next_branch, count=True).firstline())
if commits > 1 and confirm(f"Desea reemplazar los {commits} commits de la rama por uno solo?"):
Git("reset", base, soft=True).exec()
commit_type = choose("Tipo de commit", COMMIT_TYPES[:-1]) # Solo permitir commits NO wip
Git("commit", m=commit_type + ": " + get_commit_message()).exec()
Git("rebase", base, next_branch, onto=next_env).exec()
def help_command(name: str|None = None):
if name is not None:
print_error(f"Comando inválido: {name}")
script = sys.argv[0]
print(f"""USO:
{script} <COMANDO>
COMANDOS:
init inicializa el repositorio por única vez para utilizar git-flow
new crea una nueva rama para realizar cambios
commit crea un commit siguiendo el formato de Conventional Commit
merge actualiza el changelog y crea un nuevo PR para mergear la rama
tag crea un nuevo tag para el ultimo merge
release crea una nueva rama release para pasar la rama actual al siguiente entorno
help muestra este texto de ayuda""", file=sys.stderr)
def ensure_initialized():
initialized = Git.flow_config("flow.initialized").firstline()
if initialized == "":
raise RuntimeError("El repositorio no fue inicializado, antes de continuar ejecute `git flow init`.")
elif initialized != "true":
raise RuntimeError(f"El valor de `flow.initialized` ({initialized}) es inválido.")
def get_commit_message():
message = None
while message is None:
message = input("Mensaje (recomendado: 100 caracteres): ").strip()
if not message:
print_error("El mensaje no puede ser vacío.")
message = None
return message
def ensure_right_branch():
branch = Git.get_current_branch()
if branch != "HEAD":
if confirm(f"Se encuentra sobre la rama '{branch}'. ¿Es correcto?"):
return branch
raise RuntimeError("Abortando operación")
def get_base_branch(branch: str):
if branch.startswith("release/"):
start = len("release/")
end = branch.index("/", start)
return branch[start:end]
else:
return Git.flow_config("flow.branches").firstline().split(" ")[0]
def get_unique_branch_name(branch_type):
print_info(f"Ingrese las palabras clave que describan el cambio de tipo '{branch_type}'.")
branch = None
branches = Git.get_branches()
while branch is None:
keywords = input("Palabras clave (por ej. 'edit worker form'): ")
if len(keywords) == 0:
print_error("Debe ingresar al menos una palabra clave.")
else:
keywords = list(filter(lambda k: len(k) > 0, keywords.split(" ")))
branch = branch_type + "/" + "-".join(keywords)
if branch in branches:
print_error(
f"La rama '{branch}' ya existe, cambie las palabras claves para generar otra"
)
branch = None
return branch
def update_changelog(branch: str, target: str):
last_commit = Git("show", "HEAD", no_patch=True, format="%s").firstline()
if last_commit == BUMP_VERSION:
return False
else:
content = generate_changelog_entry(branch, target)
if os.path.isfile(Changelog.FILENAME):
with open(Changelog.FILENAME, "r") as changelog:
content = content + changelog.read()
with open(Changelog.FILENAME, "w") as changelog:
changelog.write(content)
return True
def get_changelog_header_lines(branch: str):
name = Git("config", "user.name").firstline()
email = Git("config", "user.email").firstline()
now = datetime.datetime.now()
weekday = now.strftime("%A").capitalize()
month = now.strftime("%B").capitalize()
timestamp = now.strftime(f"{weekday}, %e de {month} de %Y, %H:%M")
return [
f"## {timestamp}",
"",
f"- Autor: [{name}](mailto:{email})",
f"- Rama: `{branch}`",
"",
"### Commits",
""
]
def get_changelog_content_lines(branch: str, target: str):
lines = []
base = Git.get_first_fork_point(branch, target)
commits = Git("log", base + ".." + branch, first_parent=True, format="%h", merges=False).lines()
for commit in commits:
message = Git("show", commit, patch=False, format="%s").firstline()
email = Git("show", commit, patch=False, format="%ae").firstline()
username = email[:email.index('@')]
if message == BUMP_VERSION:
continue
index = message.index(":")
commit_type = message[:index]
commit_message = message[index+1:]
lines.append(f"- **{commit_type}**: {commit_message} [[{username}](mailto:{email})] ({commit})")
return lines
def get_changelog_footer_lines():
return [
"",
"---",
"",
""
]
def generate_changelog_entry(branch: str, target: str):
entry_lines = get_changelog_header_lines(branch)
entry_lines += get_changelog_content_lines(branch, target)
entry_lines += get_changelog_footer_lines()
return "\n".join(entry_lines)
def create_pull_request(service: str, repository: str, source: str, destination: str):
endpoint = get_api_endpoint(service, repository, "/pullrequests")
title = source.replace("/", ": ").replace("-", " ")
description = "\n".join(get_changelog_content_lines(source, destination))
headers = get_headers()
json = {
"title": title,
"description": description,
"source": {"branch": {"name": source}},
"destination": {"branch": {"name": destination}},
"close_source_branch": True
}
request = requests.post(endpoint, headers=headers, json=json)
if request.ok:
json = request.json()
print_success("PR creado exitosamente: " + json['links']['html']['href'])
else:
print_error("Ocurrió un error al crear el PR: " + request.text)
def create_tag(service: str, repository: str, token: str, tag: str, commit: str):
endpoint = get_api_endpoint(service, repository, "/refs/tags")
headers = get_headers(token)
json = {
"name": tag,
"target": {
"hash": commit
}
}
response = requests.post(endpoint, headers=headers, json=json)
if response.ok:
print("Tag creado exitosamente.")
else:
print_error("Ocurrió un error al crear el tag: " + response.text)
def get_headers(token: str | None = None):
if not token:
with open(REPOSITORY_TOKEN_PATH) as file:
token = file.readline().strip()
return {
"Accept": "application/json",
"Content-Type": "application/json",
"Authorization": "Bearer " + token,
}
def get_api_endpoint(service: str, repository: str, path: str):
if service == "bitbucket.org":
return "https://api.bitbucket.org/2.0/repositories/" + repository + path
return ""
if __name__ == "__main__":
main()