# -*- 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/>.

from __future__ import print_function

from __future__ import absolute_import

import base64

from gitlab.base import * # noqa

from gitlab import cli

from gitlab.exceptions import * # noqa

from gitlab.mixins import * # noqa

from gitlab import utils

VISIBILITY_PRIVATE = 'private'

VISIBILITY_INTERNAL = 'internal'

VISIBILITY_PUBLIC = 'public'

ACCESS_GUEST = 10

ACCESS_REPORTER = 20

ACCESS_DEVELOPER = 30

ACCESS_MASTER = 40

ACCESS_OWNER = 50

class SidekiqManager(RESTManager):

"""Manager for the Sidekiq methods.

This manager doesn't actually manage objects but provides helper fonction

for the sidekiq metrics API.

"""

@cli.register_custom_action('SidekiqManager')

@exc.on_http_error(exc.GitlabGetError)

def queue_metrics(self, **kwargs):

"""Return the registred queues information.

Args:

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

Raises:

GitlabAuthenticationError: If authentication is not correct

GitlabGetError: If the information couldn't be retrieved

Returns:

dict: Information about the Sidekiq queues

"""

return self.gitlab.http_get('/sidekiq/queue_metrics', **kwargs)

@cli.register_custom_action('SidekiqManager')

@exc.on_http_error(exc.GitlabGetError)

def process_metrics(self, **kwargs):

"""Return the registred sidekiq workers.

Args:

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

Raises:

GitlabAuthenticationError: If authentication is not correct

GitlabGetError: If the information couldn't be retrieved

Returns:

dict: Information about the register Sidekiq worker

"""

return self.gitlab.http_get('/sidekiq/process_metrics', **kwargs)

@cli.register_custom_action('SidekiqManager')

@exc.on_http_error(exc.GitlabGetError)

def job_stats(self, **kwargs):

"""Return statistics about the jobs performed.

Args:

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

Raises:

GitlabAuthenticationError: If authentication is not correct

GitlabGetError: If the information couldn't be retrieved

Returns:

dict: Statistics about the Sidekiq jobs performed

"""

return self.gitlab.http_get('/sidekiq/job_stats', **kwargs)

@cli.register_custom_action('SidekiqManager')

@exc.on_http_error(exc.GitlabGetError)

def compound_metrics(self, **kwargs):

"""Return all available metrics and statistics.

Args:

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

Raises:

GitlabAuthenticationError: If authentication is not correct

GitlabGetError: If the information couldn't be retrieved

Returns:

dict: All available Sidekiq metrics and statistics

"""

return self.gitlab.http_get('/sidekiq/compound_metrics', **kwargs)

class Event(RESTObject):

_id_attr = None

_short_print_attr = 'target_title'

class EventManager(ListMixin, RESTManager):

_path = '/events'

_obj_cls = Event

_list_filters = ('action', 'target_type', 'before', 'after', 'sort')

class UserActivities(RESTObject):

_id_attr = 'username'

class UserActivitiesManager(ListMixin, RESTManager):

_path = '/user/activities'

_obj_cls = UserActivities

class UserCustomAttribute(ObjectDeleteMixin, RESTObject):

_id_attr = 'key'

class UserCustomAttributeManager(RetrieveMixin, SetMixin, DeleteMixin,

RESTManager):

_path = '/users/%(user_id)s/custom_attributes'

_obj_cls = UserCustomAttribute

_from_parent_attrs = {'user_id': 'id'}

class UserEmail(ObjectDeleteMixin, RESTObject):

_short_print_attr = 'email'

class UserEmailManager(RetrieveMixin, CreateMixin, DeleteMixin, RESTManager):

_path = '/users/%(user_id)s/emails'

_obj_cls = UserEmail

_from_parent_attrs = {'user_id': 'id'}

_create_attrs = (('email', ), tuple())

class UserEvent(Event):

pass

class UserEventManager(EventManager):

_path = '/users/%(user_id)s/events'

_obj_cls = UserEvent

_from_parent_attrs = {'user_id': 'id'}

