71 lines
2.3 KiB
Python
71 lines
2.3 KiB
Python
import requests
|
|
from git_flow import GitFlowError
|
|
from git_flow.git import Commit
|
|
from git_flow.remote.base import RemoteAPI
|
|
|
|
|
|
class GithubRemoteAPI(RemoteAPI):
|
|
API_ENDPOINT = "https://api.github.com/repos"
|
|
|
|
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("/git/refs"),
|
|
headers=self.get_headers(),
|
|
json={"ref": f"refs/tags/{tag}", "sha": 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]:
|
|
prs = self.get_json(self.get_endpoint(f"/commits/{merge_commit}/pulls")).json()
|
|
pr = next((p for p in prs if p.get("merge_commit_sha") == merge_commit), None)
|
|
|
|
if not pr:
|
|
raise GitFlowError(f"No se encontró un PR para el commit {merge_commit}")
|
|
|
|
items = self.get_all_pages(self.get_endpoint(f"/pulls/{pr['number']}/commits"), 100)
|
|
|
|
# La API los devuelve del más viejo al más nuevo
|
|
return [
|
|
Commit(
|
|
c["sha"][:7],
|
|
c["commit"]["message"].splitlines()[0],
|
|
c["commit"]["author"]["email"],
|
|
len(c["parents"]) > 1,
|
|
)
|
|
for c in reversed(items)
|
|
]
|
|
|
|
def list_tags(self, suffix: str = "") -> list[str]:
|
|
return [t["name"] for t in self.get_all_pages(self.get_endpoint("/tags"), 100)]
|
|
|
|
def get_endpoint(self, resource: str) -> str:
|
|
return f"{GithubRemoteAPI.API_ENDPOINT}/{self.repository}{resource}"
|