"""Manager for the Sidekiq methods.

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

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

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

@exc.on_http_error(exc.GitlabGetError)

def compound_metrics(self, **kwargs):

"""Return all available metrics and statistics.

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

_obj_cls = UserEmail

_from_parent_attrs = {'user_id': 'id'}

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

_obj_cls = UserKey

_from_parent_attrs = {'user_id': 'id'}

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

_short_print_attr = 'username'

_managers = (

('emails', 'UserEmailManager'),

('keys', 'UserKeyManager'),

('projects', 'UserProjectManager'),

)

@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

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

_create_attrs = (

('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/keys'

_obj_cls = CurrentUserKey

_id_attr = None

_short_print_attr = 'username'

_managers = (

('emails', 'CurrentUserEmailManager'),

('keys', 'CurrentUserKeyManager'),

_path = '/user'

_obj_cls = CurrentUser

def credentials_auth(self, email, password):

data = {'email': email, 'password': password}

server_data = self.gitlab.http_post('/session', post_data=data)

_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 = '/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', ))

RESTManager):

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

_obj_cls = GroupAccessRequest

_url = '/issues'

_constructor_types = {'author': 'User',

'assignee': 'User',

'milestone': 'ProjectMilestone'}

_path = '/templates/licenses'

_obj_cls = License

_list_filters = ('popular', )

_constructor_types = {'author': 'User'}

_short_print_attr = 'title'

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

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

_constructor_types = {'author': 'User', "committer": "User"}

_id_attr = 'name'

@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

@exc.on_http_error(exc.GitlabProtectError)

def unprotect(self, **kwargs):

"""Unprotect the branch.

path = '%s/%s/protect' % (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'}

_constructor_types = {'user': 'User',

'commit': 'ProjectCommit',

'runner': 'Runner'}

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

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

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

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

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

@exc.on_http_error(exc.GitlabGetError)

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

**kwargs):

"""Get the job artifacts.

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

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

**kwargs)

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

@exc.on_http_error(exc.GitlabGetError)

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

"""Get the job trace.

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

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

**kwargs)

_path = ('/projects/%(project_id)s/repository/commits/%(commit_id)s'

'/statuses')

_obj_cls = ProjectCommitStatus

_from_parent_attrs = {'project_id': 'project_id', 'commit_id': 'id'}

_create_attrs = (('state', ),

('description', 'name', 'context', 'ref', 'target_url'))

def create(self, data, **kwargs):

"""Create a new object.

path = '/projects/%(project_id)s/statuses/%(commit_id)s'

computed_path = self._compute_path(path)

_path = ('/projects/%(project_id)s/repository/commits/%(commit_id)s'

'/comments')

_obj_cls = ProjectCommitComment

_from_parent_attrs = {'project_id': 'project_id', 'commit_id': 'id'}

_short_print_attr = 'title'

_managers = (

('comments', 'ProjectCommitCommentManager'),

('statuses', 'ProjectCommitStatusManager'),

)

@exc.on_http_error(exc.GitlabGetError)

def diff(self, **kwargs):

"""Generate the commit diff.

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

return self.manager.gitlab.http_get(path, **kwargs)

@exc.on_http_error(exc.GitlabCherryPickError)

def cherry_pick(self, branch, **kwargs):

"""Cherry-pick a commit into a branch.

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

post_data = {'branch': branch}

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

_obj_cls = ProjectCommit

_from_parent_attrs = {'project_id': 'id'}

_create_attrs = (('branch', 'commit_message', 'actions'),

DeleteMixin, RESTManager):

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

_obj_cls = ProjectEnvironment

_from_parent_attrs = {'project_id': 'id'}

_create_attrs = (('name', ), ('external_url', ))

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

_obj_cls = ProjectKey

_from_parent_attrs = {'project_id': 'id'}

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

@exc.on_http_error(exc.GitlabProjectDeployKeyError)

def enable(self, key_id, **kwargs):

"""Enable a deploy key for a project.

path = '%s/%s/enable' % (self.manager.path, key_id)

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

_obj_cls = ProjectFork

_from_parent_attrs = {'project_id': 'id'}

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

_obj_cls = ProjectHook

_from_parent_attrs = {'project_id': 'id'}

_create_attrs = (

('url', ),

('push_events', 'issues_events', 'note_events',

'merge_requests_events', 'tag_push_events', 'build_events',

'enable_ssl_verification', 'token', 'pipeline_events')

)

_update_attrs = (

('url', ),

('push_events', 'issues_events', 'note_events',

'merge_requests_events', 'tag_push_events', 'build_events',

'enable_ssl_verification', 'token', 'pipeline_events')

_path = '/projects/%(project_id)s/issues/%(issue_iid)s/notes'

_obj_cls = ProjectIssueNote

_from_parent_attrs = {'project_id': 'project_id', 'issue_iid': 'iid'}

_create_attrs = (('body', ), ('created_at'))

ObjectDeleteMixin, RESTObject):

_constructor_types = {'author': 'User', 'assignee': 'User', 'milestone':

'ProjectMilestone'}

_short_print_attr = 'title'

_id_attr = 'iid'

_managers = (('notes', 'ProjectIssueNoteManager'), )

@exc.on_http_error(exc.GitlabUpdateError)

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

"""Move the issue to another project.

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

data = {'to_project_id': to_project_id}

server_data = self.manager.gitlab.http_post(path, post_data=data,

**kwargs)

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

_obj_cls = ProjectIssue

_from_parent_attrs = {'project_id': 'id'}

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

_create_attrs = (('title', ),

('description', 'assignee_id', 'milestone_id', 'labels',

'created_at', 'due_date'))

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

'milestone_id', 'labels', 'created_at',

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

_obj_cls = ProjectMember

_from_parent_attrs = {'project_id': 'id'}

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

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

_obj_cls = ProjectNote

_from_parent_attrs = {'project_id': 'id'}

_constructor_types = {'release': 'ProjectTagRelease',

'commit': 'ProjectCommit'}

_id_attr = 'name'

_short_print_attr = 'name'

def set_release_description(self, description, **kwargs):

"""Set the release notes on the tag.

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

data = {'description': description}

if self.release is None: