feature: tag from ci fetches commits from remote api

This commit is contained in:
Jonathan Teran Carballo
2026-09-20 21:05:21 -03:00
parent 889109bf77
commit 26ab81b175
8 changed files with 195 additions and 31 deletions
+30 -1
View File
@@ -1,9 +1,10 @@
from abc import ABC, abstractmethod
import giturlparse
import requests
from git_flow import SUPPORTED_REMOTE_URL_SCHEMAS, GitFlowError
from git_flow.git import Git
from git_flow.git import Commit, Git
class RemoteAPI(ABC):
@@ -65,6 +66,12 @@ class RemoteAPI(ABC):
def create_tag(self, tag: str, commit: str) -> str:
pass
@abstractmethod
def get_merge_commits(self, merge_commit: str) -> list[Commit]:
"""Busca el PR que generó el merge commit y devuelve sus commits (del más nuevo al más
viejo), sin depender del historial local."""
pass
@abstractmethod
def get_endpoint(self, resource: str) -> str:
pass
@@ -75,3 +82,25 @@ class RemoteAPI(ABC):
"Content-Type": "application/json",
"Authorization": "Bearer " + self.token,
}
def get_json(self, url: str, params: dict | None = None):
response = requests.get(url, headers=self.get_headers(), params=params)
if not response.ok:
raise GitFlowError(f"Ocurrió un error al consultar {url}: {response.text}")
return response
def get_all_pages(self, url: str, page_size: int = 50) -> list:
"""Recorre un endpoint paginado por número de página (page / per_page)."""
items: list = []
page = 1
while True:
batch = self.get_json(url, {"page": page, "per_page": page_size}).json()
items += batch
if len(batch) < page_size:
return items
page += 1
+33
View File
@@ -1,5 +1,8 @@
import re
import requests
from git_flow import GitFlowError
from git_flow.git import Commit
from git_flow.remote.base import RemoteAPI
@@ -44,5 +47,35 @@ class BitbucketRemoteAPI(RemoteAPI):
else:
raise GitFlowError("Ocurrió un error al crear el tag: " + response.text)
def get_merge_commits(self, merge_commit: str) -> list[Commit]:
prs = self.get_json(self.get_endpoint(f"commit/{merge_commit}/pullrequests")).json()
pr = next((p for p in prs["values"] if p["state"] == "MERGED"), None)
if not pr:
raise GitFlowError(f"No se encontró un PR para el commit {merge_commit}")
commits: list[Commit] = []
url: str | None = self.get_endpoint(f"pullrequests/{pr['id']}/commits")
# Bitbucket pagina siguiendo el link "next", y devuelve del más nuevo al más viejo
while url:
page = self.get_json(url).json()
for c in page["values"]:
raw = c["author"].get("raw", "")
match = re.search(r"<(.+?)>", raw)
commits.append(
Commit(
c["hash"][:7],
c["message"].splitlines()[0],
match.group(1) if match else raw,
len(c["parents"]) > 1,
)
)
url = page.get("next")
return commits
def get_endpoint(self, resource: str) -> str:
return f"{BitbucketRemoteAPI.API_ENDPOINT}/{self.repository}/{resource}"
+30
View File
@@ -1,5 +1,6 @@
import requests
from git_flow import GitFlowError
from git_flow.git import Commit
from git_flow.remote.base import RemoteAPI
@@ -44,6 +45,35 @@ class GiteaRemoteAPI(RemoteAPI):
else:
raise GitFlowError("Ocurrió un error al crear el tag: " + response.text)
def get_merge_commits(self, merge_commit: str) -> list[Commit]:
pr = self.get_json(self.get_endpoint(f"/commits/{merge_commit}/pull")).json()
url = self.get_endpoint(f"/pulls/{pr['number']}/commits")
items: list = []
page = 1
# Gitea pagina con page / limit
while True:
batch = self.get_json(
url, {"page": page, "limit": 50, "files": "false", "verification": "false"}
).json()
items += batch
if len(batch) < 50:
break
page += 1
# La API los devuelve del más viejo al más nuevo
return [
Commit(
c["sha"][:7],
c["commit"]["message"].splitlines()[0],
c["commit"]["author"]["email"],
len(c["parents"]) > 1,
)
for c in reversed(items)
]
def get_endpoint(self, resource: str) -> str:
return f"{self.base_url}/api/v1/repos/{self.repository}{resource}"
+21
View File
@@ -1,5 +1,6 @@
import requests
from git_flow import GitFlowError
from git_flow.git import Commit
from git_flow.remote.base import RemoteAPI
@@ -42,5 +43,25 @@ class GithubRemoteAPI(RemoteAPI):
else:
raise GitFlowError("Ocurrió un error al crear el tag: " + response.text)
def get_merge_commits(self, merge_commit: str) -> list[Commit]:
prs = self.get_json(self.get_endpoint(f"/commits/{merge_commit}/pulls")).json()
pr = next((p for p in prs if p.get("merge_commit_sha") == merge_commit), None)
if not pr:
raise GitFlowError(f"No se encontró un PR para el commit {merge_commit}")
items = self.get_all_pages(self.get_endpoint(f"/pulls/{pr['number']}/commits"), 100)
# La API los devuelve del más viejo al más nuevo
return [
Commit(
c["sha"][:7],
c["commit"]["message"].splitlines()[0],
c["commit"]["author"]["email"],
len(c["parents"]) > 1,
)
for c in reversed(items)
]
def get_endpoint(self, resource: str) -> str:
return f"{GithubRemoteAPI.API_ENDPOINT}/{self.repository}{resource}"
+23
View File
@@ -2,6 +2,7 @@ from urllib.parse import quote
import requests
from git_flow import GitFlowError
from git_flow.git import Commit
from git_flow.remote.base import RemoteAPI
@@ -47,6 +48,28 @@ class GitlabRemoteAPI(RemoteAPI):
else:
raise GitFlowError("Ocurrió un error al crear el tag para repositorio " + quote(self.repository, safe="") + ": " + response.text)
def get_merge_commits(self, merge_commit: str) -> list[Commit]:
mrs = self.get_json(
self.get_endpoint(f"/repository/commits/{merge_commit}/merge_requests")
).json()
mr = next((m for m in mrs if m.get("merge_commit_sha") == merge_commit), None)
if not mr:
raise GitFlowError(f"No se encontró un MR para el commit {merge_commit}")
items = self.get_all_pages(self.get_endpoint(f"/merge_requests/{mr['iid']}/commits"), 100)
# GitLab ya los devuelve del más nuevo al más viejo
return [
Commit(
c["short_id"][:7],
c["title"],
c["author_email"],
len(c["parent_ids"]) > 1,
)
for c in items
]
def get_endpoint(self, resource: str) -> str:
project_id = quote(self.repository, safe="")
return f"{self.base_url}/api/v4/projects/{project_id}{resource}"