class UserGPGKey(ObjectDeleteMixin, RESTObject):

pass

class UserGPGKeyManager(RetrieveMixin, CreateMixin, DeleteMixin, RESTManager):

_path = '/users/%(user_id)s/gpg_keys'

_obj_cls = UserGPGKey

_from_parent_attrs = {'user_id': 'id'}

_create_attrs = (('key',), tuple())

class UserKey(ObjectDeleteMixin, RESTObject):

pass

class UserKeyManager(GetFromListMixin, CreateMixin, DeleteMixin, RESTManager):

_path = '/users/%(user_id)s/keys'

_obj_cls = UserKey

_from_parent_attrs = {'user_id': 'id'}

_create_attrs = (('title', 'key'), tuple())

class UserImpersonationToken(ObjectDeleteMixin, RESTObject):

pass

class UserImpersonationTokenManager(NoUpdateMixin, RESTManager):

_path = '/users/%(user_id)s/impersonation_tokens'

_obj_cls = UserImpersonationToken

_from_parent_attrs = {'user_id': 'id'}

_create_attrs = (('name', 'scopes'), ('expires_at',))

_list_filters = ('state',)

class UserProject(RESTObject):

pass

class UserProjectManager(ListMixin, CreateMixin, RESTManager):

_path = '/projects/user/%(user_id)s'

_obj_cls = UserProject

_from_parent_attrs = {'user_id': 'id'}

_create_attrs = (

('name', ),

('default_branch', 'issues_enabled', 'wall_enabled',

'merge_requests_enabled', 'wiki_enabled', 'snippets_enabled',

'public', 'visibility', 'description', 'builds_enabled',

'public_builds', 'import_url', 'only_allow_merge_if_build_succeeds')

)

_list_filters = ('archived', 'visibility', 'order_by', 'sort', 'search',

'simple', 'owned', 'membership', 'starred', 'statistics',

'with_issues_enabled', 'with_merge_requests_enabled')

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 Gitlab 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

"""

path = '/users/%s/projects' % self._parent.id

return ListMixin.list(self, path=path, **kwargs)

class User(SaveMixin, ObjectDeleteMixin, RESTObject):

_short_print_attr = 'username'

_managers = (

('customattributes', 'UserCustomAttributeManager'),

('emails', 'UserEmailManager'),

('events', 'UserEventManager'),

('gpgkeys', 'UserGPGKeyManager'),

('impersonationtokens', 'UserImpersonationTokenManager'),

('keys', 'UserKeyManager'),

('projects', 'UserProjectManager'),

)

@cli.register_custom_action('User')

@exc.on_http_error(exc.GitlabBlockError)

def block(self, **kwargs):

"""Block the user.

Args:

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

Raises:

GitlabAuthenticationError: If authentication is not correct

GitlabBlockError: If the user could not be blocked

Returns:

bool: Whether the user status has been changed

"""

path = '/users/%s/block' % self.id

server_data = self.manager.gitlab.http_post(path, **kwargs)

if server_data is True:

self._attrs['state'] = 'blocked'

return server_data

@cli.register_custom_action('User')

@exc.on_http_error(exc.GitlabUnblockError)

def unblock(self, **kwargs):

"""Unblock the user.

Args:

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

Raises:

GitlabAuthenticationError: If authentication is not correct

GitlabUnblockError: If the user could not be unblocked

Returns:

bool: Whether the user status has been changed

