feature: agrego soporte para apis gitea

This commit is contained in:
jt
2026-07-08 21:12:06 -03:00
parent c93833769b
commit 87c5ba81c4
6 changed files with 82 additions and 31 deletions
+4 -24
View File
@@ -12,43 +12,23 @@ class RemoteAPI(ABC):
@staticmethod
def parse(remote: str):
url = Git("remote", "get-url", remote).firstline()
invalid_url_error = GitFlowError(f"La URL del remoto {remote} es inválida: {url}")
url_type = None
if url.startswith("git@"):
url_type = "ssh"
elif url.startswith("http://") or url.startswith("https://"):
url_type = "https"
else:
raise invalid_url_error
if url_type == "ssh":
# git@host:user/repo[.git]
at_pos = url.index("@")
colon_pos = url.index(":")
host = url[at_pos+1:colon_pos]
repository = url[colon_pos+1:]
else:
# https://host/user/repo[.git]
elif url.startswith("http://") or url.startswith("https://"):
host_start = url.index("://") + 3
uri_start = url.index("/", host_start)
host = url[host_start:uri_start]
repository = url[uri_start+1:]
else:
raise GitFlowError(f"La URL del remoto {remote} es inválida: {url}")
# Esto permite configurar hosts "imaginarios" para permitir el uso de multiples claves ssh
if host.endswith(".bitbucket.org"):
host = "bitbucket.org"
elif host.endswith(".github.com"):
host = "github.com"
if host not in SUPPORTED_REMOTE_APIS:
raise GitFlowError("El host del repositorio remoto es inválido")
repository.removesuffix(".git")
repository = repository.removesuffix(".git")
return [host, repository]
@abstractmethod
def create_pull_request(
self,
+48
View File
@@ -0,0 +1,48 @@
import requests
from git_flow import GitFlowError
from git_flow.remote.base import RemoteAPI
class GiteaRemoteAPI(RemoteAPI):
def __init__(self, base_url: str, repository: str, token: str) -> None:
super().__init__(repository, token)
self.base_url = base_url
def create_pull_request(
self,
source: str,
destination: str,
title: str | None = None,
description: str | None = None,
close_source_branch: bool = True
) -> str:
response = requests.post(
self.get_endpoint("/pulls"),
headers=self.get_headers(),
json={
"title": title,
"body": description,
"head": source,
"base": destination,
},
)
if response.ok:
return "PR creado exitosamente: " + response.json()["html_url"]
else:
raise GitFlowError("Ocurrió un error al crear el PR: " + response.text)
def create_tag(self, tag: str, commit: str) -> str:
response = requests.post(
self.get_endpoint("/tags"),
headers=self.get_headers(),
json={"tag_name": tag, "target": commit},
)
if response.ok:
return "Tag creado exitosamente: " + tag
else:
raise GitFlowError("Ocurrió un error al crear el tag: " + response.text)
def get_endpoint(self, resource: str) -> str:
return f"{self.base_url}/api/v1/repos/{self.repository}{resource}"