complete refactor to use Gitea class api wrapper

lib/exceptions.py

@@ -0,0 +1,6 @@
+class AlreadyExistsException(Exception):
+    pass
+
+
+class NotFoundException(Exception):
+    pass

lib/gitea.py

@@ -0,0 +1,160 @@
+import logging
+import json
+import requests
+from immutabledict import immutabledict
+
+from .exceptions import NotFoundException, ConflictException, AlreadyExistsException
+
+class Gitea:
+    """Object to establish a session with Gitea"""
+
+    def __init__(
+        self,
+        instance_url: str,
+        token: str,
+        log_level: str,
+    ):
+        """Initializing Gitea-instance
+
+        Args:
+            instance_url (str): The Gitea instance URL.
+            token (str): The access token.
+            log_level (str): The log level.
+        """
+
+        self.logger = logging.getLogger(__name__)
+        self.logger.setLevel(log_level)
+        self.headers = {
+            "Authorization": f"token {token}",
+            "Accept": "application/json",
+            "Content-Type": "application/json",
+        }
+        self.url = instance_url
+        self.requests = requests.Session()
+
+    def __get_url(self, endpoint):
+        url = self.url + "/api/v1" + endpoint
+        self.logger.debug(">> Gitea URL: %s" % url)
+        return url
+
+    def _requests_get(self, endpoint: str, params=immutabledict()) -> requests.Response:
+        request = self.requests.get(
+            self.__get_url(endpoint), headers=self.headers, params=params
+        )
+        if request.status_code not in [200, 201]:
+            message = f">> Received status code: {request.status_code} ({request.url})"
+            if request.status_code in [404]:
+                raise NotFoundException(message)
+            if request.status_code in [403]:
+                raise Exception(
+                    f">> Unauthorized: {request.url} - Check your permissions and try again! ({message})"
+                )
+            if request.status_code in [409]:
+                raise ConflictException(message)
+            raise Exception(message)
+        return request
+
+    def requests_get(self, endpoint: str, params=immutabledict()) -> dict:
+        request = self._requests_get(endpoint, params)
+        return self.parse_result(request)
+
+    def requests_put(self, endpoint: str, data: dict = None):
+        if not data:
+            data = {}
+        request = self.requests.put(
+            self.__get_url(endpoint), headers=self.headers, data=json.dumps(data)
+        )
+        if request.status_code not in [200, 204]:
+            message = f">> Received status code: {request.status_code} ({request.url}) {request.text}"
+            self.logger.error(message)
+            raise Exception(message)
+
+    def requests_post(self, endpoint: str, data: dict):
+        request = self.requests.post(
+            self.__get_url(endpoint), headers=self.headers, data=json.dumps(data)
+        )
+        if request.status_code not in [200, 201, 202]:
+            if (
+                "already exists" in request.text
+            ):
+                self.logger.warning(request.text)
+                raise AlreadyExistsException()
+            self.logger.error(
+                f">> Received status code: {request.status_code} ({request.url})"
+            )
+            self.logger.error(f">> With info: {data} ({self.headers})")
+            self.logger.error(f">> Answer: {request.text}")
+            raise Exception(
+                f">> Received status code: {request.status_code} ({request.url}), {request.text}"
+            )
+        return self.parse_result(request)
+
+    def get_version(self) -> str:
+        path = "/version"
+        result = self.requests_get(path)
+        self.logger.debug(f">> Gitea Version: {result["version"]}")
+        return result["version"]
+
+    def get_issues(
+        self,
+        owner: str,
+        repository: str,
+    ):
+        path = f"/repos/{owner}/{repository}/issues"
+        params = {"state": "open"}
+        self.logger.debug(f">> Path used to get issues: {path}")
+
+        result = self.requests_get(path, params)
+        self.logger.debug(">> Gitea response: %s", result)
+        return result
+
+    def get_pull_requests(
+        self,
+        owner: str,
+        repository: str,
+    ):
+        path = f"/repos/{owner}/{repository}/pulls"
+        params = {"state": "open"}
+        self.logger.debug(f">> Path used to get pull requests: {path}")
+
+        result = self.requests_get(path, params)
+        self.logger.debug(">> Gitea response: %s", result)
+        return result
+
+    def update_issue_labels(
+        self,
+        issue_number: int,
+        labels: list,
+        owner: str,
+        repository: str,
+    ):
+        path = f"/repos/{owner}/{repository}/issues/{issue_number}/labels"
+        data = {"labels": labels}
+        self.logger.debug(f">> Path used to update issue label: {path}")
+
+        result = self.requests_post(path, data)
+        if "id" in result:
+            self.logger.info(">> Successfully added label")
+            self.logger.debug(">> Gitea response: %s", result)
+        else:
+            self.logger.error(result["message"])
+            # raise Exception("User not created... (gitea: %s)" % result["message"])
+
+    def update_pull_request_labels(
+        self,
+        pull_request_number: int,
+        labels: list,
+        owner: str,
+        repository: str,
+    ):
+        path = f"/repos/{owner}/{repository}/pulls/{pull_request_number}/labels"
+        data = {"labels": labels}
+        self.logger.debug(f">> Path used to update pull request labels: {path}")
+
+        result = self.requests_post(path, data)
+        if "id" in result:
+            self.logger.info(">> Successfully added label")
+            self.logger.debug(">> Gitea response: %s", result)
+        else:
+            self.logger.error(result["message"])
+            # raise Exception("User not created... (gitea: %s)" % result["message"])