"""

path = '/users/%s/unblock' % self.id

server_data = self.manager.gitlab.http_post(path, **kwargs)

if server_data is True:

self._attrs['state'] = 'active'

return server_data

class UserManager(CRUDMixin, RESTManager):

_path = '/users'

_obj_cls = User

_list_filters = ('active', 'blocked', 'username', 'extern_uid', 'provider',

'external', 'search', 'custom_attributes')

_create_attrs = (

tuple(),

('email', 'username', 'name', 'password', 'reset_password', 'skype',

'linkedin', 'twitter', 'projects_limit', 'extern_uid', 'provider',

'bio', 'admin', 'can_create_group', 'website_url',

'skip_confirmation', 'external', 'organization', 'location')

)

_update_attrs = (

('email', 'username', 'name'),

('password', 'skype', 'linkedin', 'twitter', 'projects_limit',

'extern_uid', 'provider', 'bio', 'admin', 'can_create_group',

'website_url', 'skip_confirmation', 'external', 'organization',

'location')

)

def _sanitize_data(self, data, action):

new_data = data.copy()

if 'confirm' in data:

new_data['confirm'] = str(new_data['confirm']).lower()

return new_data

class CurrentUserEmail(ObjectDeleteMixin, RESTObject):

_short_print_attr = 'email'

class CurrentUserEmailManager(RetrieveMixin, CreateMixin, DeleteMixin,

RESTManager):

_path = '/user/emails'

_obj_cls = CurrentUserEmail

_create_attrs = (('email', ), tuple())

class CurrentUserGPGKey(ObjectDeleteMixin, RESTObject):

pass

class CurrentUserGPGKeyManager(RetrieveMixin, CreateMixin, DeleteMixin,

RESTManager):

_path = '/user/gpg_keys'

_obj_cls = CurrentUserGPGKey

_create_attrs = (('key',), tuple())

class CurrentUserKey(ObjectDeleteMixin, RESTObject):

_short_print_attr = 'title'

class CurrentUserKeyManager(RetrieveMixin, CreateMixin, DeleteMixin,

RESTManager):

_path = '/user/keys'

_obj_cls = CurrentUserKey

_create_attrs = (('title', 'key'), tuple())

class CurrentUser(RESTObject):

_id_attr = None

_short_print_attr = 'username'

_managers = (

('emails', 'CurrentUserEmailManager'),

('gpgkeys', 'CurrentUserGPGKeyManager'),

('keys', 'CurrentUserKeyManager'),

)

class CurrentUserManager(GetWithoutIdMixin, RESTManager):

_path = '/user'

_obj_cls = CurrentUser

class ApplicationSettings(SaveMixin, RESTObject):

_id_attr = None

class ApplicationSettingsManager(GetWithoutIdMixin, UpdateMixin, RESTManager):

_path = '/application/settings'

_obj_cls = ApplicationSettings

_update_attrs = (

tuple(),

('after_sign_out_path', 'container_registry_token_expire_delay',

'default_branch_protection', 'default_project_visibility',

'default_projects_limit', 'default_snippet_visibility',

'domain_blacklist', 'domain_blacklist_enabled', 'domain_whitelist',

'enabled_git_access_protocol', 'gravatar_enabled', 'home_page_url',

'max_attachment_size', 'repository_storage',

'restricted_signup_domains', 'restricted_visibility_levels',

'session_expire_delay', 'sign_in_text', 'signin_enabled',

'signup_enabled', 'twitter_sharing_enabled',

'user_oauth_applications')

)

def _sanitize_data(self, data, action):

new_data = data.copy()

if 'domain_whitelist' in data and data['domain_whitelist'] is None:

new_data.pop('domain_whitelist')

return new_data

class BroadcastMessage(SaveMixin, ObjectDeleteMixin, RESTObject):

pass

class BroadcastMessageManager(CRUDMixin, RESTManager):

_path = '/broadcast_messages'

_obj_cls = BroadcastMessage

_create_attrs = (('message', ), ('starts_at', 'ends_at', 'color', 'font'))

_update_attrs = (tuple(), ('message', 'starts_at', 'ends_at', 'color',

'font'))

class DeployKey(RESTObject):

pass

class DeployKeyManager(GetFromListMixin, RESTManager):

_path = '/deploy_keys'

_obj_cls = DeployKey

class NotificationSettings(SaveMixin, RESTObject):

_id_attr = None

class NotificationSettingsManager(GetWithoutIdMixin, UpdateMixin, RESTManager):

_path = '/notification_settings'

_obj_cls = NotificationSettings

_update_attrs = (

tuple(),

('level', 'notification_email', 'new_note', 'new_issue',

'reopen_issue', 'close_issue', 'reassign_issue', 'new_merge_request',

'reopen_merge_request', 'close_merge_request',

'reassign_merge_request', 'merge_merge_request')

)

class Dockerfile(RESTObject):

_id_attr = 'name'

class DockerfileManager(RetrieveMixin, RESTManager):

_path = '/templates/dockerfiles'

_obj_cls = Dockerfile

class Feature(RESTObject):

_id_attr = 'name'

class FeatureManager(ListMixin, RESTManager):

_path = '/features/'

_obj_cls = Feature

@exc.on_http_error(exc.GitlabSetError)

def set(self, name, value, feature_group=None, user=None, **kwargs):

"""Create or update the object.

