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

)

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

)

_types = {

'confirm': types.LowercaseStringAttribute,

'avatar': types.ImageAttribute,

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

('admin_notification_email', 'after_sign_out_path',

'after_sign_up_text', 'akismet_api_key', 'akismet_enabled',

'circuitbreaker_access_retries', 'circuitbreaker_check_interval',

'circuitbreaker_failure_count_threshold',

'circuitbreaker_failure_reset_time', 'circuitbreaker_storage_timeout',

'clientside_sentry_dsn', 'clientside_sentry_enabled',

'container_registry_token_expire_delay',

'default_artifacts_expire_in', 'default_branch_protection',

'default_group_visibility', 'default_project_visibility',

'default_projects_limit', 'default_snippet_visibility',

'disabled_oauth_sign_in_sources', 'domain_blacklist_enabled',

'domain_blacklist', 'domain_whitelist', 'dsa_key_restriction',

'ecdsa_key_restriction', 'ed25519_key_restriction',

'email_author_in_body', 'enabled_git_access_protocol',

'gravatar_enabled', 'help_page_hide_commercial_content',

'help_page_support_url', 'home_page_url',

'housekeeping_bitmaps_enabled', 'housekeeping_enabled',

'housekeeping_full_repack_period', 'housekeeping_gc_period',

'housekeeping_incremental_repack_period', 'html_emails_enabled',

'import_sources', 'koding_enabled', 'koding_url',

'max_artifacts_size', 'max_attachment_size', 'max_pages_size',

'metrics_enabled', 'metrics_host', 'metrics_method_call_threshold',

'metrics_packet_size', 'metrics_pool_size', 'metrics_port',

'metrics_sample_interval', 'metrics_timeout',

'password_authentication_enabled_for_web',

'password_authentication_enabled_for_git',

'performance_bar_allowed_group_id', 'performance_bar_enabled',

'plantuml_enabled', 'plantuml_url', 'polling_interval_multiplier',

'project_export_enabled', 'prometheus_metrics_enabled',

'recaptcha_enabled', 'recaptcha_private_key', 'recaptcha_site_key',

'repository_checks_enabled', 'repository_storages',

'require_two_factor_authentication', 'restricted_visibility_levels',

'rsa_key_restriction', 'send_user_confirmation_email', 'sentry_dsn',

'sentry_enabled', 'session_expire_delay', 'shared_runners_enabled',

'shared_runners_text', 'sidekiq_throttling_enabled',

'sidekiq_throttling_factor', 'sidekiq_throttling_queues',

'sign_in_text', 'signup_enabled', 'terminal_max_session_time',

'two_factor_grace_period', 'unique_ips_limit_enabled',

'unique_ips_limit_per_user', 'unique_ips_limit_time_window',

'usage_ping_enabled', 'user_default_external',

'user_oauth_applications', 'version_check_enabled', 'enforce_terms',

'terms')

)

@exc.on_http_error(exc.GitlabUpdateError)

def update(self, id=None, new_data={}, **kwargs):

"""Update an object on the server.

data = new_data.copy()

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

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

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

_obj_cls = GroupBadge

_from_parent_attrs = {'group_id': 'id'}

_create_attrs = (('link_url', 'image_url'), tuple())

_path = '/groups/%(group_id)s/boards/%(board_id)s/lists'

_obj_cls = GroupBoardList

_from_parent_attrs = {'group_id': 'group_id',

'board_id': 'id'}

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

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

_obj_cls = GroupBoard

_from_parent_attrs = {'group_id': 'id'}

RESTManager):

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

_obj_cls = GroupCustomAttribute

_id_attr = 'epic_issue_id'

def save(self, **kwargs):

"""Save the changes made to the object to the server.

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

RESTManager):

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

_obj_cls = GroupEpicIssue

_from_parent_attrs = {'group_id': 'group_id', 'epic_iid': 'iid'}

_create_attrs = (('issue_id',), tuple())

_update_attrs = (tuple(), ('move_before_id', 'move_after_id'))

@exc.on_http_error(exc.GitlabCreateError)

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

"""Create a new object.

CreateMixin._check_missing_create_attrs(self, data)

path = '%s/%s' % (self.path, data.pop('issue_id'))

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

# The epic_issue_id attribute doesn't exist when creating the resource,

# but is used everywhere elese. Let's create it to be consistent client

# side

server_data['epic_issue_id'] = server_data['id']

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

_obj_cls = GroupEpic

_from_parent_attrs = {'group_id': 'id'}

_list_filters = ('author_id', 'labels', 'order_by', 'sort', 'search')

_create_attrs = (('title',),

('labels', 'description', 'start_date', 'end_date'))

_update_attrs = (tuple(), ('title', 'labels', 'description', 'start_date',

'end_date'))

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

_obj_cls = GroupIssue

_from_parent_attrs = {'group_id': 'id'}

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

'iids', 'author_id', 'assignee_id', 'my_reaction_emoji',

'search', 'created_after', 'created_before',

'updated_after', 'updated_before')

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

'ci_enabled_first', 'simple', 'owned', 'starred',

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

('badges', 'GroupBadgeManager'),

('boards', 'GroupBoardManager'),

('customattributes', 'GroupCustomAttributeManager'),

('epics', 'GroupEpicManager'),

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

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

@cli.register_custom_action('Group', ('scope', 'search'))

@exc.on_http_error(exc.GitlabSearchError)

def search(self, scope, search, **kwargs):

"""Search the group resources matching the provided string.'

data = {'scope': scope, 'search': search}

path = '/groups/%d/search' % self.get_id()

return self.manager.gitlab.http_list(path, query_data=data, **kwargs)

@cli.register_custom_action('Group', ('cn', 'group_access', 'provider'))

@exc.on_http_error(exc.GitlabCreateError)

def add_ldap_group_link(self, cn, group_access, provider, **kwargs):

"""Add an LDAP group link.

path = '/groups/%d/ldap_group_links' % self.get_id()

data = {'cn': cn, 'group_access': group_access, 'provider': provider}

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

@cli.register_custom_action('Group', ('cn',), ('provider',))

@exc.on_http_error(exc.GitlabDeleteError)

def delete_ldap_group_link(self, cn, provider=None, **kwargs):

"""Delete an LDAP group link.

path = '/groups/%d/ldap_group_links' % self.get_id()

if provider is not None:

path += '/%s' % provider

path += '/%s' % cn

self.manager.gitlab.http_delete(path)

@cli.register_custom_action('Group')

@exc.on_http_error(exc.GitlabCreateError)

def ldap_sync(self, **kwargs):

"""Sync LDAP groups.

path = '/groups/%d/ldap_sync' % self.get_id()

_path = '/groups'

_obj_cls = Group

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

'sort', 'statistics', 'owned', 'with_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 = '/issues'

_obj_cls = Issue

_list_filters = ('state', 'labels', 'milestone', 'scope', 'author_id',

'assignee_id', 'my_reaction_emoji', 'iids', 'order_by',

'sort', 'search', 'created_after', 'created_before',

'updated_after', 'updated_before')

_path = '/ldap/groups'

_obj_cls = LDAPGroup

_list_filters = ('search', 'provider')

@exc.on_http_error(exc.GitlabListError)

def list(self, **kwargs):

"""Retrieve a list of objects.