# coding: utf-8
from abc import abstractmethod, ABCMeta
import json
import os
from copr.client_v2.net_client import RequestError, MultiPartTuple
from .entities import ProjectChrootEntity, ProjectCreateEntity
from .resources import Project, OperationResult, ProjectList, ProjectChroot, ProjectChrootList, Build, BuildList, \
MockChroot, MockChrootList, BuildTask, BuildTaskList
class AbstractHandle(object):
"""
:param client: Should be used only to access other handlers
:type client: copr.client_v2.client.HandlersProvider
:type nc: copr.client_v2.net_client.NetClient
"""
__metaclass__ = ABCMeta
def __init__(self, client, nc, root_url):
self.client = client
self.nc = nc
self.root_url = root_url
@abstractmethod
def get_base_url(self, *args, **kwargs):
pass
[docs]class BuildHandle(AbstractHandle):
def __init__(self, client, nc, root_url, builds_href):
super(BuildHandle, self).__init__(client, nc, root_url)
self.builds_href = builds_href
self._base_url = "{0}{1}".format(self.root_url, builds_href)
def get_base_url(self):
return self._base_url
[docs] def get_one(self, build_id):
""" Retrieves builds object
:param int build_id: id of the target build
:rtype: :py:class:`~.resources.Build`
"""
options = {"build_id": build_id}
url = "{0}/{1}".format(self.get_base_url(), build_id)
response = self.nc.request(url)
return Build.from_response(
handle=self,
response=response,
data_dict=response.json,
options=options,
)
[docs] def get_list(self, project_id=None, owner=None, limit=None, offset=None):
""" Retrieves builds object according to the given parameters
:param owner: name of the project owner
:param project_id: id of the project
:param limit: limit number of builds
:param offset: number of builds to skip
:rtype: :py:class:`~.resources.BuildList`
"""
options = {
"project_id": project_id,
"owner": owner,
"limit": limit,
"offset": offset
}
response = self.nc.request(self.get_base_url(), query_params=options)
return BuildList.from_response(self, response, options)
[docs] def cancel(self, build_entity):
""" Cancels the given build
:param build_entity: build entity to delete
:type build_entity: :py:class:`~.copr.client_v2.entities.BuildEntity`
:rtype: :py:class:`.OperationResult`
"""
build_id = build_entity.id
build_entity.state = "canceled"
url = "{0}/{1}".format(self.get_base_url(), build_id)
response = self.nc.request(url, data=build_entity.to_json(), method="PUT", do_auth=True)
return OperationResult(self, response)
[docs] def delete(self, build_id):
""" Deletes the given build
:param int build_id: build id to delete
:rtype: :py:class:`.OperationResult`
"""
url = "{0}/{1}".format(self.get_base_url(), build_id)
response = self.nc.request(url, method="delete", do_auth=True)
return OperationResult(self, response, expected_status=204)
def _process_create_response(self, request_data, response):
op_result = OperationResult(self, response, expected_status=201)
if op_result.is_successful():
build_response = self.nc.get(op_result.new_location)
return Build.from_response(
handle=self, response=build_response,
data_dict=build_response.json
)
else:
raise RequestError(
"Got unexpected status code at create build request",
url=self.get_base_url(),
request_body=request_data, response=response
)
[docs] def create_from_url(self, project_id, srpm_url,
chroots=None, enable_net=True):
"""
Creates new build using public url to the srpm file
:param int project_id: id of the project where we want to submit new build
:param str srpm_url: url to the source rpm
:param list chroots: which chroots should be used during the build
:param bool enable_net: allows to disable network access during the build, default: True
:return: created build
:rtype: :py:class:`~.resources.Build`
"""
chroots = list(map(str, chroots or list()))
content = {
"project_id": int(project_id),
"srpm_url": str(srpm_url),
"chroots": chroots,
"enable_net": bool(enable_net)
}
data = json.dumps(content)
response = self.nc.request(
self.get_base_url(),
data=data, method="POST", do_auth=True,
)
return self._process_create_response(data, response)
[docs] def create_from_file(self, project_id, file_path=None,
file_obj=None, file_name=None,
chroots=None, enable_net=True):
"""
Creates new build using srpm upload, please specify
either ``file_path`` or (``file_obj``, ``file_name`` ).
:param int project_id: id of the project where we want to submit new build
:param str file_path: path to the srpm file
:param file file_obj: file-like object to read from
:param str file_name: name for the uploaded file
:param list chroots: which chroots should be used during the build
:param bool enable_net: allows to disable network access during the build, default: True
:return: created build
:rtype: :py:class:`~.resources.Build`
"""
chroots = list(map(str, chroots or list()))
content = {
"project_id": int(project_id),
"chroots": chroots,
"enable_net": bool(enable_net)
}
metadata = MultiPartTuple(
"metadata", name=None,
obj=json.dumps(content), content_type="application/json")
if file_path is not None:
with open(file_path, "rb") as f_obj:
f_name = os.path.basename(f_obj.name)
parts, response = self._do_upload_request(f_name, f_obj, metadata)
elif file_obj is not None and file_name is not None:
parts, response = self._do_upload_request(file_name, file_obj, metadata)
else:
raise RuntimeError("Please provide file_path or file_obj and file_name")
return self._process_create_response(parts, response)
def _do_upload_request(self, f_name, f_obj, metadata):
srpm = MultiPartTuple("srpm", name=f_name, obj=f_obj,
content_type="appliction/x-rpm")
parts = [metadata, srpm]
response = self.nc.request_multipart(
url=self.get_base_url(),
method="POST",
data_parts=parts,
do_auth=True,
)
return parts, response
[docs] def get_build_tasks_handle(self):
"""
:rtype: BuildTasksHandle
"""
return self.client.build_tasks
[docs]class BuildTaskHandle(AbstractHandle):
def __init__(self, client, nc, root_url, build_tasks_href):
super(BuildTaskHandle, self).__init__(client, nc, root_url)
self.build_tasks_href = build_tasks_href
self._base_url = "{0}{1}".format(self.root_url, build_tasks_href)
def get_base_url(self):
return self._base_url
[docs] def get_list(self, owner=None, project_id=None, build_id=None,
state=None, offset=None, limit=None):
""" Retrieves build tasks list according to the given parameters
:param str owner: build tasks from the project owner by this user
:param int project_id: get tasks only from this project,
when used query parameter ``owner`` is ignored
:param int build_id: get tasks only from this build,
when used query parameters ``owner`` and ``project_id`` are ignored
:param str state: get build tasks only with this state, allowed values:
``failed``, ``succeeded``, ``canceled``, ``running``,
``pending``, ``starting``, ``importing``
:param int limit: limit number of projects
:param int offset: number of projects to skip
:rtype: :py:class:`~.resources.BuildTaskList`
"""
options = {
"owner": owner,
"project_id": project_id,
"build_id": build_id,
"state": state,
"limit": limit,
"offset": offset
}
response = self.nc.request(self.get_base_url(), query_params=options)
return BuildTaskList.from_response(self, response, options)
[docs] def get_one(self, build_id, chroot_name):
""" Retrieves single build task object
:param int build_id: id of the build
:param str chroot_name: name of the build chroot
:rtype: :py:class:`~.resources.BuildTask`
"""
url = "{0}/{1}/{2}".format(self.get_base_url(), build_id, chroot_name)
response = self.nc.request(url)
return BuildTask.from_response(
handle=self,
response=response,
data_dict=response.json,
)
[docs]class ProjectHandle(AbstractHandle):
def __init__(self, client, nc, root_url, projects_href):
super(ProjectHandle, self).__init__(client, nc, root_url)
self.projects_href = projects_href
self._base_url = "{0}{1}".format(self.root_url, projects_href)
def get_base_url(self):
return self._base_url
[docs] def get_list(self, search_query=None, owner=None, name=None, limit=None, offset=None):
""" Retrieves projects object according to the given parameters
:param str search_query: search projects with such string
:param str owner: owner username
:param str name: project name
:param int limit: limit number of projects
:param int offset: number of projects to skip
:rtype: :py:class:`~.resources.ProjectList`
"""
options = {
"search_query": search_query,
"owner": owner2user(owner),
"group": owner2group(owner),
"name": name,
"limit": limit,
"offset": offset
}
response = self.nc.request(self.get_base_url(), query_params=options)
return ProjectList.from_response(self, response, options)
[docs] def get_one(self, project_id):
# todo: implement: , show_builds=False, show_chroots=False):
""" Retrieves project object.
:param int project_id: project identifier
:rtype: :py:class:`~.resources.Project`
"""
query_params = {
# "show_builds": show_builds,
# "show_chroots": show_chroots
}
url = "{0}/{1}".format(self.get_base_url(), project_id)
response = self.nc.request(url, query_params=query_params)
return Project.from_response(
handle=self,
response=response,
data_dict=response.json,
options=query_params
)
[docs] def create(
self, name, owner, chroots, description=None, instructions=None,
homepage=None, contact=None, disable_createrepo=None, build_enable_net=None,
repos=None,
):
""" Creates new project
:param name: project name
:param owner: username
:param chroots: list of mock chroot to be used in project
:param description:
:param instructions:
:param homepage:
:param contact:
:param bool disable_createrepo:
:param bool build_enable_net:
:param repos: list of additional repos enabled for builds
:rtype: :py:class:`~.resources.Project`
"""
new_entity = ProjectCreateEntity(
owner=owner,
group=owner2group(owner),
name=name,
chroots=chroots,
description=description,
instructions=instructions,
homepage=homepage,
contact=contact,
disable_createrepo=disable_createrepo,
build_enable_net=build_enable_net,
repos=repos
)
url = self.get_base_url()
request_data = new_entity.to_json()
response = self.nc.request(url, method="post", data=request_data, do_auth=True)
op_result = OperationResult(self, response, expected_status=201)
if op_result.is_successful():
response = self.nc.get(op_result.new_location)
return Project.from_response(
handle=self, response=response, data_dict=response.json)
else:
raise RequestError(
"Got unexpected status code at create build request",
url=self.get_base_url(),
request_body=request_data, response=response
)
[docs] def update(self, project_entity):
""" Updates project.
:param project_entity: project entity to use for update
:type project_entity: :py:class:`~.ProjectEntity`
:rtype: OperationResult
"""
url = "{0}/{1}".format(self.get_base_url(), project_entity.id)
data = project_entity.to_json()
response = self.nc.request(url, method="put", data=data, do_auth=True)
return OperationResult(self, response)
[docs] def delete(self, project_id):
""" Deletes project.
:param int project_id: project identifier
:rtype: OperationResult
"""
url = "{0}/{1}".format(self.get_base_url(), project_id)
response = self.nc.request(url, method="delete", do_auth=True)
return OperationResult(self, response, expected_status=204)
[docs] def get_builds_handle(self):
"""
:rtype: BuildHandle
"""
return self.client.builds
[docs] def get_build_tasks_handle(self):
"""
:rtype: BuildTasksHandle
"""
return self.client.build_tasks
[docs] def get_project_chroots_handle(self):
"""
:rtype: ProjectChrootHandle
"""
return self.client.project_chroots
[docs]class ProjectChrootHandle(AbstractHandle):
[docs] def get_base_url(self, project, **kwargs):
"""
:type project: copr.client_v2.resources.Project
"""
return "{0}{1}".format(self.root_url, project.get_href_by_name("chroots"))
[docs] def get_one(self, project, name):
""" Retrieves project chroot object.
:type project: :py:class:`~copr.client_v2.resources.Project`
:param project: parent project for the chroot
:param str name: chroot name
:rtype: :py:class:`~copr.client_v2.resources.ProjectChroot`
"""
url = "{0}/{1}".format(self.get_base_url(project), name)
response = self.nc.request(url)
return ProjectChroot.from_response(
handle=self,
response=response,
data_dict=response.json,
project=project,
)
[docs] def get_list(self, project):
""" Retrieves project chroot list object.
:type project: :py:class:`~copr.client_v2.resources.Project`
:param project: parent project for the chroot
:rtype: :py:class:`~copr.client_v2.resources.ProjectChrootList`
"""
response = self.nc.request(self.get_base_url(project))
return ProjectChrootList.from_response(
handle=self,
response=response,
project=project
)
[docs] def disable(self, project, name):
""" Disables one chroot for the project
:type project: :py:class:`~copr.client_v2.resources.Project`
:param project: parent project for the chroot
:param str name: chroot name to disable
"""
url = "{0}/{1}".format(self.get_base_url(project), name)
response = self.nc.request(url, method="DELETE", do_auth=True)
return OperationResult(self, response)
[docs] def enable(self, project, name, buildroot_pkgs=None):
""" Enables one chroot for the project
:type project: :py:class:`~copr.client_v2.resources.Project`
:param project: parent project for the chroot
:param str name: chroot name to enable
:params buildroot_pkgs: packages to add into the buildroot
:type buildroot_pkgs: list of str
:rtype: :py:class:`.OperationResult`
"""
new_entity = ProjectChrootEntity(
name=name,
buildroot_pkgs=buildroot_pkgs or list()
)
response = self.nc.request(
self.get_base_url(project),
method="POST",
data=new_entity.to_json(),
do_auth=True
)
return OperationResult(self, response)
[docs] def update(self, project, chroot_entity):
"""
:type project: copr.client_v2.resources.Project
:param chroot_entity: Entity to update
:type chroot_entity: :py:class:`.entities.ProjectChrootEntity`
:rtype: :py:class:`.OperationResult`
"""
url = "{0}/{1}".format(self.get_base_url(project), chroot_entity.name)
response = self.nc.request(
url,
method="PUT",
data=chroot_entity.to_json(),
do_auth=True
)
return OperationResult(self, response)
[docs]class MockChrootHandle(AbstractHandle):
def __init__(self, client, nc, root_url, href):
super(MockChrootHandle, self).__init__(client, nc, root_url)
self._href = href
self._base_url = "{0}{1}".format(self.root_url, href)
def get_base_url(self):
return self._base_url
[docs] def get_one(self, name):
""" Retrieves mock chroot object.
:param str name: chroot name
:rtype: :py:class:`~copr.client_v2.resources.MockChroot`
"""
url = "{0}/{1}".format(self.get_base_url(), name)
response = self.nc.get(url)
return MockChroot.from_response(
handle=self,
response=response,
data_dict=response.json,
)
[docs] def get_list(self, active_only=True):
""" Retrieves mock chroot list object.
:param bool active_only: when True, shows only chroots which can be used for builds
:rtype: :py:class:`~copr.client_v2.resources.MockChrootList`
"""
options = dict(active_only=active_only)
response = self.nc.get(
self.get_base_url(),
query_params=options
)
return MockChrootList.from_response(
handle=self,
response=response,
options=options
)
def owner2user(owner):
return owner and (owner if owner[0] != "@" else None)
def owner2group(owner):
return owner and (owner[1:] if owner[0] == "@" else None)