Args:

name (str): The value to set for the object

value (bool/int): The value to set for the object

feature_group (str): A feature group name

user (str): A GitLab username

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

data = {'value': value, 'feature_group': feature_group, 'user': user}

server_data = self.gitlab.http_post(path, post_data=data, **kwargs)

return self._obj_cls(self, server_data)

class Gitignore(RESTObject):

_id_attr = 'name'

class GitignoreManager(RetrieveMixin, RESTManager):

_path = '/templates/gitignores'

_obj_cls = Gitignore

class Gitlabciyml(RESTObject):

_id_attr = 'name'

class GitlabciymlManager(RetrieveMixin, RESTManager):

_path = '/templates/gitlab_ci_ymls'

_obj_cls = Gitlabciyml

class GroupAccessRequest(AccessRequestMixin, ObjectDeleteMixin, RESTObject):

pass

class GroupAccessRequestManager(GetFromListMixin, CreateMixin, DeleteMixin,

RESTManager):

_path = '/groups/%(group_id)s/access_requests'

_obj_cls = GroupAccessRequest

_from_parent_attrs = {'group_id': 'id'}

class GroupCustomAttribute(ObjectDeleteMixin, RESTObject):

_id_attr = 'key'

class GroupCustomAttributeManager(RetrieveMixin, SetMixin, DeleteMixin,

RESTManager):

_path = '/groups/%(group_id)s/custom_attributes'

_obj_cls = GroupCustomAttribute

_from_parent_attrs = {'group_id': 'id'}

class GroupIssue(RESTObject):

pass

class GroupIssueManager(GetFromListMixin, RESTManager):

_path = '/groups/%(group_id)s/issues'

_obj_cls = GroupIssue

_from_parent_attrs = {'group_id': 'id'}

_list_filters = ('state', 'labels', 'milestone', 'order_by', 'sort')

class GroupMember(SaveMixin, ObjectDeleteMixin, RESTObject):

_short_print_attr = 'username'

class GroupMemberManager(GetFromListMixin, CreateMixin, UpdateMixin,

DeleteMixin, RESTManager):

_path = '/groups/%(group_id)s/members'

_obj_cls = GroupMember

_from_parent_attrs = {'group_id': 'id'}

_create_attrs = (('access_level', 'user_id'), ('expires_at', ))

_update_attrs = (('access_level', ), ('expires_at', ))

class GroupMergeRequest(RESTObject):

pass

class GroupMergeRequestManager(RESTManager):

pass

class GroupMilestone(SaveMixin, ObjectDeleteMixin, RESTObject):

_short_print_attr = 'title'

@cli.register_custom_action('GroupMilestone')

@exc.on_http_error(exc.GitlabListError)

def issues(self, **kwargs):

"""List issues related to this milestone.

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 issues

"""

path = '%s/%s/issues' % (self.manager.path, self.get_id())

data_list = self.manager.gitlab.http_list(path, as_list=False,

**kwargs)

manager = GroupIssueManager(self.manager.gitlab,

parent=self.manager._parent)

# FIXME(gpocentek): the computed manager path is not correct

return RESTObjectList(manager, GroupIssue, data_list)

@cli.register_custom_action('GroupMilestone')

@exc.on_http_error(exc.GitlabListError)

def merge_requests(self, **kwargs):

"""List the merge requests related to this milestone.

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 merge requests

"""

path = '%s/%s/merge_requests' % (self.manager.path, self.get_id())

