Source code for gerrit.base

#!/usr/bin/env python
# -*- coding:utf-8 -*-
# @Author: Jialiang Shi
import logging
import netrc
from typing import Any, Dict, Optional, Tuple, Union
import requests
from requests.adapters import HTTPAdapter
from requests import Session
from gerrit.utils.requester import Requester
from gerrit.utils.common import decode_response, strip_trailing_slash

logger = logging.getLogger(__name__)




class GerritClient:
    """
    Python wrapper for the Gerrit V3.x REST API.

    """

    default_headers: Dict[str, str] = {"Content-Type": "application/json; charset=UTF-8"}

    def __init__(
        self,
        base_url: str,
        username: Optional[str] = None,
        password: Optional[str] = None,
        use_netrc: bool = False,
        ssl_verify: Union[bool, str] = True,
        cert: Optional[Union[str, Tuple[str, str]]] = None,
        cookies: Optional[Dict[str, str]] = None,
        cookie_jar: Optional[Any] = None,
        timeout: int = 60,
        max_retries: Optional[int] = None,
        session: Optional[Session] = None,
        auth_suffix: str = "/a",
    ) -> None:
        self._base_url = strip_trailing_slash(base_url)

        # make request session if one isn't provided
        if session is None:
            session = requests.Session()

        if use_netrc:
            password = self.get_password_from_netrc_file()

        if username and password:
            session.auth = (username, password)

        if ssl_verify:
            session.verify = ssl_verify

        if cert is not None:
            session.cert = cert

        if cookies is not None:
            session.cookies.update(cookies)

        if cookie_jar is not None:
            session.cookies = cookie_jar

        if max_retries is not None:
            retry_adapter = HTTPAdapter(max_retries=max_retries)
            session.mount("http://", retry_adapter)
            session.mount("https://", retry_adapter)

        self.session = session

        self.requester = Requester(
            base_url=base_url,
            session=self.session,
            timeout=timeout,
        )
        if self.session.auth is not None:
            self.auth_suffix = auth_suffix
        else:
            self.auth_suffix = ""



    def get_password_from_netrc_file(self) -> str:
        """
        Providing the password form .netrc file for getting Host name.
        :return: The related password from .netrc file as a string.
        """

        netrc_client = netrc.netrc()
        auth_tokens = netrc_client.authenticators(self._base_url)
        if not auth_tokens:
            raise ValueError(
                f"The '{self._base_url}' host name is not found in netrc file."
            )
        return auth_tokens[2]




    def get_endpoint_url(self, endpoint: str) -> str:
        """
        Return the complete url including host and port for a given endpoint.
        :param endpoint: service endpoint as str
        :return: complete url (including host and port) as str
        """
        return f"{self._base_url}{self.auth_suffix}{endpoint}"


    @property
    def access(self) -> Any:
        """
        Access related REST APIs

        :return:
        """
        from gerrit.access import GerritAccess
        return GerritAccess(gerrit=self)

    @property
    def config(self) -> Any:
        """
        Config related REST APIs

        :return:
        """
        from gerrit.config.config import GerritConfig
        return GerritConfig(gerrit=self)

    @property
    def projects(self) -> Any:
        """
        Project related REST APIs
        :return:
        """
        from gerrit.projects.projects import GerritProjects
        return GerritProjects(gerrit=self)

    @property
    def changes(self) -> Any:
        """
        Change related REST APIs

        :return:
        """
        from gerrit.changes.changes import GerritChanges
        return GerritChanges(gerrit=self)

    @property
    def accounts(self) -> Any:
        """
        Account related REST APIs

        :return:
        """
        from gerrit.accounts.accounts import GerritAccounts
        return GerritAccounts(gerrit=self)

    @property
    def groups(self) -> Any:
        """
        Group related REST APIs

        :return:
        """
        from gerrit.groups.groups import GerritGroups
        return GerritGroups(gerrit=self)

    @property
    def plugins(self) -> Any:
        """
        Plugin related REST APIs

        :return:
        """
        from gerrit.plugins.plugins import GerritPlugins
        return GerritPlugins(gerrit=self)

    @property
    def version(self) -> str:
        """
        get the version of the Gerrit server.

        :return:
        """
        return self.config.get_version()

    @property
    def server(self) -> Any:
        """
        get the information about the Gerrit server configuration.

        :return:
        """
        return self.config.get_server_info()



    def get(self, endpoint: str, **kwargs: Any) -> Any:
        """
        Send HTTP GET to the endpoint.

        :param endpoint: The endpoint to send to.
        :return:
        """
        url = self.get_endpoint_url(endpoint)
        logger.debug("Sending GET request to %s", url)
        response = self.requester.get(url, **kwargs)
        result = decode_response(response)
        return result




    def post(self, endpoint: str, **kwargs: Any) -> Any:
        """
        Send HTTP POST to the endpoint.

        :param endpoint: The endpoint to send to.
        :return:
        """
        url = self.get_endpoint_url(endpoint)
        logger.debug("Sending POST request to %s", url)
        response = self.requester.post(url, **kwargs)
        result = decode_response(response)
        return result




    def put(self, endpoint: str, **kwargs: Any) -> Any:
        """
        Send HTTP PUT to the endpoint.

        :param endpoint: The endpoint to send to.
        :return:
        """
        url = self.get_endpoint_url(endpoint)
        logger.debug("Sending PUT request to %s", url)
        response = self.requester.put(url, **kwargs)
        result = decode_response(response)
        return result




    def delete(self, endpoint: str) -> Any:
        """
        Send HTTP DELETE to the endpoint.

        :param endpoint: The endpoint to send to.
        :return:
        """
        url = self.get_endpoint_url(endpoint)
        logger.debug("Sending DELETE request to %s", url)
        response = self.requester.delete(url)
        return decode_response(response)