# pylint: disable=R0903
# coding: utf-8
import six
if six.PY2:
from collections import Iterable
else:
from collections.abc import Iterable
from six import with_metaclass
from ..util import UnicodeMixin
from .entities import Link, ProjectEntity, ProjectChrootEntity, BuildEntity, MockChrootEntity, BuildTaskEntity
from .schemas import EmptySchema, BuildSchema, ProjectSchema, ProjectChrootSchema, MockChrootSchema, BuildTaskSchema
try:
_ = ProjectSchema(strict=True)
kwargs = {"strict": True}
except TypeError:
kwargs = {}
class EntityFieldDescriptor(object):
"""
Entity Field Descriptor
"""
def __init__(self, name):
self.name = name
def __get__(self, obj, objtype):
"""
:type obj: IndividualResource
"""
return getattr(obj._entity, self.name)
def __set__(self, obj, val):
"""
:type obj: IndividualResource
"""
setattr(obj._entity, self.name, val)
class ReadOnlyFieldDescriptor(object):
"""
Entity read-only Field Descriptor
"""
def __init__(self, name):
self.name = name
def __get__(self, obj, objtype):
"""
:type obj: IndividualResource
"""
return getattr(obj._entity, self.name)
class EntityFieldsMetaClass(type):
"""
Magic: we take fields info from class._schema and attach EntityFieldDescriptor
to Resource classes
"""
def __new__(mcs, class_name, bases, class_attrs):
schema = class_attrs.get("_schema")
if schema:
for f_name, f in schema.fields.items():
class_attrs[f_name] = EntityFieldDescriptor(f_name)
entity_methods = []
for base in bases:
entity_methods.extend(getattr(base, "_entity_methods", []))
entity_methods.extend(class_attrs.get("_entity_methods", []))
for m_name in entity_methods:
class_attrs[m_name] = ReadOnlyFieldDescriptor(m_name)
return type.__new__(mcs, class_name, bases, class_attrs)
# pylint: disable=E1101
class IndividualResource(with_metaclass(EntityFieldsMetaClass, UnicodeMixin)):
"""
:type handle: client_v2.handlers.AbstractHandle or None
:type response: ResponseWrapper or None
:type links: (dict of (str, Link)) or None
"""
_schema = EmptySchema(**kwargs)
# todo: this methods only serialize fields which can be modified by the user
# think about an approach to override `load_only=True` fields in
# our schemas during the dump function
# _entity_methods = ["to_json", "to_dict"]
def __init__(self, entity, handle=None, response=None, links=None, embedded=None, options=None):
self._entity = entity
self._handle = handle
self._response = response
self._links = links
self._embedded = embedded or dict()
self._options = options or dict()
def __dir__(self):
res = list(set(
dir(self.__class__) + list(self.__dict__.keys())
))
if self._entity:
res.extend(self._schema.fields.keys())
return res
def __unicode__(self):
return self._entity.__unicode__()
def get_href_by_name(self, name):
"""
:type name: str
"""
return self._links[name].href
class Root(IndividualResource):
def __init__(self, response, links, root_url):
super(Root, self).__init__(entity=None, response=response, links=links)
self.root_url = root_url
def get_resource_base_url(self, resource_name):
"""
:param str resource_name:
"""
return "{0}{1}".format(self.root_url, self.get_href_by_name(resource_name))
@classmethod
def from_response(cls, response, root_url):
"""
:type response: ResponseWrapper
:type root_url: unicode
"""
data_dict = response.json
links = Link.from_dict(data_dict["_links"])
return Root(response=response, links=links, root_url=root_url)
[docs]class Build(IndividualResource):
"""
:type entity: BuildEntity
:type handle: copr.client_v2.handlers.BuildHandle
"""
_schema = BuildSchema(**kwargs)
_entity_methods = ["is_finished"]
def __init__(self, entity, handle, **kwargs):
super(Build, self).__init__(entity=entity, handle=handle, **kwargs)
self._entity = entity
self._handle = handle
@classmethod
def from_response(cls, handle, data_dict, response=None, options=None):
links = Link.from_dict(data_dict["_links"])
entity = BuildEntity.from_dict(data_dict["build"])
return cls(entity=entity, handle=handle,
response=response, links=links, options=options)
[docs] def get_self(self):
""" Retrieves fresh build object from the service
:rtype: :py:class:`~.Build`
"""
return self._handle.get_one(self.id)
[docs] def cancel(self):
""" Updates the current build
:rtype: :py:class:`.OperationResult`
"""
return self._handle.cancel(self._entity)
[docs] def delete(self):
""" Deletes the current build
:rtype: :py:class:`.OperationResult`
"""
return self._handle.delete(self.id)
[docs] def get_build_tasks(self, **query_options):
""" Get build tasks owned by this build
:param query_options: see :py:meth:`.handlers.BuildHandle.get_list`
:rtype: :py:class:`~.BuildTasksList`
"""
handle = self._handle.get_build_tasks_handle()
return handle.get_list(build_id=self.id, **query_options)
[docs]class BuildTask(IndividualResource):
"""
:type entity: BuildTaskEntity
:type handle: copr.client_v2.handlers.BuildTaskHandle
"""
_schema = BuildTaskSchema(**kwargs)
def __init__(self, entity, handle, **kwargs):
super(BuildTask, self).__init__(entity=entity, handle=handle, **kwargs)
self._entity = entity
self._handle = handle
[docs] def get_self(self):
""" Retrieves fresh build task object from the service
:rtype: :py:class:`~.Build`
"""
return self._handle.get_one(self.build_id, self.chroot_name)
@classmethod
def from_response(cls, handle, data_dict, response=None, options=None):
links = Link.from_dict(data_dict["_links"])
entity = BuildTaskEntity.from_dict(data_dict["build_task"])
return cls(entity=entity, handle=handle,
response=response, links=links, options=options)
[docs]class Project(IndividualResource):
"""
:type entity: ProjectEntity
:type handle: copr.client_v2.handlers.ProjectHandle
"""
_schema = ProjectSchema(**kwargs)
def __init__(self, entity, handle, **kwargs):
super(Project, self).__init__(entity=entity, handle=handle, **kwargs)
self._entity = entity
self._handle = handle
[docs] def update(self):
""" Updates project using the current state.
Shortcut for for :py:meth:`.ProjectHandle.update`
:rtype: :py:class:`.OperationResult`
"""
return self._handle.update(self._entity)
[docs] def delete(self):
""" Updates project using the current state
:rtype: :py:class:`.OperationResult`
"""
return self._handle.delete(self.id)
[docs] def get_self(self):
""" Retrieves fresh project object from the service
:rtype: :py:class:`.Project`
"""
return self._handle.get_one(self.id)
[docs] def get_builds(self, **query_options):
""" Get builds owned by this project
:param query_options: see :py:meth:`.handlers.BuildHandle.get_list`
:rtype: :py:class:`~.BuildsList`
"""
handle = self._handle.get_builds_handle()
return handle.get_list(project_id=self.id, **query_options)
[docs] def get_build_tasks(self, **query_options):
""" Get build tasks owned by this project
:param query_options: see :py:meth:`.handlers.BuildHandle.get_list`
:rtype: :py:class:`~.BuildTasksList`
"""
handle = self._handle.get_build_tasks_handle()
return handle.get_list(project_id=self.id, **query_options)
[docs] def get_project_chroot(self, name):
""" Retrieves project chroot object by the given name
:param str name: mock chroot name
:rtype: :py:class:`.ProjectChroot`
"""
handle = self._handle.get_project_chroots_handle()
return handle.get_one(self, name)
[docs] def get_project_chroot_list(self):
""" Retrieves project chroots list
:rtype: :py:class:`.ProjectChrootList`
"""
handle = self._handle.get_project_chroots_handle()
return handle.get_list(self)
[docs] def enable_project_chroot(self, name):
"""
Enables given chroot for this project
Shortcut for for :py:meth:`.ProjectChrootHandle.enable`
:param str name: mock chroot name
:rtype: :py:class:`.OperationResult`
"""
handle = self._handle.get_project_chroots_handle()
return handle.enable(self, name)
[docs] def create_build_from_file(self, *args, **kwargs):
"""
Shortcut for :py:meth:`.BuildHandle.create_from_file`
(here you don't need to specify project_id)
"""
builds = self._handle.get_builds_handle()
return builds.create_from_file(self.id, *args, **kwargs)
[docs] def create_build_from_url(self, *args, **kwargs):
"""
Shortcut for :py:meth:`.BuildHandle.create_from_file`
(here you don't need to specify project_id)
"""
builds = self._handle.get_builds_handle()
return builds.create_from_url(self.id, *args, **kwargs)
@classmethod
def from_response(cls, handle, data_dict, response=None, options=None):
links = Link.from_dict(data_dict["_links"])
entity = ProjectEntity.from_dict(data_dict["project"])
return cls(entity=entity, handle=handle,
response=response, links=links, options=options)
[docs]class ProjectChroot(IndividualResource):
"""
:type entity: copr.client_v2.entities.ProjectChrootEntity
:type handle: copr.client_v2.handlers.ProjectChrootHandle
"""
_schema = ProjectChrootSchema(**kwargs)
def __init__(self, entity, handle, project, **kwargs):
super(ProjectChroot, self).__init__(entity=entity, handle=handle, **kwargs)
self._entity = entity
self._handle = handle
self._project = project
@classmethod
def from_response(cls, handle, data_dict, project, response=None, options=None):
links = Link.from_dict(data_dict["_links"])
entity = ProjectChrootEntity.from_dict(data_dict["chroot"])
return cls(entity=entity, handle=handle, project=project,
response=response, links=links, options=options)
[docs] def disable(self):
""" Disables chroot for the bound project
:rtype: :py:class:`.OperationResult`
"""
return self._handle.disable(self._project, self.name)
[docs] def update(self):
""" Updates chroot with the current entity state
:rtype: :py:class:`.OperationResult`
"""
return self._handle.update(self._project, self._entity)
[docs]class MockChroot(IndividualResource):
"""
:type entity: copr.client_v2.entities.MockChrootEntity
:type handle: copr.client_v2.handlers.MockChrootHandle
"""
_schema = MockChrootSchema(**kwargs)
def __init__(self, entity, handle, **kwargs):
super(MockChroot, self).__init__(
entity=entity,
handle=handle,
**kwargs
)
@classmethod
def from_response(cls, handle, data_dict, response=None, options=None):
links = Link.from_dict(data_dict["_links"])
entity = MockChrootEntity.from_dict(data_dict["chroot"])
return cls(entity=entity, handle=handle,
response=response, links=links, options=options)
[docs]class OperationResult(IndividualResource):
""" Fake resource to represent results of the requested operation
"""
def __init__(self, handle, response=None, entity=None, options=None, expected_status=200):
super(OperationResult, self).__init__(handle=handle, response=response, entity=entity, options=options)
self._expected_status = expected_status
@property
def new_location(self):
""" Contains an url to the new location produced by an operation
If operation doesn't produce a new location would contain None
:rtype: str
"""
# todo: Create sub-class for results which contains `new_location`
if self._response and \
self._response.headers and \
"location" in self._response.headers:
return self._response.headers["location"]
return None
[docs] def is_successful(self):
""" Performs check if status code is equal to the expected value
of particular request.
:rtype: bool
"""
if self._response and self._response.status_code == self._expected_status:
return True
else:
return False
def __unicode__(self):
out = u"<Result: "
if self._response:
out += u" status: {0}".format(self._response.status_code)
out += u">"
return out
class CollectionResource(Iterable, UnicodeMixin):
"""
:type handle: client_v2.handlers.AbstractHandle or None
:type response: ResponseWrapper
:type links: (dict of (str, Link)) or None
"""
def __init__(self, handle=None, response=None, links=None, individuals=None, options=None):
self._handle = handle
self._response = response
self._links = links
self._options = options or dict()
self._individuals = individuals
def get_href_by_name(self, name):
"""
:type name: str
"""
return self._links[name].href
def next_page(self):
limit = self._options.get("limit", 100)
offset = self._options.get("offset", 0)
offset += limit
params = {}
params.update(self._options)
params["limit"] = limit
params["offset"] = offset
return self._handle.get_list(self, **params)
def __iter__(self):
"""
:rtype: Iterable[IndividualResource]
"""
return iter(self._individuals)
def __len__(self):
if self._individuals is None:
raise RuntimeError(u"Collection resource is missing self._individuals")
return len(self._individuals)
def __getitem__(self, item):
if self._individuals is None:
raise RuntimeError(u"Collection resource is missing self._individuals")
return self._individuals[item]
def __unicode__(self):
out = u"<{}: [".format(self.__class__.__name__)
out += u", ".join([str(x) for x in self])
out += u"]>"
return out
@classmethod
def from_response(cls, handle, response, options):
raise NotImplementedError
def _construct_collection(
resource_class, handle, response,
individuals, options=None, **kwargs):
""" Helper to avoid code repetition
:type resource_class: CollectionResource
:param handle: AbstractHandle
:param response: ResponseWrapper
:param options: dict with query options
:param individuals: individuals
:param kwargs: additional parameters for constructor
:rtype: CollectionResource
"""
return resource_class(
handle,
response=response,
links=Link.from_dict(response.json["_links"]),
individuals=individuals,
options=options,
**kwargs
)
[docs]class ProjectList(CollectionResource):
"""
:type handle: copr.client_v2.handlers.ProjectHandle
"""
def __init__(self, handle, **kwargs):
super(ProjectList, self).__init__(**kwargs)
self._handle = handle
[docs] def next_page(self):
"""
Retrieves next chunk of the Project list for the same query options
:rtype: :py:class:`.ProjectList`
"""
return super(ProjectList, self).next_page()
@property
def projects(self):
""" :rtype: list of :py:class:`~.resources.Project` """
return self._individuals
@classmethod
def from_response(cls, handle, response, options):
individuals = [
Project.from_response(
handle=handle,
data_dict=dict_part,
)
for dict_part in response.json["projects"]
]
return _construct_collection(
cls, handle, response=response,
individuals=individuals, options=options
)
[docs]class BuildList(CollectionResource):
"""
:type handle: copr.client_v2.handler.BuildHandle
"""
def __init__(self, handle, **kwargs):
super(BuildList, self).__init__(**kwargs)
self._handle = handle
[docs] def next_page(self):
"""
Retrieves next chunk of the Build list for the same query options
:rtype: :py:class:`.BuildList`
"""
return super(BuildList, self).next_page()
@property
def builds(self):
"""
:rtype: :py:class:`.BuildList`
"""
return self._individuals
@classmethod
def from_response(cls, handle, response, options):
individuals = [
Build.from_response(
handle=handle,
data_dict=dict_part,
)
for dict_part in response.json["builds"]
]
return _construct_collection(
cls, handle, response=response,
individuals=individuals, options=options
)
[docs]class ProjectChrootList(CollectionResource):
"""
List of the :class:`~.ProjectChroot` in the one Project.
:type handle: copr.client_v2.handlers.ProjectChrootHandle
"""
def __init__(self, handle, project, **kwargs):
super(ProjectChrootList, self).__init__(**kwargs)
self._handle = handle
self._project = project
@property
def chroots(self):
"""
:rtype: list of :py:class:`~.resources.ProjectChroot`
"""
return self._individuals
[docs] def enable(self, name):
"""
Enables mock chroot for the current project
:rtype: :py:class:`~.OperationResult`
"""
return self._handle.enable(self._project, name)
@classmethod
def from_response(cls, handle, response, project):
individuals = [
ProjectChroot.from_response(
handle=handle,
data_dict=dict_part,
project=project
)
for dict_part in response.json["chroots"]
]
return _construct_collection(
cls, handle, response=response,
individuals=individuals, project=project
)
[docs]class MockChrootList(CollectionResource):
"""
List of the mock chroots supported by the service
:type handle: copr.client_v2.handlers.MockChrootHandle
"""
def __init__(self, handle, **kwargs):
super(MockChrootList, self).__init__(**kwargs)
self._handle = handle
@property
def chroots(self):
"""
:rtype: list of :py:class:`~.resources.MockChroot`
"""
return self._individuals
@classmethod
def from_response(cls, handle, response, options):
individuals = [
MockChroot.from_response(
handle=handle,
data_dict=dict_part,
)
for dict_part in response.json["chroots"]
]
return _construct_collection(
cls, handle, response=response,
options=options, individuals=individuals
)
[docs]class BuildTaskList(CollectionResource):
"""
List of build tasks
:type handle: copr.client_v2.handlers.BuildTaskHandle
"""
def __init__(self, handle, **kwargs):
super(BuildTaskList, self).__init__(**kwargs)
self._handle = handle
@property
def build_tasks(self):
"""
:rtype: list of :py:class:`~.resources.BuildTask`
"""
return self._individuals
@classmethod
def from_response(cls, handle, response, options):
individuals = [
BuildTask.from_response(
handle=handle,
data_dict=dict_part
)
for dict_part in response.json["build_tasks"]
]
return _construct_collection(
cls, handle, response=response,
options=options, individuals=individuals
)