data_list = self.manager.gitlab.http_list(path, as_list=False,

**kwargs)

manager = GroupIssueManager(self.manager.gitlab,

parent=self.manager._parent)

# FIXME(gpocentek): the computed manager path is not correct

return RESTObjectList(manager, GroupMergeRequest, data_list)

class GroupMilestoneManager(CRUDMixin, RESTManager):

_path = '/groups/%(group_id)s/milestones'

_obj_cls = GroupMilestone

_from_parent_attrs = {'group_id': 'id'}

_create_attrs = (('title', ), ('description', 'due_date', 'start_date'))

_update_attrs = (tuple(), ('title', 'description', 'due_date',

'start_date', 'state_event'))

_list_filters = ('iids', 'state', 'search')

class GroupNotificationSettings(NotificationSettings):

pass

class GroupNotificationSettingsManager(NotificationSettingsManager):

_path = '/groups/%(group_id)s/notification_settings'

_obj_cls = GroupNotificationSettings

_from_parent_attrs = {'group_id': 'id'}

class GroupProject(RESTObject):

pass

class GroupProjectManager(GetFromListMixin, RESTManager):

_path = '/groups/%(group_id)s/projects'

_obj_cls = GroupProject

_from_parent_attrs = {'group_id': 'id'}

_list_filters = ('archived', 'visibility', 'order_by', 'sort', 'search',

'ci_enabled_first')

class GroupSubgroup(RESTObject):

pass

class GroupSubgroupManager(GetFromListMixin, RESTManager):

_path = '/groups/%(group_id)s/subgroups'

_obj_cls = GroupSubgroup

_from_parent_attrs = {'group_id': 'id'}

_list_filters = ('skip_groups', 'all_available', 'search', 'order_by',

'sort', 'statistics', 'owned')

class GroupVariable(SaveMixin, ObjectDeleteMixin, RESTObject):

_id_attr = 'key'

class GroupVariableManager(CRUDMixin, RESTManager):

_path = '/groups/%(group_id)s/variables'

_obj_cls = GroupVariable

_from_parent_attrs = {'group_id': 'id'}

_create_attrs = (('key', 'value'), ('protected',))

_update_attrs = (('key', 'value'), ('protected',))

class Group(SaveMixin, ObjectDeleteMixin, RESTObject):

_short_print_attr = 'name'

_managers = (

('accessrequests', 'GroupAccessRequestManager'),

('customattributes', 'GroupCustomAttributeManager'),

('issues', 'GroupIssueManager'),

('members', 'GroupMemberManager'),

('milestones', 'GroupMilestoneManager'),

('notificationsettings', 'GroupNotificationSettingsManager'),

('projects', 'GroupProjectManager'),

('subgroups', 'GroupSubgroupManager'),

('variables', 'GroupVariableManager'),

)

@cli.register_custom_action('Group', ('to_project_id', ))

@exc.on_http_error(exc.GitlabTransferProjectError)

def transfer_project(self, to_project_id, **kwargs):

"""Transfer a project to this group.

Args:

to_project_id (int): ID of the project to transfer

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

Raises:

GitlabAuthenticationError: If authentication is not correct

GitlabTransferProjectError: If the project could not be transfered

"""

path = '/groups/%d/projects/%d' % (self.id, to_project_id)

self.manager.gitlab.http_post(path, **kwargs)

class GroupManager(CRUDMixin, RESTManager):

_path = '/groups'

_obj_cls = Group

_list_filters = ('skip_groups', 'all_available', 'search', 'order_by',

'sort', 'statistics', 'owned', 'custom_attributes')

_create_attrs = (

('name', 'path'),

('description', 'visibility', 'parent_id', 'lfs_enabled',

'request_access_enabled')

)

_update_attrs = (

tuple(),

('name', 'path', 'description', 'visibility', 'lfs_enabled',

'request_access_enabled')

)

class Hook(ObjectDeleteMixin, RESTObject):

_url = '/hooks'

_short_print_attr = 'url'

class HookManager(NoUpdateMixin, RESTManager):

