"""Manager for the Sidekiq methods.

@cli.register_custom_action('SidekiqManager')

@exc.on_http_error(exc.GitlabGetError)

def queue_metrics(self, **kwargs):

"""Return the registred queues information.

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.

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.

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.

RESTManager):

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

_obj_cls = UserCustomAttribute

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

_obj_cls = UserEmail

_from_parent_attrs = {'user_id': 'id'}

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

_obj_cls = UserGPGKey

_from_parent_attrs = {'user_id': 'id'}

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

_obj_cls = UserKey

_from_parent_attrs = {'user_id': 'id'}

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

_obj_cls = UserImpersonationToken

_from_parent_attrs = {'user_id': 'id'}

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

_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.

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

_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.

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.

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

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

if server_data is True:

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

_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()

RESTManager):

_path = '/user/emails'

_obj_cls = CurrentUserEmail

RESTManager):

_path = '/user/gpg_keys'

_obj_cls = CurrentUserGPGKey

RESTManager):

_path = '/user/keys'

_obj_cls = CurrentUserKey

_id_attr = None

_short_print_attr = 'username'

_managers = (

('emails', 'CurrentUserEmailManager'),

('gpgkeys', 'CurrentUserGPGKeyManager'),

('keys', 'CurrentUserKeyManager'),

_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')

_path = '/broadcast_messages'

_obj_cls = BroadcastMessage

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

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

_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')

_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.

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)

RESTManager):

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

_obj_cls = GroupAccessRequest

RESTManager):

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

_obj_cls = GroupCustomAttribute

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

_obj_cls = GroupIssue

_from_parent_attrs = {'group_id': 'id'}

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', ))

_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.

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.

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

_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'))

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

_obj_cls = GroupProject

_from_parent_attrs = {'group_id': 'id'}

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

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

_obj_cls = GroupSubgroup

_from_parent_attrs = {'group_id': 'id'}

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

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

_obj_cls = GroupVariable

_from_parent_attrs = {'group_id': 'id'}

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

_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.

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

_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')

_path = '/templates/licenses'

_obj_cls = License

_list_filters = ('popular', )

_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.

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

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

**kwargs)

_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.

_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())

_id_attr = 'name'

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

@exc.on_http_error(exc.GitlabProtectError)

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

**kwargs):

"""Protect the branch.

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.

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

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

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

_obj_cls = ProjectBranch

_from_parent_attrs = {'project_id': 'id'}

RESTManager):

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

_obj_cls = ProjectCustomAttribute

@cli.register_custom_action('ProjectJob')

@exc.on_http_error(exc.GitlabJobCancelError)

def cancel(self, **kwargs):

"""Cancel the job.

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.

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.

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).

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.

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.