81 lines
2.7 KiB
Python
81 lines
2.7 KiB
Python
import os
|
|
|
|
from dotenv import load_dotenv
|
|
from requests import get, put
|
|
|
|
from constants import RELEASE_STATUS_ID, VETRO_PROJECT_ID, STAND_FIELD_ID, CHECK_STATUS_ID
|
|
from utils import get_current_tuesday, date_from_iso, get_responsible
|
|
|
|
load_dotenv()
|
|
|
|
SITE_URL = os.getenv("SITE_URL")
|
|
KEY = os.getenv("KEY")
|
|
|
|
class RedmineAPI:
|
|
def __init__(self, base: str, key: str) -> None:
|
|
self._base = base
|
|
self._key = key
|
|
self._headers = {"X-Redmine-API-Key": self._key}
|
|
|
|
def put(self, *args, **kwargs) -> bool:
|
|
if len(args) > 0:
|
|
args = tuple([self._base + list(args)[0]] + list(args)[1:])
|
|
|
|
elif "url" in kwargs:
|
|
kwargs["url"] = self._base + kwargs["url"]
|
|
|
|
kwargs["headers"] = self._headers
|
|
|
|
return put(*args, **kwargs).status_code == 204
|
|
|
|
def get(self, *args, **kwargs) -> dict:
|
|
if len(args) > 0:
|
|
args = tuple([self._base + list(args)[0]] + list(args)[1:])
|
|
|
|
elif "url" in kwargs:
|
|
kwargs["url"] = self._base + kwargs["url"]
|
|
|
|
kwargs["headers"] = self._headers
|
|
|
|
return get(*args, **kwargs).json()
|
|
|
|
def get_issue(self, issue_id: int) -> dict:
|
|
result = self.get(f"/issues/{issue_id}.json")
|
|
return result["issue"]
|
|
|
|
def get_release_issues(self, version_id: int) -> list[dict]:
|
|
result = self.get(
|
|
"/issues.json",
|
|
params={"assigned_to_id": "me", "status_id": RELEASE_STATUS_ID, "fixed_version_id": version_id}
|
|
)
|
|
return result["issues"]
|
|
|
|
def get_versions(self) -> list[dict]:
|
|
result = self.get(f"/projects/{VETRO_PROJECT_ID}/versions.json")
|
|
return result["versions"]
|
|
|
|
def get_current_version(self) -> dict | None:
|
|
versions = self.get_versions()
|
|
tuesday = get_current_tuesday()
|
|
return next(filter(lambda version: date_from_iso(version["created_on"]) == tuesday, versions), None)
|
|
|
|
def get_user(self, last_name: str, first_name: str) -> dict | None:
|
|
result = self.get("/users.json", params={"status": 1, "lastname": last_name, "firstname": first_name})
|
|
return next(iter(result["users"]), None)
|
|
|
|
def edit_issue(self, issue_id: int, data: dict) -> bool:
|
|
return self.put(f"/issues/{issue_id}.json", json={"issue": data})
|
|
|
|
def release_issue(self, issue_id: int) -> bool:
|
|
issue = self.get_issue(issue_id)
|
|
last_name, first_name = get_responsible(issue)
|
|
user = api.get_user(last_name=last_name, first_name=first_name)
|
|
assigned_to_id = user["id"]
|
|
return self.edit_issue(issue_id, data={
|
|
"assigned_to_id": assigned_to_id,
|
|
"status_id": CHECK_STATUS_ID,
|
|
"custom_fields": [{"id": STAND_FIELD_ID, "value": "prod"}]
|
|
})
|
|
|
|
api = RedmineAPI(SITE_URL, KEY)
|