_path = '/hooks'

_obj_cls = Hook

_create_attrs = (('url', ), tuple())

class Issue(RESTObject):

_url = '/issues'

_short_print_attr = 'title'

class IssueManager(GetFromListMixin, RESTManager):

_path = '/issues'

_obj_cls = Issue

_list_filters = ('state', 'labels', 'order_by', 'sort')

class License(RESTObject):

_id_attr = 'key'

class LicenseManager(RetrieveMixin, RESTManager):

_path = '/templates/licenses'

_obj_cls = License

_list_filters = ('popular', )

_optional_get_attrs = ('project', 'fullname')

class Snippet(SaveMixin, ObjectDeleteMixin, RESTObject):

_short_print_attr = 'title'

@cli.register_custom_action('Snippet')

@exc.on_http_error(exc.GitlabGetError)

def content(self, streamed=False, action=None, chunk_size=1024, **kwargs):

"""Return the content of a snippet.

Args:

streamed (bool): If True the data will be processed by chunks of

`chunk_size` and each chunk is passed to `action` for

treatment.

action (callable): Callable responsible of dealing with chunk of

data

chunk_size (int): Size of each chunk

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

Raises:

GitlabAuthenticationError: If authentication is not correct

GitlabGetError: If the content could not be retrieved

Returns:

str: The snippet content

"""

path = '/snippets/%s/raw' % self.get_id()

result = self.manager.gitlab.http_get(path, streamed=streamed,

**kwargs)

return utils.response_content(result, streamed, action, chunk_size)

class SnippetManager(CRUDMixin, RESTManager):

_path = '/snippets'

_obj_cls = Snippet

_create_attrs = (('title', 'file_name', 'content'),

('lifetime', 'visibility'))

_update_attrs = (tuple(),

('title', 'file_name', 'content', 'visibility'))

@cli.register_custom_action('SnippetManager')

def public(self, **kwargs):

"""List all the public snippets.

Args:

all (bool): If True the returned object will be a list

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

Raises:

GitlabListError: If the list could not be retrieved

Returns:

RESTObjectList: A generator for the snippets list

"""

return self.list(path='/snippets/public', **kwargs)

class Namespace(RESTObject):

pass

class NamespaceManager(GetFromListMixin, RESTManager):

_path = '/namespaces'

_obj_cls = Namespace

_list_filters = ('search', )

class PagesDomain(RESTObject):

_id_attr = 'domain'

class PagesDomainManager(ListMixin, RESTManager):

_path = '/pages/domains'

_obj_cls = PagesDomain

class ProjectBoardList(SaveMixin, ObjectDeleteMixin, RESTObject):

pass

class ProjectBoardListManager(CRUDMixin, RESTManager):

_path = '/projects/%(project_id)s/boards/%(board_id)s/lists'

_obj_cls = ProjectBoardList

_from_parent_attrs = {'project_id': 'project_id',

'board_id': 'id'}

_create_attrs = (('label_id', ), tuple())

_update_attrs = (('position', ), tuple())

class ProjectBoard(RESTObject):

_managers = (('lists', 'ProjectBoardListManager'), )

class ProjectBoardManager(GetFromListMixin, RESTManager):

_path = '/projects/%(project_id)s/boards'

_obj_cls = ProjectBoard

_from_parent_attrs = {'project_id': 'id'}

class ProjectBranch(ObjectDeleteMixin, RESTObject):

_id_attr = 'name'

@cli.register_custom_action('ProjectBranch', tuple(),

('developers_can_push',

'developers_can_merge'))

@exc.on_http_error(exc.GitlabProtectError)

def protect(self, developers_can_push=False, developers_can_merge=False,

**kwargs):

"""Protect the branch.

Args:

developers_can_push (bool): Set to True if developers are allowed

to push to the branch

developers_can_merge (bool): Set to True if developers are allowed

to merge to the branch

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

Raises:

GitlabAuthenticationError: If authentication is not correct

GitlabProtectError: If the branch could not be protected

"""

path = '%s/%s/protect' % (self.manager.path, self.get_id())

