105 lines
3.5 KiB
Python
105 lines
3.5 KiB
Python
import re
|
|
|
|
import requests
|
|
from git_flow import GitFlowError
|
|
from git_flow.git import Commit
|
|
from git_flow.remote.base import RemoteAPI
|
|
|
|
|
|
class BitbucketRemoteAPI(RemoteAPI):
|
|
API_ENDPOINT = "https://api.bitbucket.org/2.0/repositories"
|
|
|
|
def create_pull_request(
|
|
self,
|
|
source: str,
|
|
destination: str,
|
|
title: str | None = None,
|
|
description: str | None = None,
|
|
close_source_branch: bool = True
|
|
):
|
|
request = requests.post(
|
|
self.get_endpoint("/pullrequests"),
|
|
headers=self.get_headers(),
|
|
json={
|
|
"title": title,
|
|
"description": description,
|
|
"source": {"branch": {"name": source}},
|
|
"destination": {"branch": {"name": destination}},
|
|
"close_source_branch": close_source_branch,
|
|
},
|
|
)
|
|
|
|
if request.ok:
|
|
json = request.json()
|
|
return "PR creado exitosamente: " + json["links"]["html"]["href"]
|
|
else:
|
|
raise GitFlowError("Ocurrió un error al crear el PR: " + request.text)
|
|
|
|
def create_tag(self, tag: str, commit: str):
|
|
response = requests.post(
|
|
self.get_endpoint("/refs/tags"),
|
|
headers=self.get_headers(),
|
|
json={"name": tag, "target": {"hash": 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]:
|
|
pr = self._get_merge_pull_request(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_merge_source_branch(self, merge_commit: str) -> str:
|
|
return self._get_merge_pull_request(merge_commit)["source"]["branch"]["name"]
|
|
|
|
def _get_merge_pull_request(self, merge_commit: str) -> dict:
|
|
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}")
|
|
|
|
return pr
|
|
|
|
def list_tags(self, suffix: str = "") -> list[str]:
|
|
names: list[str] = []
|
|
url: str | None = self.get_endpoint("refs/tags")
|
|
params: dict[str, int|str] | None = {"pagelen": 100}
|
|
|
|
if suffix:
|
|
params["q"] = f'name ~ "{suffix}"'
|
|
|
|
while url:
|
|
page = self.get_json(url, params).json()
|
|
names += [t["name"] for t in page["values"]]
|
|
url = page.get("next")
|
|
params = None # "next" ya incluye la query
|
|
|
|
return names
|
|
|
|
def get_endpoint(self, resource: str) -> str:
|
|
return f"{BitbucketRemoteAPI.API_ENDPOINT}/{self.repository}/{resource}"
|