97 lines
3.2 KiB
Python
97 lines
3.2 KiB
Python
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
|
|
|
|
|
|
class GitlabRemoteAPI(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("/merge_requests"),
|
|
headers=self.get_headers(),
|
|
json={
|
|
"title": title,
|
|
"description": description,
|
|
"source_branch": source,
|
|
"target_branch": destination,
|
|
"remove_source_branch": close_source_branch,
|
|
},
|
|
)
|
|
|
|
if response.ok:
|
|
return "PR creado exitosamente: " + response.json()["web_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("/repository/tags"),
|
|
headers=self.get_headers(),
|
|
json={"tag_name": tag, "ref": commit},
|
|
)
|
|
|
|
if response.ok:
|
|
return "Tag creado exitosamente: " + tag
|
|
else:
|
|
raise GitFlowError("Ocurrió un error al crear el tag: " + response.text)
|
|
|
|
def get_merge_commits(self, merge_commit: str) -> list[Commit]:
|
|
mr = self._get_merge_request(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_merge_source_branch(self, merge_commit: str) -> str:
|
|
return self._get_merge_request(merge_commit)["source_branch"]
|
|
|
|
def _get_merge_request(self, merge_commit: str) -> dict:
|
|
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}")
|
|
|
|
return mr
|
|
|
|
def list_tags(self, suffix: str = "") -> list[str]:
|
|
# "search" soporta "$" para anclar al final del nombre
|
|
params = {"search": suffix + "$"} if suffix else None
|
|
tags = self.get_all_pages(self.get_endpoint("/repository/tags"), 100, params)
|
|
|
|
return [t["name"] for t in tags]
|
|
|
|
def get_endpoint(self, resource: str) -> str:
|
|
project_id = quote(self.repository, safe="")
|
|
return f"{self.base_url}/api/v4/projects/{project_id}{resource}"
|
|
|
|
def get_headers(self):
|
|
return {
|
|
"Accept": "application/json",
|
|
"Content-Type": "application/json",
|
|
"PRIVATE-TOKEN": self.token,
|
|
}
|