Source code for copr.client_v2.client

# coding: utf-8

from __future__ import print_function
from __future__ import unicode_literals
from __future__ import division
from __future__ import absolute_import
from abc import ABCMeta, abstractproperty

import sys
import os
import logging
import weakref
import warnings

import six
from six import with_metaclass
from six.moves import configparser

from .resources import Root
from .handlers import ProjectHandle, ProjectChrootHandle, BuildHandle, MockChrootHandle, BuildTaskHandle
from .net_client import NetClient

if sys.version_info < (2, 7):
    class NullHandler(logging.Handler):
        def emit(self, record):
            pass
else:
    from logging import NullHandler

from ..exceptions import CoprConfigException, CoprNoConfException
from ..util import UnicodeMixin

log = logging.getLogger(__name__)
log.addHandler(NullHandler())


class HandlersProvider(object, with_metaclass(ABCMeta)):

    @abstractproperty
    def projects(self):
        """
        :rtype: ProjectHandle
        """
        pass

    @abstractproperty
    def project_chroots(self):
        """
        :rtype: ProjectChrootHandle
        """
        pass

    @abstractproperty
    def builds(self):
        """
        :rtype: BuildHandle
        """
        pass

    @abstractproperty
    def build_tasks(self):
        """
        :rtype: BuildTaskHandle
        """
        pass

    @abstractproperty
    def mock_chroots(self):
        """
        :rtype: MockChrootHandle
        """
        pass


[docs]class CoprClient(UnicodeMixin, HandlersProvider): """ Main interface to the copr service :param NetClient net_client: wrapper for http requests :param unicode root_url: used as copr projects root :param bool no_config: helper flag to indicate that no config was provided Could be created: - using static method :meth:`~copr.client_v2.client.CoprClient.create_from_file_config` - using static method :meth:`~copr.client_v2.client.CoprClient.create_from_params` If you create Client directly call :meth:`CoprClient.post_init` method after the creation. """ def __init__(self, net_client, root_url=None, no_config=False): """ """ self.nc = net_client self.root_url = root_url or u"https://copr.fedorainfracloud.org" self.no_config = no_config self._post_init_done = False self.root = None self._projects = None self._project_chroots = None self._builds = None self._build_tasks = None self._mock_chroots = None warnings.warn("You are using Copr's deprecated APIv2. " "Please migrate to APIv3. " "See https://python-copr.readthedocs.io/en/latest/ClientV3.html", FutureWarning) def _check_client_init(self): if not self._post_init_done: raise RuntimeError("CoprClient wasn't initialized, use class-methods " "create_from* to get instance of CoprClient") @property def projects(self): """ :rtype: :py:class:`~copr.client_v2.handlers.ProjectHandle` """ self._check_client_init() return self._projects @property def project_chroots(self): """ :rtype: :py:class:`~copr.client_v2.handlers.ProjectChrootHandle` """ self._check_client_init() return self._project_chroots @property def builds(self): """ :rtype: :py:class:`~copr.client_v2.handlers.BuildHandle` """ self._check_client_init() return self._builds @property def build_tasks(self): """ :rtype: :py:class:`~copr.client_v2.handlers.BuildTaskHandle` """ self._check_client_init() return self._build_tasks @property def mock_chroots(self): """ :rtype: :py:class:`~copr.client_v2.handlers.MockChrootHandle` """ self._check_client_init() return self._mock_chroots def __unicode__(self): return ( u"<Copr client. api root url: {}, config provided: {}, net client: {}>" .format(self.root_url, not self.no_config, self.nc) ) @property def api_root(self): """ Url to API endpoint """ return "{0}/api_2".format(self.root_url)
[docs] @classmethod def create_from_params(cls, root_url=None, login=None, token=None): """ Create client instance using the given parameters :param str root_url: Url to the Copr service, default: "http://copr.fedoraproject.org" :param str login: api login :param str token: api token :rtype: :py:class:`.client_v2.client.CoprClient` """ nc = NetClient(login, token) client = cls(nc, root_url, no_config=True) client.post_init() return client
[docs] @classmethod def create_from_file_config(cls, filepath=None, ignore_error=False): """ Creates Copr client using the information from the config file. :param filepath: specifies config location, default: "~/.config/copr" :type filepath: `str` :param bool ignore_error: When true and config is missing, creates default Client without credentionals :rtype: :py:class:`.client_v2.client.CoprClient` """ raw_config = configparser.ConfigParser() if not filepath: filepath = os.path.join(os.path.expanduser("~"), ".config", "copr") config = {} if not raw_config.read(filepath): log.warning( "No configuration file '~/.config/copr' found. " "See man copr-cli for more information") config["no_config"] = True if not ignore_error: raise CoprNoConfException() else: return cls.create_from_params() else: try: for field in ["login", "token", "copr_url"]: if six.PY3: config[field] = raw_config["copr-cli"].get(field, None) else: config[field] = raw_config.get("copr-cli", field, None) nc = NetClient(config["login"], config["token"]) client = cls(nc, root_url=config["copr_url"], ) except configparser.Error as err: if not ignore_error: raise CoprConfigException( "Bad configuration file: {0}".format(err)) else: return cls.create_from_params() client.post_init() return client
[docs] def post_init(self): """ Finalizes client initialization by querying API root info """ log.debug("Getting root resources") response = self.nc.request(self.api_root) # obtain api info info self.root = Root.from_response(response, self.root_url) # instantiating handlers self._projects = ProjectHandle( weakref.proxy(self), self.nc, root_url=self.root_url, projects_href=self.root.get_href_by_name(u"projects"),) self._project_chroots = ProjectChrootHandle( weakref.proxy(self), self.nc, root_url=self.root_url) self._builds = BuildHandle( weakref.proxy(self), self.nc, root_url=self.root_url, builds_href=self.root.get_href_by_name(u"builds"),) self._build_tasks = BuildTaskHandle( weakref.proxy(self), self.nc, root_url=self.root_url, build_tasks_href=self.root.get_href_by_name("build_tasks") ) self._mock_chroots = MockChrootHandle( weakref.proxy(self), self.nc, root_url=self.root_url, href=self.root.get_href_by_name(u"mock_chroots") ) self._post_init_done = True