135 lines
4.3 KiB
Python
135 lines
4.3 KiB
Python
from abc import ABC, abstractmethod
|
|
|
|
import re
|
|
|
|
import giturlparse
|
|
import requests
|
|
|
|
from git_flow import SUPPORTED_REMOTE_URL_SCHEMAS, GitFlowError
|
|
from git_flow.git import Commit, Git
|
|
|
|
|
|
class RemoteAPI(ABC):
|
|
def __init__(self, repository: str, token: str) -> None:
|
|
self.repository = repository
|
|
self.token = token
|
|
|
|
@staticmethod
|
|
def parse(remote: str):
|
|
"""Soportamos 3 tipos de URL: http, https y ssh. Si no se especifica el esquema, asumimos ssh:
|
|
Ejemplos:
|
|
- http://gitea/user/repo
|
|
- https://gitea.com/user/repo.git
|
|
- git@bitbucket.org:user/repo
|
|
- git.mygiteainstance.com:user/repo.git
|
|
- ssh://git@git.mygiteainstance.com:1234/user/repo.git
|
|
"""
|
|
url = Git("remote", "get-url", remote).firstline()
|
|
|
|
parsed_url = giturlparse.parse(url) # Validamos que la URL sea válida
|
|
|
|
if not parsed_url.valid:
|
|
raise GitFlowError(f"La URL del remoto {remote} es inválida: {url}")
|
|
|
|
if not parsed_url.protocol in SUPPORTED_REMOTE_URL_SCHEMAS:
|
|
raise GitFlowError(
|
|
f"La URL del remoto {remote} no tiene un esquema soportado: {url}"
|
|
)
|
|
|
|
# Cuando el host es seteado por pipeline puede ser "http://gitea/OWNER/REPO", en ese caso
|
|
# la libreria parsea OWNER="" y REPO="OWNER/REPO"
|
|
# En GitLab, los proyectos pueden estar dentro de grupos y subgrupos (ej:
|
|
# owner/grupo/subgrupo/repo), que la libreria expone por separado en groups_path
|
|
repository = (
|
|
"/".join(
|
|
filter(
|
|
None,
|
|
[parsed_url.owner, parsed_url.groups_path, parsed_url.repo],
|
|
)
|
|
)
|
|
.removeprefix("/")
|
|
.removesuffix(".git")
|
|
)
|
|
|
|
return [parsed_url.protocol + "://", parsed_url.host, repository]
|
|
|
|
@abstractmethod
|
|
def create_pull_request(
|
|
self,
|
|
source: str,
|
|
destination: str,
|
|
title: str | None = None,
|
|
description: str | None = None,
|
|
close_source_branch: bool = True,
|
|
) -> str:
|
|
pass
|
|
|
|
@abstractmethod
|
|
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 list_tags(self, suffix: str = "") -> list[str]:
|
|
"""Nombres de los tags del repositorio, sin orden garantizado. `suffix` es solo una pista
|
|
para que la API filtre; el filtrado definitivo se hace en `get_latest_tag`."""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def get_endpoint(self, resource: str) -> str:
|
|
pass
|
|
|
|
def get_headers(self):
|
|
return {
|
|
"Accept": "application/json",
|
|
"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_latest_tag(self, suffix: str = "") -> str | None:
|
|
"""Tag SemVer más alto (vX.Y.Z + suffix exacto), o None si no hay ninguno."""
|
|
pattern = re.compile(rf"^v(\d+)\.(\d+)\.(\d+){re.escape(suffix)}$")
|
|
versions = {}
|
|
|
|
for name in self.list_tags(suffix):
|
|
match = pattern.match(name)
|
|
|
|
if match:
|
|
versions[tuple(map(int, match.groups()))] = name
|
|
|
|
return versions[max(versions)] if versions else None
|
|
|
|
def get_all_pages(
|
|
self,
|
|
url: str,
|
|
page_size: int = 50,
|
|
params: dict | None = None,
|
|
size_param: str = "per_page",
|
|
) -> list:
|
|
"""Recorre un endpoint paginado por número de página (page / per_page)."""
|
|
items: list = []
|
|
page = 1
|
|
|
|
while True:
|
|
query = {**(params or {}), "page": page, size_param: page_size}
|
|
batch = self.get_json(url, query).json()
|
|
items += batch
|
|
|
|
if len(batch) < page_size:
|
|
return items
|
|
|
|
page += 1
|