post_data = {'developers_can_push': developers_can_push,

'developers_can_merge': developers_can_merge}

self.manager.gitlab.http_put(path, post_data=post_data, **kwargs)

self._attrs['protected'] = True

@cli.register_custom_action('ProjectBranch')

@exc.on_http_error(exc.GitlabProtectError)

def unprotect(self, **kwargs):

"""Unprotect the branch.

Args:

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

Raises:

GitlabAuthenticationError: If authentication is not correct

GitlabProtectError: If the branch could not be unprotected

"""

path = '%s/%s/unprotect' % (self.manager.path, self.get_id())

self.manager.gitlab.http_put(path, **kwargs)

self._attrs['protected'] = False

class ProjectBranchManager(NoUpdateMixin, RESTManager):

_path = '/projects/%(project_id)s/repository/branches'

_obj_cls = ProjectBranch

_from_parent_attrs = {'project_id': 'id'}

_create_attrs = (('branch', 'ref'), tuple())

class ProjectCustomAttribute(ObjectDeleteMixin, RESTObject):

_id_attr = 'key'

class ProjectCustomAttributeManager(RetrieveMixin, SetMixin, DeleteMixin,

RESTManager):

_path = '/projects/%(project_id)s/custom_attributes'

_obj_cls = ProjectCustomAttribute

_from_parent_attrs = {'project_id': 'id'}

class ProjectJob(RESTObject):

@cli.register_custom_action('ProjectJob')

@exc.on_http_error(exc.GitlabJobCancelError)

def cancel(self, **kwargs):

"""Cancel the job.

Args:

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

Raises:

GitlabAuthenticationError: If authentication is not correct

GitlabJobCancelError: If the job could not be canceled

"""

path = '%s/%s/cancel' % (self.manager.path, self.get_id())

self.manager.gitlab.http_post(path)

@cli.register_custom_action('ProjectJob')

@exc.on_http_error(exc.GitlabJobRetryError)

def retry(self, **kwargs):

"""Retry the job.

Args:

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

Raises:

GitlabAuthenticationError: If authentication is not correct

GitlabJobRetryError: If the job could not be retried

"""

path = '%s/%s/retry' % (self.manager.path, self.get_id())

self.manager.gitlab.http_post(path)

@cli.register_custom_action('ProjectJob')

@exc.on_http_error(exc.GitlabJobPlayError)

def play(self, **kwargs):

"""Trigger a job explicitly.

Args:

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

Raises:

GitlabAuthenticationError: If authentication is not correct

GitlabJobPlayError: If the job could not be triggered

"""

path = '%s/%s/play' % (self.manager.path, self.get_id())

self.manager.gitlab.http_post(path)

@cli.register_custom_action('ProjectJob')

@exc.on_http_error(exc.GitlabJobEraseError)

def erase(self, **kwargs):

"""Erase the job (remove job artifacts and trace).

Args:

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

Raises:

GitlabAuthenticationError: If authentication is not correct

GitlabJobEraseError: If the job could not be erased

"""

path = '%s/%s/erase' % (self.manager.path, self.get_id())

self.manager.gitlab.http_post(path)

@cli.register_custom_action('ProjectJob')

@exc.on_http_error(exc.GitlabCreateError)

def keep_artifacts(self, **kwargs):

"""Prevent artifacts from being deleted when expiration is set.

Args:

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

Raises:

GitlabAuthenticationError: If authentication is not correct

GitlabCreateError: If the request could not be performed

"""

path = '%s/%s/artifacts/keep' % (self.manager.path, self.get_id())

self.manager.gitlab.http_post(path)

@cli.register_custom_action('ProjectJob')

@exc.on_http_error(exc.GitlabGetError)

def artifacts(self, streamed=False, action=None, chunk_size=1024,

**kwargs):

"""Get the job artifacts.

Args:

streamed (bool): If True the data will be processed by chunks of

`chunk_size` and each chunk is passed to `action` for

treatment

action (callable): Callable responsible of dealing with chunk of

data

chunk_size (int): Size of each chunk