workflow-scripts/lib/gitea.py

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

    @staticmethod
    def parse_result(result) -> dict:
        """Parses the result-JSON to a dict."""
        if result.text and len(result.text) > 3:
            return json.loads(result.text)
        return {}

    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"])