apps-store/github_connector/models/github.py

# Copyright (C) 2016-Today: Odoo Community Association (OCA)
# @author: Sylvain LE GAL (https://twitter.com/legalsylvain)
# @author: Sébastien BEAU <sebastien.beau@akretion.com>
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl.html).


import json
import logging

import requests
from requests.auth import AuthBase, HTTPBasicAuth

from odoo import _, exceptions

_logger = logging.getLogger(__name__)

_MAX_NUMBER_REQUEST = 30

_BASE_URL = "https://api.github.com/"

_GITHUB_URL = "https://github.com/"

_GITHUB_TYPE = [
    ("organization", "Organization"),
    ("repository", "Repository"),
    ("user", "User"),
]

_GITHUB_TYPE_URL = {
    "organization": {"url_get_by_name": "orgs/%s"},
    "user": {"url_get_by_id": "user/%s", "url_get_by_name": "users/%s"},
    "repository": {
        "url_get_by_id": "repositories/%s",
        "url_get_by_name": "repos/%s",
        "url_create": "orgs/%s/repos",
    },
    "team": {"url_get_by_id": "teams/%s", "url_create": "orgs/%s/teams"},
    "organization_members": {"url_get_by_name": "orgs/%s/members"},
    "organization_repositories": {"url_get_by_name": "orgs/%s/repos"},
    "organization_teams": {"url_get_by_name": "orgs/%s/teams"},
    "team_members_member": {"url_get_by_name": "teams/%s/members?role=member"},
    "team_members_maintainer": {"url_get_by_name": "teams/%s/members?role=maintainer"},
    "team_repositories": {"url_get_by_name": "teams/%s/repos"},
    "repository_branches": {"url_get_by_name": "repos/%s/branches"},
}

_CODE_401 = 401
_CODE_403 = 403
_CODE_422 = 422
_CODE_200 = 200
_CODE_201 = 201


class Github(object):
    def __init__(self, github_type, login, password, max_try, token=None):
        super().__init__()
        self.github_type = github_type
        self.max_try = max_try
        if token:
            self.auth = HTTPTokenAuth(token)
        else:
            self.auth = HTTPBasicAuth(login, password)

    def _build_url(self, arguments, url_type, page):
        arguments = arguments and arguments or {}
        url = _GITHUB_TYPE_URL[self.github_type][url_type]
        if self.github_type not in _GITHUB_TYPE_URL.keys():
            raise exceptions.Warning(_("'%s' is not implemented.") % self.github_type)
        complete_url = _BASE_URL + url % tuple(arguments)

        if page:
            complete_url += (
                "?" in complete_url and "&" or "?"
            ) + "per_page=%d&page=%d" % (_MAX_NUMBER_REQUEST, page)
        return complete_url

    def list(self, arguments):
        page = 1
        datas = []
        while True:
            pending_datas = self.get(arguments, False, page)
            datas += pending_datas
            if pending_datas == [] or len(pending_datas) < _MAX_NUMBER_REQUEST:
                break
            page += 1
        return datas

    def get_by_url(self, url, call_type, data=False):
        _logger.info("Calling %s" % url)
        for i in range(self.max_try):
            try:
                if call_type == "get":
                    response = requests.get(url, auth=self.auth)
                    break
                elif call_type == "post":
                    json_data = json.dumps(data)
                    response = requests.post(url, auth=self.auth, data=json_data)
                    break
            except Exception as err:
                _logger.warning(
                    "URL Call Error. %d/%d. URL: %s", i, self.max_try, err.__str__(),
                )
        else:
            raise exceptions.Warning(_("Maximum attempts reached."))

        if isinstance(self.auth, HTTPBasicAuth):
            auth_err_msg = _("login '%s'") % self.auth.login
        elif isinstance(self.auth, HTTPTokenAuth):
            auth_err_msg = _("provided token")
        if response.status_code == _CODE_401:
            raise exceptions.Warning(
                _(
                    "401 - Unable to authenticate to Github with the %s.\n"
                    "You should check your credentials in the Odoo"
                    " configuration file."
                )
                % auth_err_msg
            )
        elif response.status_code == _CODE_403:
            raise exceptions.Warning(
                _(
                    "Unable to realize the current operation. The %s"
                    " does not have the correct access rights."
                )
                % auth_err_msg
            )
        elif response.status_code == _CODE_422:
            raise exceptions.Warning(
                _(
                    "Unable to realize the current operation. Possible reasons:\n"
                    " * You try to create a duplicated item\n"
                    " * Some of the arguments are incorrect"
                )
            )
        elif response.status_code not in [_CODE_200, _CODE_201]:
            raise exceptions.Warning(
                _("The call to '%s' failed:\n" "- Status Code: %d\n" "- Reason: %s")
                % (response.url, response.status_code, response.reason)
            )
        return response.json()

    def get(self, arguments, by_id=False, page=None):
        url = self._build_url(
            arguments, by_id and "url_get_by_id" or "url_get_by_name", page
        )
        return self.get_by_url(url, "get")

    def create(self, arguments, data):
        # pylint: disable=method-required-super
        url = self._build_url(arguments, "url_create", None)
        res = self.get_by_url(url, "post", data)
        return res


class HTTPTokenAuth(AuthBase):

    header_format_str = "token {}"

    def __init__(self, token):
        self.token = token

    def __call__(self, request):
        request.headers["Authorization"] = self.header_format_str.format(self.token)
        return request