# -*- coding: utf-8 -*-
#
# Copyright (C) 2013-2017 Gauvain Pocentek
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with this program. If not, see .
import gitlab
from gitlab import base
from gitlab import cli
from gitlab import exceptions as exc
from gitlab import types as g_types
from gitlab import utils
class GetMixin(object):
@exc.on_http_error(exc.GitlabGetError)
def get(self, id, lazy=False, **kwargs):
"""Retrieve a single object.
Args:
id (int or str): ID of the object to retrieve
lazy (bool): If True, don't request the server, but create a
shallow object giving access to the managers. This is
useful if you want to avoid useless calls to the API.
**kwargs: Extra options to send to the server (e.g. sudo)
Returns:
object: The generated RESTObject.
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabGetError: If the server cannot perform the request
"""
if not isinstance(id, int):
id = utils.clean_str_id(id)
path = "%s/%s" % (self.path, id)
if lazy is True:
return self._obj_cls(self, {self._obj_cls._id_attr: id})
server_data = self.gitlab.http_get(path, **kwargs)
return self._obj_cls(self, server_data)
class GetWithoutIdMixin(object):
@exc.on_http_error(exc.GitlabGetError)
def get(self, id=None, **kwargs):
"""Retrieve a single object.
Args:
**kwargs: Extra options to send to the server (e.g. sudo)
Returns:
object: The generated RESTObject
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabGetError: If the server cannot perform the request
"""
server_data = self.gitlab.http_get(self.path, **kwargs)
if server_data is None:
return None
return self._obj_cls(self, server_data)
class RefreshMixin(object):
@exc.on_http_error(exc.GitlabGetError)
def refresh(self, **kwargs):
"""Refresh a single object from server.
Args:
**kwargs: Extra options to send to the server (e.g. sudo)
Returns None (updates the object)
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabGetError: If the server cannot perform the request
"""
if self._id_attr:
path = "%s/%s" % (self.manager.path, self.id)
else:
path = self.manager.path
server_data = self.manager.gitlab.http_get(path, **kwargs)
self._update_attrs(server_data)
class ListMixin(object):
@exc.on_http_error(exc.GitlabListError)
def list(self, **kwargs):
"""Retrieve a list of objects.
Args:
all (bool): If True, return all the items, without pagination
per_page (int): Number of items to retrieve per request
page (int): ID of the page to return (starts with page 1)
as_list (bool): If set to False and no pagination option is
defined, return a generator instead of a list
**kwargs: Extra options to send to the server (e.g. sudo)
Returns:
list: The list of objects, or a generator if `as_list` is False
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabListError: If the server cannot perform the request
"""
# Duplicate data to avoid messing with what the user sent us
data = kwargs.copy()
if self.gitlab.per_page:
data.setdefault("per_page", self.gitlab.per_page)
# We get the attributes that need some special transformation
types = getattr(self, "_types", {})
if types:
for attr_name, type_cls in types.items():
if attr_name in data.keys():
type_obj = type_cls(data[attr_name])
data[attr_name] = type_obj.get_for_api()
# Allow to overwrite the path, handy for custom listings
path = data.pop("path", self.path)
obj = self.gitlab.http_list(path, **data)
if isinstance(obj, list):
return [self._obj_cls(self, item) for item in obj]
else:
return base.RESTObjectList(self, self._obj_cls, obj)
class RetrieveMixin(ListMixin, GetMixin):
pass
class CreateMixin(object):
def _check_missing_create_attrs(self, data):
required, optional = self.get_create_attrs()
missing = []
for attr in required:
if attr not in data:
missing.append(attr)
continue
if missing:
raise AttributeError("Missing attributes: %s" % ", ".join(missing))
def get_create_attrs(self):
"""Return the required and optional arguments.
Returns:
tuple: 2 items: list of required arguments and list of optional
arguments for creation (in that order)
"""
return getattr(self, "_create_attrs", (tuple(), tuple()))
@exc.on_http_error(exc.GitlabCreateError)
def create(self, data, **kwargs):
"""Create a new object.
Args:
data (dict): parameters to send to the server to create the
resource
**kwargs: Extra options to send to the server (e.g. sudo)
Returns:
RESTObject: a new instance of the managed object class built with
the data sent by the server
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabCreateError: If the server cannot perform the request
"""
self._check_missing_create_attrs(data)
files = {}
# We get the attributes that need some special transformation
types = getattr(self, "_types", {})
if types:
# Duplicate data to avoid messing with what the user sent us
data = data.copy()
for attr_name, type_cls in types.items():
if attr_name in data.keys():
type_obj = type_cls(data[attr_name])
# if the type if FileAttribute we need to pass the data as
# file
if issubclass(type_cls, g_types.FileAttribute):
k = type_obj.get_file_name(attr_name)
files[attr_name] = (k, data.pop(attr_name))
else:
data[attr_name] = type_obj.get_for_api()
# Handle specific URL for creation
path = kwargs.pop("path", self.path)
server_data = self.gitlab.http_post(path, post_data=data, files=files, **kwargs)
return self._obj_cls(self, server_data)
class UpdateMixin(object):
def _check_missing_update_attrs(self, data):
required, optional = self.get_update_attrs()
missing = []
for attr in required:
if attr not in data:
missing.append(attr)
continue
if missing:
raise AttributeError("Missing attributes: %s" % ", ".join(missing))
def get_update_attrs(self):
"""Return the required and optional arguments.
Returns:
tuple: 2 items: list of required arguments and list of optional
arguments for update (in that order)
"""
return getattr(self, "_update_attrs", (tuple(), tuple()))
def _get_update_method(self):
"""Return the HTTP method to use.
Returns:
object: http_put (default) or http_post
"""
if getattr(self, "_update_uses_post", False):
http_method = self.gitlab.http_post
else:
http_method = self.gitlab.http_put
return http_method
@exc.on_http_error(exc.GitlabUpdateError)
def update(self, id=None, new_data={}, **kwargs):
"""Update an object on the server.
Args:
id: ID of the object to update (can be None if not required)
new_data: the update data for the object
**kwargs: Extra options to send to the server (e.g. sudo)
Returns:
dict: The new object data (*not* a RESTObject)
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabUpdateError: If the server cannot perform the request
"""
if id is None:
path = self.path
else:
path = "%s/%s" % (self.path, id)
self._check_missing_update_attrs(new_data)
files = {}
# We get the attributes that need some special transformation
types = getattr(self, "_types", {})
if types:
# Duplicate data to avoid messing with what the user sent us
new_data = new_data.copy()
for attr_name, type_cls in types.items():
if attr_name in new_data.keys():
type_obj = type_cls(new_data[attr_name])
# if the type if FileAttribute we need to pass the data as
# file
if issubclass(type_cls, g_types.FileAttribute):
k = type_obj.get_file_name(attr_name)
files[attr_name] = (k, new_data.pop(attr_name))
else:
new_data[attr_name] = type_obj.get_for_api()
http_method = self._get_update_method()
return http_method(path, post_data=new_data, files=files, **kwargs)
class SetMixin(object):
@exc.on_http_error(exc.GitlabSetError)
def set(self, key, value, **kwargs):
"""Create or update the object.
Args:
key (str): The key of the object to create/update
value (str): The value to set for the object
**kwargs: Extra options to send to the server (e.g. sudo)
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabSetError: If an error occured
Returns:
obj: The created/updated attribute
"""
path = "%s/%s" % (self.path, utils.clean_str_id(key))
data = {"value": value}
server_data = self.gitlab.http_put(path, post_data=data, **kwargs)
return self._obj_cls(self, server_data)
class DeleteMixin(object):
@exc.on_http_error(exc.GitlabDeleteError)
def delete(self, id, **kwargs):
"""Delete an object on the server.
Args:
id: ID of the object to delete
**kwargs: Extra options to send to the server (e.g. sudo)
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabDeleteError: If the server cannot perform the request
"""
if id is None:
path = self.path
else:
if not isinstance(id, int):
id = utils.clean_str_id(id)
path = "%s/%s" % (self.path, id)
self.gitlab.http_delete(path, **kwargs)
class CRUDMixin(GetMixin, ListMixin, CreateMixin, UpdateMixin, DeleteMixin):
pass
class NoUpdateMixin(GetMixin, ListMixin, CreateMixin, DeleteMixin):
pass
class SaveMixin(object):
"""Mixin for RESTObject's that can be updated."""
def _get_updated_data(self):
updated_data = {}
required, optional = self.manager.get_update_attrs()
for attr in required:
# Get everything required, no matter if it's been updated
updated_data[attr] = getattr(self, attr)
# Add the updated attributes
updated_data.update(self._updated_attrs)
return updated_data
def save(self, **kwargs):
"""Save the changes made to the object to the server.
The object is updated to match what the server returns.
Args:
**kwargs: Extra options to send to the server (e.g. sudo)
Raise:
GitlabAuthenticationError: If authentication is not correct
GitlabUpdateError: If the server cannot perform the request
"""
updated_data = self._get_updated_data()
# Nothing to update. Server fails if sent an empty dict.
if not updated_data:
return
# call the manager
obj_id = self.get_id()
server_data = self.manager.update(obj_id, updated_data, **kwargs)
if server_data is not None:
self._update_attrs(server_data)
class ObjectDeleteMixin(object):
"""Mixin for RESTObject's that can be deleted."""
def delete(self, **kwargs):
"""Delete the object from the server.
Args:
**kwargs: Extra options to send to the server (e.g. sudo)
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabDeleteError: If the server cannot perform the request
"""
self.manager.delete(self.get_id())
class UserAgentDetailMixin(object):
@cli.register_custom_action(("Snippet", "ProjectSnippet", "ProjectIssue"))
@exc.on_http_error(exc.GitlabGetError)
def user_agent_detail(self, **kwargs):
"""Get the user agent detail.
Args:
**kwargs: Extra options to send to the server (e.g. sudo)
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabGetError: If the server cannot perform the request
"""
path = "%s/%s/user_agent_detail" % (self.manager.path, self.get_id())
return self.manager.gitlab.http_get(path, **kwargs)
class AccessRequestMixin(object):
@cli.register_custom_action(
("ProjectAccessRequest", "GroupAccessRequest"), tuple(), ("access_level",)
)
@exc.on_http_error(exc.GitlabUpdateError)
def approve(self, access_level=gitlab.DEVELOPER_ACCESS, **kwargs):
"""Approve an access request.
Args:
access_level (int): The access level for the user
**kwargs: Extra options to send to the server (e.g. sudo)
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabUpdateError: If the server fails to perform the request
"""
path = "%s/%s/approve" % (self.manager.path, self.id)
data = {"access_level": access_level}
server_data = self.manager.gitlab.http_put(path, post_data=data, **kwargs)
self._update_attrs(server_data)
class SubscribableMixin(object):
@cli.register_custom_action(("ProjectIssue", "ProjectMergeRequest", "ProjectLabel"))
@exc.on_http_error(exc.GitlabSubscribeError)
def subscribe(self, **kwargs):
"""Subscribe to the object notifications.
Args:
**kwargs: Extra options to send to the server (e.g. sudo)
raises:
GitlabAuthenticationError: If authentication is not correct
GitlabSubscribeError: If the subscription cannot be done
"""
path = "%s/%s/subscribe" % (self.manager.path, self.get_id())
server_data = self.manager.gitlab.http_post(path, **kwargs)
self._update_attrs(server_data)
@cli.register_custom_action(("ProjectIssue", "ProjectMergeRequest", "ProjectLabel"))
@exc.on_http_error(exc.GitlabUnsubscribeError)
def unsubscribe(self, **kwargs):
"""Unsubscribe from the object notifications.
Args:
**kwargs: Extra options to send to the server (e.g. sudo)
raises:
GitlabAuthenticationError: If authentication is not correct
GitlabUnsubscribeError: If the unsubscription cannot be done
"""
path = "%s/%s/unsubscribe" % (self.manager.path, self.get_id())
server_data = self.manager.gitlab.http_post(path, **kwargs)
self._update_attrs(server_data)
class TodoMixin(object):
@cli.register_custom_action(("ProjectIssue", "ProjectMergeRequest"))
@exc.on_http_error(exc.GitlabTodoError)
def todo(self, **kwargs):
"""Create a todo associated to the object.
Args:
**kwargs: Extra options to send to the server (e.g. sudo)
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabTodoError: If the todo cannot be set
"""
path = "%s/%s/todo" % (self.manager.path, self.get_id())
self.manager.gitlab.http_post(path, **kwargs)
class TimeTrackingMixin(object):
@cli.register_custom_action(("ProjectIssue", "ProjectMergeRequest"))
@exc.on_http_error(exc.GitlabTimeTrackingError)
def time_stats(self, **kwargs):
"""Get time stats for the object.
Args:
**kwargs: Extra options to send to the server (e.g. sudo)
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabTimeTrackingError: If the time tracking update cannot be done
"""
# Use the existing time_stats attribute if it exist, otherwise make an
# API call
if "time_stats" in self.attributes:
return self.attributes["time_stats"]
path = "%s/%s/time_stats" % (self.manager.path, self.get_id())
return self.manager.gitlab.http_get(path, **kwargs)
@cli.register_custom_action(("ProjectIssue", "ProjectMergeRequest"), ("duration",))
@exc.on_http_error(exc.GitlabTimeTrackingError)
def time_estimate(self, duration, **kwargs):
"""Set an estimated time of work for the object.
Args:
duration (str): Duration in human format (e.g. 3h30)
**kwargs: Extra options to send to the server (e.g. sudo)
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabTimeTrackingError: If the time tracking update cannot be done
"""
path = "%s/%s/time_estimate" % (self.manager.path, self.get_id())
data = {"duration": duration}
return self.manager.gitlab.http_post(path, post_data=data, **kwargs)
@cli.register_custom_action(("ProjectIssue", "ProjectMergeRequest"))
@exc.on_http_error(exc.GitlabTimeTrackingError)
def reset_time_estimate(self, **kwargs):
"""Resets estimated time for the object to 0 seconds.
Args:
**kwargs: Extra options to send to the server (e.g. sudo)
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabTimeTrackingError: If the time tracking update cannot be done
"""
path = "%s/%s/reset_time_estimate" % (self.manager.path, self.get_id())
return self.manager.gitlab.http_post(path, **kwargs)
@cli.register_custom_action(("ProjectIssue", "ProjectMergeRequest"), ("duration",))
@exc.on_http_error(exc.GitlabTimeTrackingError)
def add_spent_time(self, duration, **kwargs):
"""Add time spent working on the object.
Args:
duration (str): Duration in human format (e.g. 3h30)
**kwargs: Extra options to send to the server (e.g. sudo)
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabTimeTrackingError: If the time tracking update cannot be done
"""
path = "%s/%s/add_spent_time" % (self.manager.path, self.get_id())
data = {"duration": duration}
return self.manager.gitlab.http_post(path, post_data=data, **kwargs)
@cli.register_custom_action(("ProjectIssue", "ProjectMergeRequest"))
@exc.on_http_error(exc.GitlabTimeTrackingError)
def reset_spent_time(self, **kwargs):
"""Resets the time spent working on the object.
Args:
**kwargs: Extra options to send to the server (e.g. sudo)
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabTimeTrackingError: If the time tracking update cannot be done
"""
path = "%s/%s/reset_spent_time" % (self.manager.path, self.get_id())
return self.manager.gitlab.http_post(path, **kwargs)
class ParticipantsMixin(object):
@cli.register_custom_action(("ProjectMergeRequest", "ProjectIssue"))
@exc.on_http_error(exc.GitlabListError)
def participants(self, **kwargs):
"""List the participants.
Args:
all (bool): If True, return all the items, without pagination
per_page (int): Number of items to retrieve per request
page (int): ID of the page to return (starts with page 1)
as_list (bool): If set to False and no pagination option is
defined, return a generator instead of a list
**kwargs: Extra options to send to the server (e.g. sudo)
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabListError: If the list could not be retrieved
Returns:
RESTObjectList: The list of participants
"""
path = "%s/%s/participants" % (self.manager.path, self.get_id())
return self.manager.gitlab.http_get(path, **kwargs)
class BadgeRenderMixin(object):
@cli.register_custom_action(
("GroupBadgeManager", "ProjectBadgeManager"), ("link_url", "image_url")
)
@exc.on_http_error(exc.GitlabRenderError)
def render(self, link_url, image_url, **kwargs):
"""Preview link_url and image_url after interpolation.
Args:
link_url (str): URL of the badge link
image_url (str): URL of the badge image
**kwargs: Extra options to send to the server (e.g. sudo)
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabRenderError: If the rendering failed
Returns:
dict: The rendering properties
"""
path = "%s/render" % self.path
data = {"link_url": link_url, "image_url": image_url}
return self.gitlab.http_get(path, data, **kwargs)