61 lines
3.3 KiB
Python
61 lines
3.3 KiB
Python
from __future__ import annotations
|
|
import json, urllib.error, urllib.parse, urllib.request
|
|
from typing import Any
|
|
|
|
class GiteaApiError(RuntimeError):
|
|
def __init__(self, code: str, status: int | None=None):
|
|
super().__init__(code); self.code=code; self.status=status
|
|
|
|
class GiteaApi:
|
|
def __init__(self, base_url: str, token: str):
|
|
self.base_url=base_url.rstrip("/"); self.token=token
|
|
def request(self, method: str, path: str, body: dict[str,Any] | None=None,
|
|
expected: tuple[int,...]=(200,)) -> Any:
|
|
raw=None if body is None else json.dumps(body).encode()
|
|
req=urllib.request.Request(self.base_url+path,data=raw,method=method)
|
|
req.add_header("Authorization",f"token {self.token}")
|
|
req.add_header("Accept","application/json")
|
|
if raw is not None: req.add_header("Content-Type","application/json")
|
|
try:
|
|
with urllib.request.urlopen(req,timeout=20) as resp:
|
|
status,payload=resp.status,resp.read()
|
|
except urllib.error.HTTPError as exc:
|
|
raise GiteaApiError(f"GITEA_HTTP_{exc.code}",exc.code) from exc
|
|
except urllib.error.URLError as exc:
|
|
raise GiteaApiError("GITEA_UNAVAILABLE") from exc
|
|
if status not in expected:
|
|
raise GiteaApiError(f"GITEA_HTTP_{status}",status)
|
|
if not payload: return None
|
|
try: return json.loads(payload)
|
|
except json.JSONDecodeError as exc:
|
|
raise GiteaApiError("GITEA_RESPONSE_INVALID") from exc
|
|
def branch(self, owner: str, repo: str, branch: str) -> dict[str,Any]:
|
|
value=self.request("GET",f"/api/v1/repos/{owner}/{repo}/branches/{urllib.parse.quote(branch,safe='')}")
|
|
if not isinstance(value,dict): raise GiteaApiError("GITEA_BRANCH_RESPONSE_INVALID")
|
|
return value
|
|
def list_open_pulls(self, owner: str, repo: str) -> list[dict[str,Any]]:
|
|
value=self.request("GET",f"/api/v1/repos/{owner}/{repo}/pulls?state=open&limit=50")
|
|
if not isinstance(value,list): raise GiteaApiError("GITEA_PULL_LIST_INVALID")
|
|
return [v for v in value if isinstance(v,dict)]
|
|
def create_pull(self, owner: str, repo: str, *, head: str, base: str,
|
|
title: str, body: str) -> dict[str,Any]:
|
|
value=self.request("POST",f"/api/v1/repos/{owner}/{repo}/pulls",
|
|
{"head":head,"base":base,"title":title,"body":body},expected=(201,))
|
|
if not isinstance(value,dict): raise GiteaApiError("GITEA_PULL_CREATE_INVALID")
|
|
return value
|
|
def pull(self, owner: str, repo: str, index: int) -> dict[str,Any]:
|
|
value=self.request("GET",f"/api/v1/repos/{owner}/{repo}/pulls/{index}")
|
|
if not isinstance(value,dict): raise GiteaApiError("GITEA_PULL_RESPONSE_INVALID")
|
|
return value
|
|
def request_reviewers(self, owner: str, repo: str, index: int, reviewers: list[str]) -> None:
|
|
try:
|
|
self.request("POST",f"/api/v1/repos/{owner}/{repo}/pulls/{index}/requested_reviewers",
|
|
{"reviewers":reviewers},expected=(201,204))
|
|
except GiteaApiError as exc:
|
|
if exc.status != 422:
|
|
raise
|
|
pull=self.pull(owner,repo,index)
|
|
present={item.get("login") for item in pull.get("requested_reviewers",[]) if isinstance(item,dict)}
|
|
if not set(reviewers).issubset(present):
|
|
raise
|