Merge pull request 'feature: add gitea remote api' (#1) from feature/add-gitea-remote-api into main

Reviewed-on: #1

.flowconfig

@@ -2,3 +2,5 @@
 	remote = origin
 	branches = main
 	initialized = true
+	version = 2
+	remote-type = gitea

src/git_flow/__init__.py

@@ -51,3 +51,5 @@ SUPPORTED_REMOTE_APIS = [
 REPOSITORY_TOKEN_FILENAME = ".repository-token"
 
 FLOWCONFIG_FILENAME = ".flowconfig"
+
+FLOWCONFIG_VERSION = 2

src/git_flow/command/base.py

@@ -2,12 +2,14 @@ from git_flow import (
     BRANCH_TYPES,
     REPOSITORY_TOKEN_FILENAME,
     FLOWCONFIG_FILENAME,
+    FLOWCONFIG_VERSION,
     GitFlowError,
 )
 from git_flow.git import Git
 from git_flow.remote.base import RemoteAPI
 from git_flow.remote.bitbucket import BitbucketRemoteAPI
 from git_flow.remote.github import GithubRemoteAPI
+from git_flow.remote.gitea import GiteaRemoteAPI
 from git_flow.io import *
 from os.path import isfile
 
@@ -32,6 +34,14 @@ def ensure_initialized():
             f"El valor de `flow.initialized` ({initialized}) es inválido."
         )
 
+    version = int(flowconfig.get("flow.version", "1"))
+
+    if version != FLOWCONFIG_VERSION:
+        raise GitFlowError(
+            f"La versión del flowconfig ({version}) no es compatible con esta versión de git-flow ({FLOWCONFIG_VERSION}). "
+            f"Debe reinicializar el repositorio."
+        )
+
 
 def ensure_right_branch():
     branch = Git.get_current_branch()
@@ -104,14 +114,18 @@ def get_remote_api(token: str) -> RemoteAPI:
     if "flow.remote" not in flowconfig:
         raise GitFlowError("El repositorio no tiene configurado un remoto.")
 
+    remote_type = flowconfig.get("flow.remote-type", "").lower()
     [host, repository] = RemoteAPI.parse(flowconfig["flow.remote"])
 
-    if host == "bitbucket.org":
+    if remote_type == "gitea":
+        return GiteaRemoteAPI("https://" + host, repository, token)
+    elif remote_type == "bitbucket":
         return BitbucketRemoteAPI(repository, token)
-    elif host == "github.com":
+    elif remote_type == "github":
         return GithubRemoteAPI(repository, token)
     else:
-        raise GitFlowError("El host del repositorio remoto es inválido")
+        raise GitFlowError("El tipo del remoto es inválido")
+
 
 def is_valid_ticket(ticket: str) -> bool:
     """Check if the specified string is a valid ticket number (ABC-123)"""

src/git_flow/command/init.py

@@ -1,6 +1,6 @@
 import typer
 
-from git_flow import FLOWCONFIG_FILENAME, GitFlowError
+from git_flow import FLOWCONFIG_FILENAME, FLOWCONFIG_VERSION, GitFlowError
 from git_flow.git import Git
 from git_flow.command.base import *
 
@@ -15,15 +15,17 @@ def init():
     _ensure_not_already_initialized()
 
     branches = _setup_flow_branches()
-    remote = _setup_flow_remote()
+    (remote, remote_type) = _setup_flow_remote()
 
     flowconfig = {
+        "flow.version": str(FLOWCONFIG_VERSION),
         "flow.initialized": "true",
         "flow.branches": ",".join(branches),
     }
 
-    if remote:
+    if remote and remote_type:
         flowconfig["flow.remote"] = remote
+        flowconfig["flow.remote-type"] = remote_type
 
     Git.set_config(flowconfig, FLOWCONFIG_FILENAME)
     Git("add", FLOWCONFIG_FILENAME).exec()
@@ -74,6 +76,7 @@ def _setup_flow_remote():
     )
 
     remote = None
+    remote_type = None
 
     if confirm("¿Configurar repositorio remoto?"):
         remotes = Git("remote").lines(print="Listando remotos disponibles")
@@ -93,7 +96,9 @@ def _setup_flow_remote():
             info("Tiene más de un remoto, seleccione el que va a utilizar.")
             remote = choice("Remoto", remotes)
 
-    return remote
+        remote_type = choice("Tipo de remoto:", ["bitbucket", "github", "gitea"])
+
+    return (remote, remote_type)
 
 
 def _ensure_all_flow_branches_exist(branches: list[str], remote: str | None):

src/git_flow/remote/base.py

@@ -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,

src/git_flow/remote/gitea.py

@@ -0,0 +1,55 @@
+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}"
+    
+    def get_headers(self):
+        return {
+            "Accept": "application/json",
+            "Content-Type": "application/json",
+            "Authorization": "token " + self.token,
+        }