# -*- coding: utf-8 -*-

#

# Copyright (C) 2013-2017 Gauvain Pocentek <gauvain@pocentek.net>

#

# 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 <http://www.gnu.org/licenses/>.

import gitlab

from gitlab import base

from gitlab import cli

from gitlab import exceptions as exc

from gitlab import types as g_types

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 = id.replace('/', '%2F')

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)

@exc.on_http_error(exc.GitlabListError)

def count(self, **kwargs):

"""Retrieve the number of available objects.

Args:

**kwargs: Extra options to send to the server (e.g. sudo)

Returns:

int: The number of objects

Raises:

GitlabAuthenticationError: If authentication is not correct

GitlabCountError: If the server cannot perform the request

"""

# Allow to overwrite the path, handy for custom calls

path = kwargs.pop('path', self.path)

headers = self.gitlab.http_head(path, **kwargs)

return int(headers['x-total'])

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, key.replace('/', '%2F'))

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 = id.replace('/', '%2F')

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)