# -*- coding: utf-8 -*-
#
# Copyright (C) 2013-2017 Gauvain Pocentek
#
# 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 .
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 types
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(ListMixin, 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 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', '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,
}
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(),
('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.
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
"""
data = new_data.copy()
if 'domain_whitelist' in data and data['domain_whitelist'] is None:
data.pop('domain_whitelist')
super(ApplicationSettingsManager, self).update(id, data, **kwargs)
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(ListMixin, 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(ObjectDeleteMixin, RESTObject):
_id_attr = 'name'
class FeatureManager(ListMixin, DeleteMixin, 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(ListMixin, CreateMixin, DeleteMixin,
RESTManager):
_path = '/groups/%(group_id)s/access_requests'
_obj_cls = GroupAccessRequest
_from_parent_attrs = {'group_id': 'id'}
class GroupBadge(SaveMixin, ObjectDeleteMixin, RESTObject):
pass
class GroupBadgeManager(BadgeRenderMixin, CRUDMixin, RESTManager):
_path = '/groups/%(group_id)s/badges'
_obj_cls = GroupBadge
_from_parent_attrs = {'group_id': 'id'}
_create_attrs = (('link_url', 'image_url'), tuple())
_update_attrs = (tuple(), ('link_url', 'image_url'))
class GroupBoardList(SaveMixin, ObjectDeleteMixin, RESTObject):
pass
class GroupBoardListManager(CRUDMixin, RESTManager):
_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())
_update_attrs = (('position', ), tuple())
class GroupBoard(ObjectDeleteMixin, RESTObject):
_managers = (('lists', 'GroupBoardListManager'), )
class GroupBoardManager(NoUpdateMixin, RESTManager):
_path = '/groups/%(group_id)s/boards'
_obj_cls = GroupBoard
_from_parent_attrs = {'group_id': 'id'}
_create_attrs = (('name', ), tuple())
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 GroupEpicIssue(ObjectDeleteMixin, SaveMixin, RESTObject):
_id_attr = 'epic_issue_id'
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()
self.manager.update(obj_id, updated_data, **kwargs)
class GroupEpicIssueManager(ListMixin, CreateMixin, UpdateMixin, DeleteMixin,
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.
Args:
data (dict): Parameters to send to the server to create the
resource
**kwargs: Extra options to send to the server (e.g. sudo)
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabCreateError: If the server cannot perform the request
Returns:
RESTObject: A new instance of the manage object class build with
the data sent by the server
"""
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']
return self._obj_cls(self, server_data)
class GroupEpic(ObjectDeleteMixin, SaveMixin, RESTObject):
_id_attr = 'iid'
_managers = (('issues', 'GroupEpicIssueManager'),)
class GroupEpicManager(CRUDMixin, RESTManager):
_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'))
_types = {'labels': types.ListAttribute}
class GroupIssue(RESTObject):
pass
class GroupIssueManager(ListMixin, RESTManager):
_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')
_types = {'labels': types.ListAttribute}
class GroupMember(SaveMixin, ObjectDeleteMixin, RESTObject):
_short_print_attr = 'username'
class GroupMemberManager(CRUDMixin, 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(ListMixin, RESTManager):
_path = '/groups/%(group_id)s/merge_requests'
_obj_cls = GroupMergeRequest
_from_parent_attrs = {'group_id': 'id'}
_list_filters = ('state', 'order_by', 'sort', 'milestone', 'view',
'labels', 'created_after', 'created_before',
'updated_after', 'updated_before', 'scope', 'author_id',
'assignee_id', 'my_reaction_emoji', 'source_branch',
'target_branch', 'search')
_types = {'labels': types.ListAttribute}
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(ListMixin, 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', 'simple', 'owned', 'starred',
'with_custom_attributes')
class GroupSubgroup(RESTObject):
pass
class GroupSubgroupManager(ListMixin, 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', 'with_custom_attributes')
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'),
('badges', 'GroupBadgeManager'),
('boards', 'GroupBoardManager'),
('customattributes', 'GroupCustomAttributeManager'),
('epics', 'GroupEpicManager'),
('issues', 'GroupIssueManager'),
('members', 'GroupMemberManager'),
('mergerequests', 'GroupMergeRequestManager'),
('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)
@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.'
Args:
scope (str): Scope of the search
search (str): Search string
**kwargs: Extra options to send to the server (e.g. sudo)
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabSearchError: If the server failed to perform the request
Returns:
GitlabList: A list of dicts describing the resources found.
"""
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.
Args:
cn (str): CN of the LDAP group
group_access (int): Minimum access level for members of the LDAP
group
provider (str): LDAP provider for the LDAP group
**kwargs: Extra options to send to the server (e.g. sudo)
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabCreateError: If the server cannot perform the request
"""
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.
Args:
cn (str): CN of the LDAP group
provider (str): LDAP provider for the LDAP group
**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
"""
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.
Args:
**kwargs: Extra options to send to the server (e.g. sudo)
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabCreateError: If the server cannot perform the request
"""
path = '/groups/%d/ldap_sync' % self.get_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', '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')
)
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(ListMixin, RESTManager):
_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')
_types = {'labels': types.ListAttribute}
class LDAPGroup(RESTObject):
_id_attr = None
class LDAPGroupManager(RESTManager):
_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.
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
"""
data = kwargs.copy()
if self.gitlab.per_page:
data.setdefault('per_page', self.gitlab.per_page)
if 'provider' in data:
path = '/ldap/%s/groups' % data['provider']
else:
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)
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 MergeRequest(RESTObject):
pass
class MergeRequestManager(ListMixin, RESTManager):
_path = '/merge_requests'
_obj_cls = MergeRequest
_from_parent_attrs = {'group_id': 'id'}
_list_filters = ('state', 'order_by', 'sort', 'milestone', 'view',
'labels', 'created_after', 'created_before',
'updated_after', 'updated_before', 'scope', 'author_id',
'assignee_id', 'my_reaction_emoji', 'source_branch',
'target_branch', 'search')
_types = {'labels': types.ListAttribute}
class Snippet(UserAgentDetailMixin, 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(RetrieveMixin, 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(ObjectDeleteMixin, RESTObject):
_managers = (('lists', 'ProjectBoardListManager'), )
class ProjectBoardManager(NoUpdateMixin, RESTManager):
_path = '/projects/%(project_id)s/boards'
_obj_cls = ProjectBoard
_from_parent_attrs = {'project_id': 'id'}
_create_attrs = (('name', ), tuple())
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
"""
id = self.get_id().replace('/', '%2F')
path = '%s/%s/protect' % (self.manager.path, 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
"""
id = self.get_id().replace('/', '%2F')
path = '%s/%s/unprotect' % (self.manager.path, 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, RefreshMixin):
@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
**kwargs: Extra options to send to the server (e.g. sudo)
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabGetError: If the artifacts could not be retrieved
Returns:
str: The artifacts if `streamed` is False, None otherwise.
"""
path = '%s/%s/artifacts' % (self.manager.path, self.get_id())
result = self.manager.gitlab.http_get(path, streamed=streamed,
**kwargs)
return utils.response_content(result, streamed, action, chunk_size)
@cli.register_custom_action('ProjectJob')
@exc.on_http_error(exc.GitlabGetError)
def artifact(self, path, streamed=False, action=None, chunk_size=1024,
**kwargs):
"""Get a single artifact file from within the job's artifacts archive.
Args:
path (str): Path of the artifact
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 artifacts could not be retrieved
Returns:
str: The artifacts if `streamed` is False, None otherwise.
"""
path = '%s/%s/artifacts/%s' % (self.manager.path, self.get_id(), path)
result = self.manager.gitlab.http_get(path, streamed=streamed,
**kwargs)
return utils.response_content(result, streamed, action, chunk_size)
@cli.register_custom_action('ProjectJob')
@exc.on_http_error(exc.GitlabGetError)
def trace(self, streamed=False, action=None, chunk_size=1024, **kwargs):
"""Get the job trace.
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 artifacts could not be retrieved
Returns:
str: The trace
"""
path = '%s/%s/trace' % (self.manager.path, self.get_id())
result = self.manager.gitlab.http_get(path, streamed=streamed,
**kwargs)
return utils.response_content(result, streamed, action, chunk_size)
class ProjectJobManager(RetrieveMixin, RESTManager):
_path = '/projects/%(project_id)s/jobs'
_obj_cls = ProjectJob
_from_parent_attrs = {'project_id': 'id'}
class ProjectCommitStatus(RESTObject, RefreshMixin):
pass
class ProjectCommitStatusManager(ListMixin, CreateMixin, RESTManager):
_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',
'coverage'))
@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 or
'ref_name', 'stage', 'name', 'all')
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabCreateError: If the server cannot perform the request
Returns:
RESTObject: A new instance of the manage object class build with
the data sent by the server
"""
# project_id and commit_id are in the data dict when using the CLI, but
# they are missing when using only the API
# See #511
base_path = '/projects/%(project_id)s/statuses/%(commit_id)s'
if 'project_id' in data and 'commit_id' in data:
path = base_path % data
else:
path = self._compute_path(base_path)
return CreateMixin.create(self, data, path=path, **kwargs)
class ProjectCommitComment(RESTObject):
_id_attr = None
_short_print_attr = 'note'
class ProjectCommitCommentManager(ListMixin, CreateMixin, RESTManager):
_path = ('/projects/%(project_id)s/repository/commits/%(commit_id)s'
'/comments')
_obj_cls = ProjectCommitComment
_from_parent_attrs = {'project_id': 'project_id', 'commit_id': 'id'}
_create_attrs = (('note', ), ('path', 'line', 'line_type'))
class ProjectCommitDiscussionNote(SaveMixin, ObjectDeleteMixin, RESTObject):
pass
class ProjectCommitDiscussionNoteManager(GetMixin, CreateMixin, UpdateMixin,
DeleteMixin, RESTManager):
_path = ('/projects/%(project_id)s/repository/commits/%(commit_id)s/'
'discussions/%(discussion_id)s/notes')
_obj_cls = ProjectCommitDiscussionNote
_from_parent_attrs = {'project_id': 'project_id',
'commit_id': 'commit_id',
'discussion_id': 'id'}
_create_attrs = (('body',), ('created_at', 'position'))
_update_attrs = (('body',), tuple())
class ProjectCommitDiscussion(RESTObject):
_managers = (('notes', 'ProjectCommitDiscussionNoteManager'),)
class ProjectCommitDiscussionManager(RetrieveMixin, CreateMixin, RESTManager):
_path = ('/projects/%(project_id)s/repository/commits/%(commit_id)s/'
'discussions')
_obj_cls = ProjectCommitDiscussion
_from_parent_attrs = {'project_id': 'project_id', 'commit_id': 'id'}
_create_attrs = (('body',), ('created_at',))
class ProjectCommit(RESTObject):
_short_print_attr = 'title'
_managers = (
('comments', 'ProjectCommitCommentManager'),
('discussions', 'ProjectCommitDiscussionManager'),
('statuses', 'ProjectCommitStatusManager'),
)
@cli.register_custom_action('ProjectCommit')
@exc.on_http_error(exc.GitlabGetError)
def diff(self, **kwargs):
"""Generate the commit diff.
Args:
**kwargs: Extra options to send to the server (e.g. sudo)
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabGetError: If the diff could not be retrieved
Returns:
list: The changes done in this commit
"""
path = '%s/%s/diff' % (self.manager.path, self.get_id())
return self.manager.gitlab.http_get(path, **kwargs)
@cli.register_custom_action('ProjectCommit', ('branch',))
@exc.on_http_error(exc.GitlabCherryPickError)
def cherry_pick(self, branch, **kwargs):
"""Cherry-pick a commit into a branch.
Args:
branch (str): Name of target branch
**kwargs: Extra options to send to the server (e.g. sudo)
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabCherryPickError: If the cherry-pick could not be performed
"""
path = '%s/%s/cherry_pick' % (self.manager.path, self.get_id())
post_data = {'branch': branch}
self.manager.gitlab.http_post(path, post_data=post_data, **kwargs)
@cli.register_custom_action('ProjectCommit', optional=('type',))
@exc.on_http_error(exc.GitlabGetError)
def refs(self, type='all', **kwargs):
"""List the references the commit is pushed to.
Args:
type (str): The scope of references ('branch', 'tag' or 'all')
**kwargs: Extra options to send to the server (e.g. sudo)
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabGetError: If the references could not be retrieved
Returns:
list: The references the commit is pushed to.
"""
path = '%s/%s/refs' % (self.manager.path, self.get_id())
data = {'type': type}
return self.manager.gitlab.http_get(path, query_data=data, **kwargs)
@cli.register_custom_action('ProjectCommit')
@exc.on_http_error(exc.GitlabGetError)
def merge_requests(self, **kwargs):
"""List the merge requests related to the commit.
Args:
**kwargs: Extra options to send to the server (e.g. sudo)
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabGetError: If the references could not be retrieved
Returns:
list: The merge requests related to the commit.
"""
path = '%s/%s/merge_requests' % (self.manager.path, self.get_id())
return self.manager.gitlab.http_get(path, **kwargs)
class ProjectCommitManager(RetrieveMixin, CreateMixin, RESTManager):
_path = '/projects/%(project_id)s/repository/commits'
_obj_cls = ProjectCommit
_from_parent_attrs = {'project_id': 'id'}
_create_attrs = (('branch', 'commit_message', 'actions'),
('author_email', 'author_name'))
class ProjectEnvironment(SaveMixin, ObjectDeleteMixin, RESTObject):
@cli.register_custom_action('ProjectEnvironment')
@exc.on_http_error(exc.GitlabStopError)
def stop(self, **kwargs):
"""Stop the environment.
Args:
**kwargs: Extra options to send to the server (e.g. sudo)
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabStopError: If the operation failed
"""
path = '%s/%s/stop' % (self.manager.path, self.get_id())
self.manager.gitlab.http_post(path, **kwargs)
class ProjectEnvironmentManager(ListMixin, CreateMixin, UpdateMixin,
DeleteMixin, RESTManager):
_path = '/projects/%(project_id)s/environments'
_obj_cls = ProjectEnvironment
_from_parent_attrs = {'project_id': 'id'}
_create_attrs = (('name', ), ('external_url', ))
_update_attrs = (tuple(), ('name', 'external_url'))
class ProjectKey(SaveMixin, ObjectDeleteMixin, RESTObject):
pass
class ProjectKeyManager(CRUDMixin, RESTManager):
_path = '/projects/%(project_id)s/deploy_keys'
_obj_cls = ProjectKey
_from_parent_attrs = {'project_id': 'id'}
_create_attrs = (('title', 'key'), ('can_push',))
_update_attrs = (tuple(), ('title', 'can_push'))
@cli.register_custom_action('ProjectKeyManager', ('key_id',))
@exc.on_http_error(exc.GitlabProjectDeployKeyError)
def enable(self, key_id, **kwargs):
"""Enable a deploy key for a project.
Args:
key_id (int): The ID of the key to enable
**kwargs: Extra options to send to the server (e.g. sudo)
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabProjectDeployKeyError: If the key could not be enabled
"""
path = '%s/%s/enable' % (self.path, key_id)
self.gitlab.http_post(path, **kwargs)
class ProjectBadge(SaveMixin, ObjectDeleteMixin, RESTObject):
pass
class ProjectBadgeManager(BadgeRenderMixin, CRUDMixin, RESTManager):
_path = '/projects/%(project_id)s/badges'
_obj_cls = ProjectBadge
_from_parent_attrs = {'project_id': 'id'}
_create_attrs = (('link_url', 'image_url'), tuple())
_update_attrs = (tuple(), ('link_url', 'image_url'))
class ProjectEvent(Event):
pass
class ProjectEventManager(EventManager):
_path = '/projects/%(project_id)s/events'
_obj_cls = ProjectEvent
_from_parent_attrs = {'project_id': 'id'}
class ProjectFork(RESTObject):
pass
class ProjectForkManager(CreateMixin, ListMixin, RESTManager):
_path = '/projects/%(project_id)s/fork'
_obj_cls = ProjectFork
_from_parent_attrs = {'project_id': 'id'}
_list_filters = ('archived', 'visibility', 'order_by', 'sort', 'search',
'simple', 'owned', 'membership', 'starred', 'statistics',
'with_custom_attributes', 'with_issues_enabled',
'with_merge_requests_enabled')
_create_attrs = (tuple(), ('namespace', ))
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
"""
path = self._compute_path('/projects/%(project_id)s/forks')
return ListMixin.list(self, path=path, **kwargs)
class ProjectHook(SaveMixin, ObjectDeleteMixin, RESTObject):
_short_print_attr = 'url'
class ProjectHookManager(CRUDMixin, RESTManager):
_path = '/projects/%(project_id)s/hooks'
_obj_cls = ProjectHook
_from_parent_attrs = {'project_id': 'id'}
_create_attrs = (
('url', ),
('push_events', 'issues_events', 'confidential_issues_events',
'merge_requests_events', 'tag_push_events', 'note_events',
'job_events', 'pipeline_events', 'wiki_page_events',
'enable_ssl_verification', 'token')
)
_update_attrs = (
('url', ),
('push_events', 'issues_events', 'confidential_issues_events',
'merge_requests_events', 'tag_push_events', 'note_events',
'job_events', 'pipeline_events', 'wiki_events',
'enable_ssl_verification', 'token')
)
class ProjectIssueAwardEmoji(ObjectDeleteMixin, RESTObject):
pass
class ProjectIssueAwardEmojiManager(NoUpdateMixin, RESTManager):
_path = '/projects/%(project_id)s/issues/%(issue_iid)s/award_emoji'
_obj_cls = ProjectIssueAwardEmoji
_from_parent_attrs = {'project_id': 'project_id', 'issue_iid': 'iid'}
_create_attrs = (('name', ), tuple())
class ProjectIssueNoteAwardEmoji(ObjectDeleteMixin, RESTObject):
pass
class ProjectIssueNoteAwardEmojiManager(NoUpdateMixin, RESTManager):
_path = ('/projects/%(project_id)s/issues/%(issue_iid)s'
'/notes/%(note_id)s/award_emoji')
_obj_cls = ProjectIssueNoteAwardEmoji
_from_parent_attrs = {'project_id': 'project_id',
'issue_iid': 'issue_iid',
'note_id': 'id'}
_create_attrs = (('name', ), tuple())
class ProjectIssueNote(SaveMixin, ObjectDeleteMixin, RESTObject):
_managers = (('awardemojis', 'ProjectIssueNoteAwardEmojiManager'),)
class ProjectIssueNoteManager(CRUDMixin, RESTManager):
_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', ))
_update_attrs = (('body', ), tuple())
class ProjectIssueDiscussionNote(SaveMixin, ObjectDeleteMixin, RESTObject):
pass
class ProjectIssueDiscussionNoteManager(GetMixin, CreateMixin, UpdateMixin,
DeleteMixin, RESTManager):
_path = ('/projects/%(project_id)s/issues/%(issue_iid)s/'
'discussions/%(discussion_id)s/notes')
_obj_cls = ProjectIssueDiscussionNote
_from_parent_attrs = {'project_id': 'project_id',
'issue_iid': 'issue_iid',
'discussion_id': 'id'}
_create_attrs = (('body',), ('created_at',))
_update_attrs = (('body',), tuple())
class ProjectIssueDiscussion(RESTObject):
_managers = (('notes', 'ProjectIssueDiscussionNoteManager'),)
class ProjectIssueDiscussionManager(RetrieveMixin, CreateMixin, RESTManager):
_path = '/projects/%(project_id)s/issues/%(issue_iid)s/discussions'
_obj_cls = ProjectIssueDiscussion
_from_parent_attrs = {'project_id': 'project_id', 'issue_iid': 'iid'}
_create_attrs = (('body',), ('created_at',))
class ProjectIssueLink(ObjectDeleteMixin, RESTObject):
_id_attr = 'issue_link_id'
class ProjectIssueLinkManager(ListMixin, CreateMixin, DeleteMixin,
RESTManager):
_path = '/projects/%(project_id)s/issues/%(issue_iid)s/links'
_obj_cls = ProjectIssueLink
_from_parent_attrs = {'project_id': 'project_id', 'issue_iid': 'iid'}
_create_attrs = (('target_project_id', 'target_issue_iid'), 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, RESTObject: The source and target issues
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabCreateError: If the server cannot perform the request
"""
self._check_missing_create_attrs(data)
server_data = self.gitlab.http_post(self.path, post_data=data,
**kwargs)
source_issue = ProjectIssue(self._parent.manager,
server_data['source_issue'])
target_issue = ProjectIssue(self._parent.manager,
server_data['target_issue'])
return source_issue, target_issue
class ProjectIssue(UserAgentDetailMixin, SubscribableMixin, TodoMixin,
TimeTrackingMixin, ParticipantsMixin, SaveMixin,
ObjectDeleteMixin, RESTObject):
_short_print_attr = 'title'
_id_attr = 'iid'
_managers = (
('awardemojis', 'ProjectIssueAwardEmojiManager'),
('discussions', 'ProjectIssueDiscussionManager'),
('links', 'ProjectIssueLinkManager'),
('notes', 'ProjectIssueNoteManager'),
)
@cli.register_custom_action('ProjectIssue', ('to_project_id',))
@exc.on_http_error(exc.GitlabUpdateError)
def move(self, to_project_id, **kwargs):
"""Move the issue to another project.
Args:
to_project_id(int): ID of the target project
**kwargs: Extra options to send to the server (e.g. sudo)
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabUpdateError: If the issue could not be moved
"""
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)
self._update_attrs(server_data)
@cli.register_custom_action('ProjectIssue')
@exc.on_http_error(exc.GitlabGetError)
def closed_by(self, **kwargs):
"""List merge requests that will close the issue when merged.
Args:
**kwargs: Extra options to send to the server (e.g. sudo)
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabGetErrot: If the merge requests could not be retrieved
Returns:
list: The list of merge requests.
"""
path = '%s/%s/closed_by' % (self.manager.path, self.get_id())
return self.manager.gitlab.http_get(path, **kwargs)
class ProjectIssueManager(CRUDMixin, RESTManager):
_path = '/projects/%(project_id)s/issues'
_obj_cls = ProjectIssue
_from_parent_attrs = {'project_id': 'id'}
_list_filters = ('iids', 'state', 'labels', 'milestone', 'scope',
'author_id', 'assignee_id', 'my_reaction_emoji',
'order_by', 'sort', 'search', 'created_after',
'created_before', 'updated_after', 'updated_before')
_create_attrs = (('title', ),
('description', 'confidential', 'assignee_ids',
'assignee_id', 'milestone_id', 'labels', 'created_at',
'due_date', 'merge_request_to_resolve_discussions_of',
'discussion_to_resolve'))
_update_attrs = (tuple(), ('title', 'description', 'confidential',
'assignee_ids', 'assignee_id', 'milestone_id',
'labels', 'state_event', 'updated_at',
'due_date', 'discussion_locked'))
_types = {'labels': types.ListAttribute}
class ProjectMember(SaveMixin, ObjectDeleteMixin, RESTObject):
_short_print_attr = 'username'
class ProjectMemberManager(CRUDMixin, RESTManager):
_path = '/projects/%(project_id)s/members'
_obj_cls = ProjectMember
_from_parent_attrs = {'project_id': 'id'}
_create_attrs = (('access_level', 'user_id'), ('expires_at', ))
_update_attrs = (('access_level', ), ('expires_at', ))
class ProjectNote(RESTObject):
pass
class ProjectNoteManager(RetrieveMixin, RESTManager):
_path = '/projects/%(project_id)s/notes'
_obj_cls = ProjectNote
_from_parent_attrs = {'project_id': 'id'}
_create_attrs = (('body', ), tuple())
class ProjectNotificationSettings(NotificationSettings):
pass
class ProjectNotificationSettingsManager(NotificationSettingsManager):
_path = '/projects/%(project_id)s/notification_settings'
_obj_cls = ProjectNotificationSettings
_from_parent_attrs = {'project_id': 'id'}
class ProjectPagesDomain(SaveMixin, ObjectDeleteMixin, RESTObject):
_id_attr = 'domain'
class ProjectPagesDomainManager(CRUDMixin, RESTManager):
_path = '/projects/%(project_id)s/pages/domains'
_obj_cls = ProjectPagesDomain
_from_parent_attrs = {'project_id': 'id'}
_create_attrs = (('domain', ), ('certificate', 'key'))
_update_attrs = (tuple(), ('certificate', 'key'))
class ProjectTag(ObjectDeleteMixin, RESTObject):
_id_attr = 'name'
_short_print_attr = 'name'
@cli.register_custom_action('ProjectTag', ('description', ))
def set_release_description(self, description, **kwargs):
"""Set the release notes on the tag.
If the release doesn't exist yet, it will be created. If it already
exists, its description will be updated.
Args:
description (str): Description of the release.
**kwargs: Extra options to send to the server (e.g. sudo)
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabCreateError: If the server fails to create the release
GitlabUpdateError: If the server fails to update the release
"""
id = self.get_id().replace('/', '%2F')
path = '%s/%s/release' % (self.manager.path, id)
data = {'description': description}
if self.release is None:
try:
server_data = self.manager.gitlab.http_post(path,
post_data=data,
**kwargs)
except exc.GitlabHttpError as e:
raise exc.GitlabCreateError(e.response_code, e.error_message)
else:
try:
server_data = self.manager.gitlab.http_put(path,
post_data=data,
**kwargs)
except exc.GitlabHttpError as e:
raise exc.GitlabUpdateError(e.response_code, e.error_message)
self.release = server_data
class ProjectTagManager(NoUpdateMixin, RESTManager):
_path = '/projects/%(project_id)s/repository/tags'
_obj_cls = ProjectTag
_from_parent_attrs = {'project_id': 'id'}
_create_attrs = (('tag_name', 'ref'), ('message',))
class ProjectProtectedTag(ObjectDeleteMixin, RESTObject):
_id_attr = 'name'
_short_print_attr = 'name'
class ProjectProtectedTagManager(NoUpdateMixin, RESTManager):
_path = '/projects/%(project_id)s/protected_tags'
_obj_cls = ProjectProtectedTag
_from_parent_attrs = {'project_id': 'id'}
_create_attrs = (('name',), ('create_access_level',))
class ProjectMergeRequestApproval(SaveMixin, RESTObject):
_id_attr = None
class ProjectMergeRequestApprovalManager(GetWithoutIdMixin, UpdateMixin,
RESTManager):
_path = '/projects/%(project_id)s/merge_requests/%(mr_iid)s/approvals'
_obj_cls = ProjectMergeRequestApproval
_from_parent_attrs = {'project_id': 'project_id', 'mr_iid': 'iid'}
_update_attrs = (('approvals_required',), tuple())
_update_uses_post = True
@exc.on_http_error(exc.GitlabUpdateError)
def set_approvers(self, approver_ids=[], approver_group_ids=[], **kwargs):
"""Change MR-level allowed approvers and approver groups.
Args:
approver_ids (list): User IDs that can approve MRs
approver_group_ids (list): Group IDs whose members can approve MRs
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabUpdateError: If the server failed to perform the request
"""
path = '%s/%s/approvers' % (self._parent.manager.path,
self._parent.get_id())
data = {'approver_ids': approver_ids,
'approver_group_ids': approver_group_ids}
self.gitlab.http_put(path, post_data=data, **kwargs)
class ProjectMergeRequestAwardEmoji(ObjectDeleteMixin, RESTObject):
pass
class ProjectMergeRequestAwardEmojiManager(NoUpdateMixin, RESTManager):
_path = '/projects/%(project_id)s/merge_requests/%(mr_iid)s/award_emoji'
_obj_cls = ProjectMergeRequestAwardEmoji
_from_parent_attrs = {'project_id': 'project_id', 'mr_iid': 'iid'}
_create_attrs = (('name', ), tuple())
class ProjectMergeRequestDiff(RESTObject):
pass
class ProjectMergeRequestDiffManager(RetrieveMixin, RESTManager):
_path = '/projects/%(project_id)s/merge_requests/%(mr_iid)s/versions'
_obj_cls = ProjectMergeRequestDiff
_from_parent_attrs = {'project_id': 'project_id', 'mr_iid': 'iid'}
class ProjectMergeRequestNoteAwardEmoji(ObjectDeleteMixin, RESTObject):
pass
class ProjectMergeRequestNoteAwardEmojiManager(NoUpdateMixin, RESTManager):
_path = ('/projects/%(project_id)s/merge_requests/%(mr_iid)s'
'/notes/%(note_id)s/award_emoji')
_obj_cls = ProjectMergeRequestNoteAwardEmoji
_from_parent_attrs = {'project_id': 'project_id',
'mr_iid': 'mr_iid',
'note_id': 'id'}
_create_attrs = (('name', ), tuple())
class ProjectMergeRequestNote(SaveMixin, ObjectDeleteMixin, RESTObject):
_managers = (('awardemojis', 'ProjectMergeRequestNoteAwardEmojiManager'),)
class ProjectMergeRequestNoteManager(CRUDMixin, RESTManager):
_path = '/projects/%(project_id)s/merge_requests/%(mr_iid)s/notes'
_obj_cls = ProjectMergeRequestNote
_from_parent_attrs = {'project_id': 'project_id', 'mr_iid': 'iid'}
_create_attrs = (('body', ), tuple())
_update_attrs = (('body', ), tuple())
class ProjectMergeRequestDiscussionNote(SaveMixin, ObjectDeleteMixin,
RESTObject):
pass
class ProjectMergeRequestDiscussionNoteManager(GetMixin, CreateMixin,
UpdateMixin, DeleteMixin,
RESTManager):
_path = ('/projects/%(project_id)s/merge_requests/%(mr_iid)s/'
'discussions/%(discussion_id)s/notes')
_obj_cls = ProjectMergeRequestDiscussionNote
_from_parent_attrs = {'project_id': 'project_id',
'mr_iid': 'mr_iid',
'discussion_id': 'id'}
_create_attrs = (('body',), ('created_at',))
_update_attrs = (('body',), tuple())
class ProjectMergeRequestDiscussion(SaveMixin, RESTObject):
_managers = (('notes', 'ProjectMergeRequestDiscussionNoteManager'),)
class ProjectMergeRequestDiscussionManager(RetrieveMixin, CreateMixin,
UpdateMixin, RESTManager):
_path = '/projects/%(project_id)s/merge_requests/%(mr_iid)s/discussions'
_obj_cls = ProjectMergeRequestDiscussion
_from_parent_attrs = {'project_id': 'project_id', 'mr_iid': 'iid'}
_create_attrs = (('body',), ('created_at', 'position'))
_update_attrs = (('resolved',), tuple())
class ProjectMergeRequest(SubscribableMixin, TodoMixin, TimeTrackingMixin,
ParticipantsMixin, SaveMixin, ObjectDeleteMixin,
RESTObject):
_id_attr = 'iid'
_managers = (
('approvals', 'ProjectMergeRequestApprovalManager'),
('awardemojis', 'ProjectMergeRequestAwardEmojiManager'),
('diffs', 'ProjectMergeRequestDiffManager'),
('discussions', 'ProjectMergeRequestDiscussionManager'),
('notes', 'ProjectMergeRequestNoteManager'),
)
@cli.register_custom_action('ProjectMergeRequest')
@exc.on_http_error(exc.GitlabMROnBuildSuccessError)
def cancel_merge_when_pipeline_succeeds(self, **kwargs):
"""Cancel merge when the pipeline succeeds.
Args:
**kwargs: Extra options to send to the server (e.g. sudo)
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabMROnBuildSuccessError: If the server could not handle the
request
"""
path = ('%s/%s/cancel_merge_when_pipeline_succeeds' %
(self.manager.path, self.get_id()))
server_data = self.manager.gitlab.http_put(path, **kwargs)
self._update_attrs(server_data)
@cli.register_custom_action('ProjectMergeRequest')
@exc.on_http_error(exc.GitlabListError)
def closes_issues(self, **kwargs):
"""List issues that will close on merge."
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: List of issues
"""
path = '%s/%s/closes_issues' % (self.manager.path, self.get_id())
data_list = self.manager.gitlab.http_list(path, as_list=False,
**kwargs)
manager = ProjectIssueManager(self.manager.gitlab,
parent=self.manager._parent)
return RESTObjectList(manager, ProjectIssue, data_list)
@cli.register_custom_action('ProjectMergeRequest')
@exc.on_http_error(exc.GitlabListError)
def commits(self, **kwargs):
"""List the merge request commits.
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 commits
"""
path = '%s/%s/commits' % (self.manager.path, self.get_id())
data_list = self.manager.gitlab.http_list(path, as_list=False,
**kwargs)
manager = ProjectCommitManager(self.manager.gitlab,
parent=self.manager._parent)
return RESTObjectList(manager, ProjectCommit, data_list)
@cli.register_custom_action('ProjectMergeRequest')
@exc.on_http_error(exc.GitlabListError)
def changes(self, **kwargs):
"""List the merge request changes.
Args:
**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: List of changes
"""
path = '%s/%s/changes' % (self.manager.path, self.get_id())
return self.manager.gitlab.http_get(path, **kwargs)
@cli.register_custom_action('ProjectMergeRequest')
@exc.on_http_error(exc.GitlabListError)
def pipelines(self, **kwargs):
"""List the merge request pipelines.
Args:
**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: List of changes
"""
path = '%s/%s/pipelines' % (self.manager.path, self.get_id())
return self.manager.gitlab.http_get(path, **kwargs)
@cli.register_custom_action('ProjectMergeRequest', tuple(),
('merge_commit_message',
'should_remove_source_branch',
'merge_when_pipeline_succeeds'))
@exc.on_http_error(exc.GitlabMRClosedError)
def merge(self, merge_commit_message=None,
should_remove_source_branch=False,
merge_when_pipeline_succeeds=False,
**kwargs):
"""Accept the merge request.
Args:
merge_commit_message (bool): Commit message
should_remove_source_branch (bool): If True, removes the source
branch
merge_when_pipeline_succeeds (bool): Wait for the build to succeed,
then merge
**kwargs: Extra options to send to the server (e.g. sudo)
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabMRClosedError: If the merge failed
"""
path = '%s/%s/merge' % (self.manager.path, self.get_id())
data = {}
if merge_commit_message:
data['merge_commit_message'] = merge_commit_message
if should_remove_source_branch:
data['should_remove_source_branch'] = True
if merge_when_pipeline_succeeds:
data['merge_when_pipeline_succeeds'] = True
server_data = self.manager.gitlab.http_put(path, post_data=data,
**kwargs)
self._update_attrs(server_data)
class ProjectMergeRequestManager(CRUDMixin, RESTManager):
_path = '/projects/%(project_id)s/merge_requests'
_obj_cls = ProjectMergeRequest
_from_parent_attrs = {'project_id': 'id'}
_create_attrs = (
('source_branch', 'target_branch', 'title'),
('assignee_id', 'description', 'target_project_id', 'labels',
'milestone_id', 'remove_source_branch', 'allow_maintainer_to_push',
'squash')
)
_update_attrs = (
tuple(),
('target_branch', 'assignee_id', 'title', 'description', 'state_event',
'labels', 'milestone_id', 'remove_source_branch', 'discussion_locked',
'allow_maintainer_to_push', 'squash'))
_list_filters = ('state', 'order_by', 'sort', 'milestone', 'view',
'labels', 'created_after', 'created_before',
'updated_after', 'updated_before', 'scope', 'author_id',
'assignee_id', 'my_reaction_emoji', 'source_branch',
'target_branch', 'search')
_types = {'labels': types.ListAttribute}
class ProjectMilestone(SaveMixin, ObjectDeleteMixin, RESTObject):
_short_print_attr = 'title'
@cli.register_custom_action('ProjectMilestone')
@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 = ProjectIssueManager(self.manager.gitlab,
parent=self.manager._parent)
# FIXME(gpocentek): the computed manager path is not correct
return RESTObjectList(manager, ProjectIssue, data_list)
@cli.register_custom_action('ProjectMilestone')
@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 = ProjectMergeRequestManager(self.manager.gitlab,
parent=self.manager._parent)
# FIXME(gpocentek): the computed manager path is not correct
return RESTObjectList(manager, ProjectMergeRequest, data_list)
class ProjectMilestoneManager(CRUDMixin, RESTManager):
_path = '/projects/%(project_id)s/milestones'
_obj_cls = ProjectMilestone
_from_parent_attrs = {'project_id': 'id'}
_create_attrs = (('title', ), ('description', 'due_date', 'start_date',
'state_event'))
_update_attrs = (tuple(), ('title', 'description', 'due_date',
'start_date', 'state_event'))
_list_filters = ('iids', 'state', 'search')
class ProjectLabel(SubscribableMixin, SaveMixin, ObjectDeleteMixin,
RESTObject):
_id_attr = 'name'
# Update without ID, but we need an ID to get from list.
@exc.on_http_error(exc.GitlabUpdateError)
def save(self, **kwargs):
"""Saves 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)
Raises:
GitlabAuthenticationError: If authentication is not correct.
GitlabUpdateError: If the server cannot perform the request.
"""
updated_data = self._get_updated_data()
# call the manager
server_data = self.manager.update(None, updated_data, **kwargs)
self._update_attrs(server_data)
class ProjectLabelManager(ListMixin, CreateMixin, UpdateMixin, DeleteMixin,
RESTManager):
_path = '/projects/%(project_id)s/labels'
_obj_cls = ProjectLabel
_from_parent_attrs = {'project_id': 'id'}
_create_attrs = (('name', 'color'), ('description', 'priority'))
_update_attrs = (('name', ),
('new_name', 'color', 'description', 'priority'))
# Delete without ID.
@exc.on_http_error(exc.GitlabDeleteError)
def delete(self, name, **kwargs):
"""Delete a Label on the server.
Args:
name: The name of the label
**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.gitlab.http_delete(self.path, query_data={'name': name}, **kwargs)
class ProjectFile(SaveMixin, ObjectDeleteMixin, RESTObject):
_id_attr = 'file_path'
_short_print_attr = 'file_path'
def decode(self):
"""Returns the decoded content of the file.
Returns:
(str): the decoded content.
"""
return base64.b64decode(self.content)
def save(self, branch, commit_message, **kwargs):
"""Save the changes made to the file to the server.
The object is updated to match what the server returns.
Args:
branch (str): Branch in which the file will be updated
commit_message (str): Message to send with the commit
**kwargs: Extra options to send to the server (e.g. sudo)
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabUpdateError: If the server cannot perform the request
"""
self.branch = branch
self.commit_message = commit_message
self.file_path = self.file_path.replace('/', '%2F')
super(ProjectFile, self).save(**kwargs)
def delete(self, branch, commit_message, **kwargs):
"""Delete the file from the server.
Args:
branch (str): Branch from which the file will be removed
commit_message (str): Commit message for the deletion
**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
"""
file_path = self.get_id().replace('/', '%2F')
self.manager.delete(file_path, branch, commit_message, **kwargs)
class ProjectFileManager(GetMixin, CreateMixin, UpdateMixin, DeleteMixin,
RESTManager):
_path = '/projects/%(project_id)s/repository/files'
_obj_cls = ProjectFile
_from_parent_attrs = {'project_id': 'id'}
_create_attrs = (('file_path', 'branch', 'content', 'commit_message'),
('encoding', 'author_email', 'author_name'))
_update_attrs = (('file_path', 'branch', 'content', 'commit_message'),
('encoding', 'author_email', 'author_name'))
@cli.register_custom_action('ProjectFileManager', ('file_path', 'ref'))
def get(self, file_path, ref, **kwargs):
"""Retrieve a single file.
Args:
file_path (str): Path of the file to retrieve
ref (str): Name of the branch, tag or commit
**kwargs: Extra options to send to the server (e.g. sudo)
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabGetError: If the file could not be retrieved
Returns:
object: The generated RESTObject
"""
file_path = file_path.replace('/', '%2F')
return GetMixin.get(self, file_path, ref=ref, **kwargs)
@cli.register_custom_action('ProjectFileManager',
('file_path', 'branch', 'content',
'commit_message'),
('encoding', 'author_email', 'author_name'))
@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)
new_data = data.copy()
file_path = new_data.pop('file_path').replace('/', '%2F')
path = '%s/%s' % (self.path, file_path)
server_data = self.gitlab.http_post(path, post_data=new_data, **kwargs)
return self._obj_cls(self, server_data)
@exc.on_http_error(exc.GitlabUpdateError)
def update(self, file_path, 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
"""
data = new_data.copy()
file_path = file_path.replace('/', '%2F')
data['file_path'] = file_path
path = '%s/%s' % (self.path, file_path)
self._check_missing_update_attrs(data)
return self.gitlab.http_put(path, post_data=data, **kwargs)
@cli.register_custom_action('ProjectFileManager', ('file_path', 'branch',
'commit_message'))
@exc.on_http_error(exc.GitlabDeleteError)
def delete(self, file_path, branch, commit_message, **kwargs):
"""Delete a file on the server.
Args:
file_path (str): Path of the file to remove
branch (str): Branch from which the file will be removed
commit_message (str): Commit message for the deletion
**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
"""
path = '%s/%s' % (self.path, file_path.replace('/', '%2F'))
data = {'branch': branch, 'commit_message': commit_message}
self.gitlab.http_delete(path, query_data=data, **kwargs)
@cli.register_custom_action('ProjectFileManager', ('file_path', 'ref'))
@exc.on_http_error(exc.GitlabGetError)
def raw(self, file_path, ref, streamed=False, action=None, chunk_size=1024,
**kwargs):
"""Return the content of a file for a commit.
Args:
ref (str): ID of the commit
filepath (str): Path of the file to return
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 file could not be retrieved
Returns:
str: The file content
"""
file_path = file_path.replace('/', '%2F').replace('.', '%2E')
path = '%s/%s/raw' % (self.path, file_path)
query_data = {'ref': ref}
result = self.gitlab.http_get(path, query_data=query_data,
streamed=streamed, **kwargs)
return utils.response_content(result, streamed, action, chunk_size)
class ProjectPipelineJob(RESTObject):
pass
class ProjectPipelineJobManager(ListMixin, RESTManager):
_path = '/projects/%(project_id)s/pipelines/%(pipeline_id)s/jobs'
_obj_cls = ProjectPipelineJob
_from_parent_attrs = {'project_id': 'project_id',
'pipeline_id': 'id'}
_list_filters = ('scope',)
class ProjectPipeline(RESTObject, RefreshMixin):
_managers = (('jobs', 'ProjectPipelineJobManager'), )
@cli.register_custom_action('ProjectPipeline')
@exc.on_http_error(exc.GitlabPipelineCancelError)
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
GitlabPipelineCancelError: If the request failed
"""
path = '%s/%s/cancel' % (self.manager.path, self.get_id())
self.manager.gitlab.http_post(path)
@cli.register_custom_action('ProjectPipeline')
@exc.on_http_error(exc.GitlabPipelineRetryError)
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
GitlabPipelineRetryError: If the request failed
"""
path = '%s/%s/retry' % (self.manager.path, self.get_id())
self.manager.gitlab.http_post(path)
class ProjectPipelineManager(RetrieveMixin, CreateMixin, RESTManager):
_path = '/projects/%(project_id)s/pipelines'
_obj_cls = ProjectPipeline
_from_parent_attrs = {'project_id': 'id'}
_list_filters = ('scope', 'status', 'ref', 'sha', 'yaml_errors', 'name',
'username', 'order_by', 'sort')
_create_attrs = (('ref', ), tuple())
def create(self, data, **kwargs):
"""Creates 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)
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabCreateError: If the server cannot perform the request
Returns:
RESTObject: A new instance of the managed object class build with
the data sent by the server
"""
path = self.path[:-1] # drop the 's'
return CreateMixin.create(self, data, path=path, **kwargs)
class ProjectPipelineScheduleVariable(SaveMixin, ObjectDeleteMixin,
RESTObject):
_id_attr = 'key'
class ProjectPipelineScheduleVariableManager(CreateMixin, UpdateMixin,
DeleteMixin, RESTManager):
_path = ('/projects/%(project_id)s/pipeline_schedules/'
'%(pipeline_schedule_id)s/variables')
_obj_cls = ProjectPipelineScheduleVariable
_from_parent_attrs = {'project_id': 'project_id',
'pipeline_schedule_id': 'id'}
_create_attrs = (('key', 'value'), tuple())
_update_attrs = (('key', 'value'), tuple())
class ProjectPipelineSchedule(SaveMixin, ObjectDeleteMixin, RESTObject):
_managers = (('variables', 'ProjectPipelineScheduleVariableManager'),)
@cli.register_custom_action('ProjectPipelineSchedule')
@exc.on_http_error(exc.GitlabOwnershipError)
def take_ownership(self, **kwargs):
"""Update the owner of a pipeline schedule.
Args:
**kwargs: Extra options to send to the server (e.g. sudo)
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabOwnershipError: If the request failed
"""
path = '%s/%s/take_ownership' % (self.manager.path, self.get_id())
server_data = self.manager.gitlab.http_post(path, **kwargs)
self._update_attrs(server_data)
class ProjectPipelineScheduleManager(CRUDMixin, RESTManager):
_path = '/projects/%(project_id)s/pipeline_schedules'
_obj_cls = ProjectPipelineSchedule
_from_parent_attrs = {'project_id': 'id'}
_create_attrs = (('description', 'ref', 'cron'),
('cron_timezone', 'active'))
_update_attrs = (tuple(),
('description', 'ref', 'cron', 'cron_timezone', 'active'))
class ProjectPushRules(SaveMixin, ObjectDeleteMixin, RESTObject):
_id_attr = None
class ProjectPushRulesManager(GetWithoutIdMixin, CreateMixin, UpdateMixin,
DeleteMixin, RESTManager):
_path = '/projects/%(project_id)s/push_rule'
_obj_cls = ProjectPushRules
_from_parent_attrs = {'project_id': 'id'}
_create_attrs = (tuple(),
('deny_delete_tag', 'member_check',
'prevent_secrets', 'commit_message_regex',
'branch_name_regex', 'author_email_regex',
'file_name_regex', 'max_file_size'))
_update_attrs = (tuple(),
('deny_delete_tag', 'member_check',
'prevent_secrets', 'commit_message_regex',
'branch_name_regex', 'author_email_regex',
'file_name_regex', 'max_file_size'))
class ProjectSnippetNoteAwardEmoji(ObjectDeleteMixin, RESTObject):
pass
class ProjectSnippetNoteAwardEmojiManager(NoUpdateMixin, RESTManager):
_path = ('/projects/%(project_id)s/snippets/%(snippet_id)s'
'/notes/%(note_id)s/award_emoji')
_obj_cls = ProjectSnippetNoteAwardEmoji
_from_parent_attrs = {'project_id': 'project_id',
'snippet_id': 'snippet_id',
'note_id': 'id'}
_create_attrs = (('name', ), tuple())
class ProjectSnippetNote(SaveMixin, ObjectDeleteMixin, RESTObject):
_managers = (('awardemojis', 'ProjectSnippetNoteAwardEmojiManager'),)
class ProjectSnippetNoteManager(CRUDMixin, RESTManager):
_path = '/projects/%(project_id)s/snippets/%(snippet_id)s/notes'
_obj_cls = ProjectSnippetNote
_from_parent_attrs = {'project_id': 'project_id',
'snippet_id': 'id'}
_create_attrs = (('body', ), tuple())
_update_attrs = (('body', ), tuple())
class ProjectSnippetAwardEmoji(ObjectDeleteMixin, RESTObject):
pass
class ProjectSnippetAwardEmojiManager(NoUpdateMixin, RESTManager):
_path = '/projects/%(project_id)s/snippets/%(snippet_id)s/award_emoji'
_obj_cls = ProjectSnippetAwardEmoji
_from_parent_attrs = {'project_id': 'project_id', 'snippet_id': 'id'}
_create_attrs = (('name', ), tuple())
class ProjectSnippetDiscussionNote(SaveMixin, ObjectDeleteMixin, RESTObject):
pass
class ProjectSnippetDiscussionNoteManager(GetMixin, CreateMixin, UpdateMixin,
DeleteMixin, RESTManager):
_path = ('/projects/%(project_id)s/snippets/%(snippet_id)s/'
'discussions/%(discussion_id)s/notes')
_obj_cls = ProjectSnippetDiscussionNote
_from_parent_attrs = {'project_id': 'project_id',
'snippet_id': 'snippet_id',
'discussion_id': 'id'}
_create_attrs = (('body',), ('created_at',))
_update_attrs = (('body',), tuple())
class ProjectSnippetDiscussion(RESTObject):
_managers = (('notes', 'ProjectSnippetDiscussionNoteManager'),)
class ProjectSnippetDiscussionManager(RetrieveMixin, CreateMixin, RESTManager):
_path = '/projects/%(project_id)s/snippets/%(snippet_id)s/discussions'
_obj_cls = ProjectSnippetDiscussion
_from_parent_attrs = {'project_id': 'project_id', 'snippet_id': 'id'}
_create_attrs = (('body',), ('created_at',))
class ProjectSnippet(UserAgentDetailMixin, SaveMixin, ObjectDeleteMixin,
RESTObject):
_url = '/projects/%(project_id)s/snippets'
_short_print_attr = 'title'
_managers = (
('awardemojis', 'ProjectSnippetAwardEmojiManager'),
('discussions', 'ProjectSnippetDiscussionManager'),
('notes', 'ProjectSnippetNoteManager'),
)
@cli.register_custom_action('ProjectSnippet')
@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 = "%s/%s/raw" % (self.manager.path, self.get_id())
result = self.manager.gitlab.http_get(path, streamed=streamed,
**kwargs)
return utils.response_content(result, streamed, action, chunk_size)
class ProjectSnippetManager(CRUDMixin, RESTManager):
_path = '/projects/%(project_id)s/snippets'
_obj_cls = ProjectSnippet
_from_parent_attrs = {'project_id': 'id'}
_create_attrs = (('title', 'file_name', 'code'),
('lifetime', 'visibility'))
_update_attrs = (tuple(), ('title', 'file_name', 'code', 'visibility'))
class ProjectTrigger(SaveMixin, ObjectDeleteMixin, RESTObject):
@cli.register_custom_action('ProjectTrigger')
@exc.on_http_error(exc.GitlabOwnershipError)
def take_ownership(self, **kwargs):
"""Update the owner of a trigger.
Args:
**kwargs: Extra options to send to the server (e.g. sudo)
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabOwnershipError: If the request failed
"""
path = '%s/%s/take_ownership' % (self.manager.path, self.get_id())
server_data = self.manager.gitlab.http_post(path, **kwargs)
self._update_attrs(server_data)
class ProjectTriggerManager(CRUDMixin, RESTManager):
_path = '/projects/%(project_id)s/triggers'
_obj_cls = ProjectTrigger
_from_parent_attrs = {'project_id': 'id'}
_create_attrs = (('description', ), tuple())
_update_attrs = (('description', ), tuple())
class ProjectUser(RESTObject):
pass
class ProjectUserManager(ListMixin, RESTManager):
_path = '/projects/%(project_id)s/users'
_obj_cls = ProjectUser
_from_parent_attrs = {'project_id': 'id'}
_list_filters = ('search',)
class ProjectVariable(SaveMixin, ObjectDeleteMixin, RESTObject):
_id_attr = 'key'
class ProjectVariableManager(CRUDMixin, RESTManager):
_path = '/projects/%(project_id)s/variables'
_obj_cls = ProjectVariable
_from_parent_attrs = {'project_id': 'id'}
_create_attrs = (('key', 'value'), tuple())
_update_attrs = (('key', 'value'), tuple())
class ProjectService(SaveMixin, ObjectDeleteMixin, RESTObject):
pass
class ProjectServiceManager(GetMixin, UpdateMixin, DeleteMixin, RESTManager):
_path = '/projects/%(project_id)s/services'
_from_parent_attrs = {'project_id': 'id'}
_obj_cls = ProjectService
_service_attrs = {
'asana': (('api_key', ), ('restrict_to_branch', )),
'assembla': (('token', ), ('subdomain', )),
'bamboo': (('bamboo_url', 'build_key', 'username', 'password'),
tuple()),
'buildkite': (('token', 'project_url'), ('enable_ssl_verification', )),
'campfire': (('token', ), ('subdomain', 'room')),
'custom-issue-tracker': (('new_issue_url', 'issues_url',
'project_url'),
('description', 'title')),
'drone-ci': (('token', 'drone_url'), ('enable_ssl_verification', )),
'emails-on-push': (('recipients', ), ('disable_diffs',
'send_from_committer_email')),
'builds-email': (('recipients', ), ('add_pusher',
'notify_only_broken_builds')),
'pipelines-email': (('recipients', ), ('add_pusher',
'notify_only_broken_builds')),
'external-wiki': (('external_wiki_url', ), tuple()),
'flowdock': (('token', ), tuple()),
'gemnasium': (('api_key', 'token', ), tuple()),
'hipchat': (('token', ), ('color', 'notify', 'room', 'api_version',
'server')),
'irker': (('recipients', ), ('default_irc_uri', 'server_port',
'server_host', 'colorize_messages')),
'jira': (('url', 'project_key'),
('new_issue_url', 'project_url', 'issues_url', 'api_url',
'description', 'username', 'password',
'jira_issue_transition_id')),
'mattermost': (('webhook',), ('username', 'channel')),
'pivotaltracker': (('token', ), tuple()),
'pushover': (('api_key', 'user_key', 'priority'), ('device', 'sound')),
'redmine': (('new_issue_url', 'project_url', 'issues_url'),
('description', )),
'slack': (('webhook', ), ('username', 'channel')),
'teamcity': (('teamcity_url', 'build_type', 'username', 'password'),
tuple())
}
def get(self, id, **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
"""
obj = super(ProjectServiceManager, self).get(id, **kwargs)
obj.id = id
return obj
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
"""
super(ProjectServiceManager, self).update(id, new_data, **kwargs)
self.id = id
@cli.register_custom_action('ProjectServiceManager')
def available(self, **kwargs):
"""List the services known by python-gitlab.
Returns:
list (str): The list of service code names.
"""
return list(self._service_attrs.keys())
class ProjectAccessRequest(AccessRequestMixin, ObjectDeleteMixin, RESTObject):
pass
class ProjectAccessRequestManager(ListMixin, CreateMixin, DeleteMixin,
RESTManager):
_path = '/projects/%(project_id)s/access_requests'
_obj_cls = ProjectAccessRequest
_from_parent_attrs = {'project_id': 'id'}
class ProjectApproval(SaveMixin, RESTObject):
_id_attr = None
class ProjectApprovalManager(GetWithoutIdMixin, UpdateMixin, RESTManager):
_path = '/projects/%(project_id)s/approvals'
_obj_cls = ProjectApproval
_from_parent_attrs = {'project_id': 'id'}
_update_attrs = (tuple(),
('approvals_before_merge', 'reset_approvals_on_push',
'disable_overriding_approvers_per_merge_request'))
_update_uses_post = True
@exc.on_http_error(exc.GitlabUpdateError)
def set_approvers(self, approver_ids=[], approver_group_ids=[], **kwargs):
"""Change project-level allowed approvers and approver groups.
Args:
approver_ids (list): User IDs that can approve MRs
approver_group_ids (list): Group IDs whose members can approve MRs
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabUpdateError: If the server failed to perform the request
"""
path = '/projects/%s/approvers' % self._parent.get_id()
data = {'approver_ids': approver_ids,
'approver_group_ids': approver_group_ids}
self.gitlab.http_put(path, post_data=data, **kwargs)
class ProjectDeployment(RESTObject):
pass
class ProjectDeploymentManager(RetrieveMixin, RESTManager):
_path = '/projects/%(project_id)s/deployments'
_obj_cls = ProjectDeployment
_from_parent_attrs = {'project_id': 'id'}
_list_filters = ('order_by', 'sort')
class ProjectProtectedBranch(ObjectDeleteMixin, RESTObject):
_id_attr = 'name'
class ProjectProtectedBranchManager(NoUpdateMixin, RESTManager):
_path = '/projects/%(project_id)s/protected_branches'
_obj_cls = ProjectProtectedBranch
_from_parent_attrs = {'project_id': 'id'}
_create_attrs = (('name', ), ('push_access_level', 'merge_access_level'))
class ProjectRunner(ObjectDeleteMixin, RESTObject):
pass
class ProjectRunnerManager(NoUpdateMixin, RESTManager):
_path = '/projects/%(project_id)s/runners'
_obj_cls = ProjectRunner
_from_parent_attrs = {'project_id': 'id'}
_create_attrs = (('runner_id', ), tuple())
class ProjectWiki(SaveMixin, ObjectDeleteMixin, RESTObject):
_id_attr = 'slug'
_short_print_attr = 'slug'
class ProjectWikiManager(CRUDMixin, RESTManager):
_path = '/projects/%(project_id)s/wikis'
_obj_cls = ProjectWiki
_from_parent_attrs = {'project_id': 'id'}
_create_attrs = (('title', 'content'), ('format', ))
_update_attrs = (tuple(), ('title', 'content', 'format'))
_list_filters = ('with_content', )
class ProjectExport(RefreshMixin, RESTObject):
_id_attr = None
@cli.register_custom_action('ProjectExport')
@exc.on_http_error(exc.GitlabGetError)
def download(self, streamed=False, action=None, chunk_size=1024, **kwargs):
"""Download the archive of a project export.
Args:
streamed (bool): If True the data will be processed by chunks of
`chunk_size` and each chunk is passed to `action` for
reatment
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 server failed to perform the request
Returns:
str: The blob content if streamed is False, None otherwise
"""
path = '/projects/%d/export/download' % self.project_id
result = self.manager.gitlab.http_get(path, streamed=streamed,
**kwargs)
return utils.response_content(result, streamed, action, chunk_size)
class ProjectExportManager(GetWithoutIdMixin, CreateMixin, RESTManager):
_path = '/projects/%(project_id)s/export'
_obj_cls = ProjectExport
_from_parent_attrs = {'project_id': 'id'}
_create_attrs = (tuple(), ('description',))
class ProjectImport(RefreshMixin, RESTObject):
_id_attr = None
class ProjectImportManager(GetWithoutIdMixin, RESTManager):
_path = '/projects/%(project_id)s/import'
_obj_cls = ProjectImport
_from_parent_attrs = {'project_id': 'id'}
class Project(SaveMixin, ObjectDeleteMixin, RESTObject):
_short_print_attr = 'path'
_managers = (
('accessrequests', 'ProjectAccessRequestManager'),
('approvals', 'ProjectApprovalManager'),
('badges', 'ProjectBadgeManager'),
('boards', 'ProjectBoardManager'),
('branches', 'ProjectBranchManager'),
('jobs', 'ProjectJobManager'),
('commits', 'ProjectCommitManager'),
('customattributes', 'ProjectCustomAttributeManager'),
('deployments', 'ProjectDeploymentManager'),
('environments', 'ProjectEnvironmentManager'),
('events', 'ProjectEventManager'),
('exports', 'ProjectExportManager'),
('files', 'ProjectFileManager'),
('forks', 'ProjectForkManager'),
('hooks', 'ProjectHookManager'),
('keys', 'ProjectKeyManager'),
('imports', 'ProjectImportManager'),
('issues', 'ProjectIssueManager'),
('labels', 'ProjectLabelManager'),
('members', 'ProjectMemberManager'),
('mergerequests', 'ProjectMergeRequestManager'),
('milestones', 'ProjectMilestoneManager'),
('notes', 'ProjectNoteManager'),
('notificationsettings', 'ProjectNotificationSettingsManager'),
('pagesdomains', 'ProjectPagesDomainManager'),
('pipelines', 'ProjectPipelineManager'),
('protectedbranches', 'ProjectProtectedBranchManager'),
('protectedtags', 'ProjectProtectedTagManager'),
('pipelineschedules', 'ProjectPipelineScheduleManager'),
('pushrules', 'ProjectPushRulesManager'),
('runners', 'ProjectRunnerManager'),
('services', 'ProjectServiceManager'),
('snippets', 'ProjectSnippetManager'),
('tags', 'ProjectTagManager'),
('users', 'ProjectUserManager'),
('triggers', 'ProjectTriggerManager'),
('variables', 'ProjectVariableManager'),
('wikis', 'ProjectWikiManager'),
)
@cli.register_custom_action('Project', tuple(), ('path', 'ref'))
@exc.on_http_error(exc.GitlabGetError)
def repository_tree(self, path='', ref='', recursive=False, **kwargs):
"""Return a list of files in the repository.
Args:
path (str): Path of the top folder (/ by default)
ref (str): Reference to a commit or branch
recursive (bool): Whether to get the tree recursively
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
GitlabGetError: If the server failed to perform the request
Returns:
list: The representation of the tree
"""
gl_path = '/projects/%s/repository/tree' % self.get_id()
query_data = {'recursive': recursive}
if path:
query_data['path'] = path
if ref:
query_data['ref'] = ref
return self.manager.gitlab.http_list(gl_path, query_data=query_data,
**kwargs)
@cli.register_custom_action('Project', ('sha', ))
@exc.on_http_error(exc.GitlabGetError)
def repository_blob(self, sha, **kwargs):
"""Return a file by blob SHA.
Args:
sha(str): ID of the blob
**kwargs: Extra options to send to the server (e.g. sudo)
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabGetError: If the server failed to perform the request
Returns:
dict: The blob content and metadata
"""
path = '/projects/%s/repository/blobs/%s' % (self.get_id(), sha)
return self.manager.gitlab.http_get(path, **kwargs)
@cli.register_custom_action('Project', ('sha', ))
@exc.on_http_error(exc.GitlabGetError)
def repository_raw_blob(self, sha, streamed=False, action=None,
chunk_size=1024, **kwargs):
"""Return the raw file contents for a blob.
Args:
sha(str): ID of the blob
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 server failed to perform the request
Returns:
str: The blob content if streamed is False, None otherwise
"""
path = '/projects/%s/repository/blobs/%s/raw' % (self.get_id(), sha)
result = self.manager.gitlab.http_get(path, streamed=streamed,
**kwargs)
return utils.response_content(result, streamed, action, chunk_size)
@cli.register_custom_action('Project', ('from_', 'to'))
@exc.on_http_error(exc.GitlabGetError)
def repository_compare(self, from_, to, **kwargs):
"""Return a diff between two branches/commits.
Args:
from_(str): Source branch/SHA
to(str): Destination branch/SHA
**kwargs: Extra options to send to the server (e.g. sudo)
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabGetError: If the server failed to perform the request
Returns:
str: The diff
"""
path = '/projects/%s/repository/compare' % self.get_id()
query_data = {'from': from_, 'to': to}
return self.manager.gitlab.http_get(path, query_data=query_data,
**kwargs)
@cli.register_custom_action('Project')
@exc.on_http_error(exc.GitlabGetError)
def repository_contributors(self, **kwargs):
"""Return a list of contributors for the project.
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
GitlabGetError: If the server failed to perform the request
Returns:
list: The contributors
"""
path = '/projects/%s/repository/contributors' % self.get_id()
return self.manager.gitlab.http_list(path, **kwargs)
@cli.register_custom_action('Project', tuple(), ('sha', ))
@exc.on_http_error(exc.GitlabListError)
def repository_archive(self, sha=None, streamed=False, action=None,
chunk_size=1024, **kwargs):
"""Return a tarball of the repository.
Args:
sha (str): ID of the commit (default branch by default)
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
GitlabListError: If the server failed to perform the request
Returns:
str: The binary data of the archive
"""
path = '/projects/%s/repository/archive' % self.get_id()
query_data = {}
if sha:
query_data['sha'] = sha
result = self.manager.gitlab.http_get(path, query_data=query_data,
streamed=streamed, **kwargs)
return utils.response_content(result, streamed, action, chunk_size)
@cli.register_custom_action('Project', ('forked_from_id', ))
@exc.on_http_error(exc.GitlabCreateError)
def create_fork_relation(self, forked_from_id, **kwargs):
"""Create a forked from/to relation between existing projects.
Args:
forked_from_id (int): The ID of the project that was forked from
**kwargs: Extra options to send to the server (e.g. sudo)
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabCreateError: If the relation could not be created
"""
path = '/projects/%s/fork/%s' % (self.get_id(), forked_from_id)
self.manager.gitlab.http_post(path, **kwargs)
@cli.register_custom_action('Project')
@exc.on_http_error(exc.GitlabDeleteError)
def delete_fork_relation(self, **kwargs):
"""Delete a forked relation between existing projects.
Args:
**kwargs: Extra options to send to the server (e.g. sudo)
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabDeleteError: If the server failed to perform the request
"""
path = '/projects/%s/fork' % self.get_id()
self.manager.gitlab.http_delete(path, **kwargs)
@cli.register_custom_action('Project')
@exc.on_http_error(exc.GitlabDeleteError)
def delete_merged_branches(self, **kwargs):
"""Delete merged branches.
Args:
**kwargs: Extra options to send to the server (e.g. sudo)
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabDeleteError: If the server failed to perform the request
"""
path = '/projects/%s/repository/merged_branches' % self.get_id()
self.manager.gitlab.http_delete(path, **kwargs)
@cli.register_custom_action('Project')
@exc.on_http_error(exc.GitlabGetError)
def languages(self, **kwargs):
"""Get languages used in the project with percentage value.
Args:
**kwargs: Extra options to send to the server (e.g. sudo)
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabGetError: If the server failed to perform the request
"""
path = '/projects/%s/languages' % self.get_id()
return self.manager.gitlab.http_get(path, **kwargs)
@cli.register_custom_action('Project')
@exc.on_http_error(exc.GitlabCreateError)
def star(self, **kwargs):
"""Star a project.
Args:
**kwargs: Extra options to send to the server (e.g. sudo)
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabCreateError: If the server failed to perform the request
"""
path = '/projects/%s/star' % self.get_id()
server_data = self.manager.gitlab.http_post(path, **kwargs)
self._update_attrs(server_data)
@cli.register_custom_action('Project')
@exc.on_http_error(exc.GitlabDeleteError)
def unstar(self, **kwargs):
"""Unstar a project.
Args:
**kwargs: Extra options to send to the server (e.g. sudo)
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabDeleteError: If the server failed to perform the request
"""
path = '/projects/%s/unstar' % self.get_id()
server_data = self.manager.gitlab.http_post(path, **kwargs)
self._update_attrs(server_data)
@cli.register_custom_action('Project')
@exc.on_http_error(exc.GitlabCreateError)
def archive(self, **kwargs):
"""Archive a project.
Args:
**kwargs: Extra options to send to the server (e.g. sudo)
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabCreateError: If the server failed to perform the request
"""
path = '/projects/%s/archive' % self.get_id()
server_data = self.manager.gitlab.http_post(path, **kwargs)
self._update_attrs(server_data)
@cli.register_custom_action('Project')
@exc.on_http_error(exc.GitlabDeleteError)
def unarchive(self, **kwargs):
"""Unarchive a project.
Args:
**kwargs: Extra options to send to the server (e.g. sudo)
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabDeleteError: If the server failed to perform the request
"""
path = '/projects/%s/unarchive' % self.get_id()
server_data = self.manager.gitlab.http_post(path, **kwargs)
self._update_attrs(server_data)
@cli.register_custom_action('Project', ('group_id', 'group_access'),
('expires_at', ))
@exc.on_http_error(exc.GitlabCreateError)
def share(self, group_id, group_access, expires_at=None, **kwargs):
"""Share the project with a group.
Args:
group_id (int): ID of the group.
group_access (int): Access level for the group.
**kwargs: Extra options to send to the server (e.g. sudo)
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabCreateError: If the server failed to perform the request
"""
path = '/projects/%s/share' % self.get_id()
data = {'group_id': group_id,
'group_access': group_access,
'expires_at': expires_at}
self.manager.gitlab.http_post(path, post_data=data, **kwargs)
@cli.register_custom_action('Project', ('group_id', ))
@exc.on_http_error(exc.GitlabDeleteError)
def unshare(self, group_id, **kwargs):
"""Delete a shared project link within a group.
Args:
group_id (int): ID of the group.
**kwargs: Extra options to send to the server (e.g. sudo)
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabDeleteError: If the server failed to perform the request
"""
path = '/projects/%s/share/%s' % (self.get_id(), group_id)
self.manager.gitlab.http_delete(path, **kwargs)
# variables not supported in CLI
@cli.register_custom_action('Project', ('ref', 'token'))
@exc.on_http_error(exc.GitlabCreateError)
def trigger_pipeline(self, ref, token, variables={}, **kwargs):
"""Trigger a CI build.
See https://gitlab.com/help/ci/triggers/README.md#trigger-a-build
Args:
ref (str): Commit to build; can be a branch name or a tag
token (str): The trigger token
variables (dict): Variables passed to the build script
**kwargs: Extra options to send to the server (e.g. sudo)
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabCreateError: If the server failed to perform the request
"""
path = '/projects/%s/trigger/pipeline' % self.get_id()
post_data = {'ref': ref, 'token': token, 'variables': variables}
attrs = self.manager.gitlab.http_post(
path, post_data=post_data, **kwargs)
return ProjectPipeline(self.pipelines, attrs)
@cli.register_custom_action('Project')
@exc.on_http_error(exc.GitlabHousekeepingError)
def housekeeping(self, **kwargs):
"""Start the housekeeping task.
Args:
**kwargs: Extra options to send to the server (e.g. sudo)
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabHousekeepingError: If the server failed to perform the
request
"""
path = '/projects/%s/housekeeping' % self.get_id()
self.manager.gitlab.http_post(path, **kwargs)
# see #56 - add file attachment features
@cli.register_custom_action('Project', ('filename', 'filepath'))
@exc.on_http_error(exc.GitlabUploadError)
def upload(self, filename, filedata=None, filepath=None, **kwargs):
"""Upload the specified file into the project.
.. note::
Either ``filedata`` or ``filepath`` *MUST* be specified.
Args:
filename (str): The name of the file being uploaded
filedata (bytes): The raw data of the file being uploaded
filepath (str): The path to a local file to upload (optional)
Raises:
GitlabConnectionError: If the server cannot be reached
GitlabUploadError: If the file upload fails
GitlabUploadError: If ``filedata`` and ``filepath`` are not
specified
GitlabUploadError: If both ``filedata`` and ``filepath`` are
specified
Returns:
dict: A ``dict`` with the keys:
* ``alt`` - The alternate text for the upload
* ``url`` - The direct url to the uploaded file
* ``markdown`` - Markdown for the uploaded file
"""
if filepath is None and filedata is None:
raise GitlabUploadError("No file contents or path specified")
if filedata is not None and filepath is not None:
raise GitlabUploadError("File contents and file path specified")
if filepath is not None:
with open(filepath, "rb") as f:
filedata = f.read()
url = ('/projects/%(id)s/uploads' % {
'id': self.id,
})
file_info = {
'file': (filename, filedata),
}
data = self.manager.gitlab.http_post(url, files=file_info)
return {
"alt": data['alt'],
"url": data['url'],
"markdown": data['markdown']
}
@cli.register_custom_action('Project', optional=('wiki',))
@exc.on_http_error(exc.GitlabGetError)
def snapshot(self, wiki=False, streamed=False, action=None,
chunk_size=1024, **kwargs):
"""Return a snapshot of the repository.
Args:
wiki (bool): If True return the wiki repository
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 uncompressed tar archive of the repository
"""
path = '/projects/%d/snapshot' % self.get_id()
result = self.manager.gitlab.http_get(path, streamed=streamed,
**kwargs)
return utils.response_content(result, streamed, action, chunk_size)
@cli.register_custom_action('Project', ('scope', 'search'))
@exc.on_http_error(exc.GitlabSearchError)
def search(self, scope, search, **kwargs):
"""Search the project resources matching the provided string.'
Args:
scope (str): Scope of the search
search (str): Search string
**kwargs: Extra options to send to the server (e.g. sudo)
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabSearchError: If the server failed to perform the request
Returns:
GitlabList: A list of dicts describing the resources found.
"""
data = {'scope': scope, 'search': search}
path = '/projects/%d/search' % self.get_id()
return self.manager.gitlab.http_list(path, query_data=data, **kwargs)
@cli.register_custom_action('Project')
@exc.on_http_error(exc.GitlabCreateError)
def mirror_pull(self, **kwargs):
"""Start the pull mirroring process for the project.
Args:
**kwargs: Extra options to send to the server (e.g. sudo)
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabCreateError: If the server failed to perform the request
"""
path = '/projects/%d/mirror/pull' % self.get_id()
self.manager.gitlab.http_post(path, **kwargs)
@cli.register_custom_action('Project', ('to_namespace', ))
@exc.on_http_error(exc.GitlabTransferProjectError)
def transfer_project(self, to_namespace, **kwargs):
"""Transfer a project to the given namespace ID
Args:
to_namespace (str): ID or path of the namespace to transfer the
project to
**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 = '/projects/%d/transfer' % (self.id,)
self.manager.gitlab.http_put(path,
post_data={"namespace": to_namespace},
**kwargs)
class ProjectManager(CRUDMixin, RESTManager):
_path = '/projects'
_obj_cls = Project
_create_attrs = (
tuple(),
('name', 'path', 'namespace_id', 'description', 'issues_enabled',
'merge_requests_enabled', 'jobs_enabled', 'wiki_enabled',
'snippets_enabled', 'resolve_outdated_diff_discussions',
'container_registry_enabled', 'shared_runners_enabled', 'visibility',
'import_url', 'public_jobs', 'only_allow_merge_if_pipeline_succeeds',
'only_allow_merge_if_all_discussions_are_resolved', 'merge_method',
'lfs_enabled', 'request_access_enabled', 'tag_list', 'avatar',
'printing_merge_request_link_enabled', 'ci_config_path')
)
_update_attrs = (
tuple(),
('name', 'path', 'default_branch', 'description', 'issues_enabled',
'merge_requests_enabled', 'jobs_enabled', 'wiki_enabled',
'snippets_enabled', 'resolve_outdated_diff_discussions',
'container_registry_enabled', 'shared_runners_enabled', 'visibility',
'import_url', 'public_jobs', 'only_allow_merge_if_pipeline_succeeds',
'only_allow_merge_if_all_discussions_are_resolved', 'merge_method',
'lfs_enabled', 'request_access_enabled', 'tag_list', 'avatar',
'ci_config_path')
)
_list_filters = ('search', 'owned', 'starred', 'archived', 'visibility',
'order_by', 'sort', 'simple', 'membership', 'statistics',
'with_issues_enabled', 'with_merge_requests_enabled',
'with_custom_attributes')
def import_project(self, file, path, namespace=None, overwrite=False,
override_params=None, **kwargs):
"""Import a project from an archive file.
Args:
file: Data or file object containing the project
path (str): Name and path for the new project
namespace (str): The ID or path of the namespace that the project
will be imported to
overwrite (bool): If True overwrite an existing project with the
same path
override_params (dict): Set the specific settings for the project
**kwargs: Extra options to send to the server (e.g. sudo)
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabListError: If the server failed to perform the request
Returns:
dict: A representation of the import status.
"""
files = {
'file': ('file.tar.gz', file)
}
data = {
'path': path,
'overwrite': overwrite
}
if override_params:
for k, v in override_params.items():
data['override_params[%s]' % k] = v
if namespace:
data['namespace'] = namespace
return self.gitlab.http_post('/projects/import', post_data=data,
files=files, **kwargs)
class RunnerJob(RESTObject):
pass
class RunnerJobManager(ListMixin, RESTManager):
_path = '/runners/%(runner_id)s/jobs'
_obj_cls = RunnerJob
_from_parent_attrs = {'runner_id': 'id'}
_list_filters = ('status',)
class Runner(SaveMixin, ObjectDeleteMixin, RESTObject):
_managers = (('jobs', 'RunnerJobManager'),)
class RunnerManager(CRUDMixin, RESTManager):
_path = '/runners'
_obj_cls = Runner
_list_filters = ('scope', )
_create_attrs = (('token',), ('description', 'info', 'active', 'locked',
'run_untagged', 'tag_list',
'maximum_timeout'))
_update_attrs = (tuple(), ('description', 'active', 'tag_list',
'run_untagged', 'locked', 'access_level',
'maximum_timeout'))
@cli.register_custom_action('RunnerManager', tuple(), ('scope', ))
@exc.on_http_error(exc.GitlabListError)
def all(self, scope=None, **kwargs):
"""List all the runners.
Args:
scope (str): The scope of runners to show, one of: specific,
shared, active, paused, online
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 server failed to perform the request
Returns:
list(Runner): a list of runners matching the scope.
"""
path = '/runners/all'
query_data = {}
if scope is not None:
query_data['scope'] = scope
return self.gitlab.http_list(path, query_data, **kwargs)
@cli.register_custom_action('RunnerManager', ('token',))
@exc.on_http_error(exc.GitlabVerifyError)
def verify(self, token, **kwargs):
"""Validates authentication credentials for a registered Runner.
Args:
token (str): The runner's authentication token
**kwargs: Extra options to send to the server (e.g. sudo)
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabVerifyError: If the server failed to verify the token
"""
path = '/runners/verify'
post_data = {'token': token}
self.gitlab.http_post(path, post_data=post_data, **kwargs)
class Todo(ObjectDeleteMixin, RESTObject):
@cli.register_custom_action('Todo')
@exc.on_http_error(exc.GitlabTodoError)
def mark_as_done(self, **kwargs):
"""Mark the todo as done.
Args:
**kwargs: Extra options to send to the server (e.g. sudo)
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabTodoError: If the server failed to perform the request
"""
path = '%s/%s/mark_as_done' % (self.manager.path, self.id)
server_data = self.manager.gitlab.http_post(path, **kwargs)
self._update_attrs(server_data)
class TodoManager(ListMixin, DeleteMixin, RESTManager):
_path = '/todos'
_obj_cls = Todo
_list_filters = ('action', 'author_id', 'project_id', 'state', 'type')
@cli.register_custom_action('TodoManager')
@exc.on_http_error(exc.GitlabTodoError)
def mark_all_as_done(self, **kwargs):
"""Mark all the todos as done.
Args:
**kwargs: Extra options to send to the server (e.g. sudo)
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabTodoError: If the server failed to perform the request
Returns:
int: The number of todos maked done
"""
result = self.gitlab.http_post('/todos/mark_as_done', **kwargs)
try:
return int(result)
except ValueError:
return 0
class GeoNode(SaveMixin, ObjectDeleteMixin, RESTObject):
@cli.register_custom_action('GeoNode')
@exc.on_http_error(exc.GitlabRepairError)
def repair(self, **kwargs):
"""Repair the OAuth authentication of the geo node.
Args:
**kwargs: Extra options to send to the server (e.g. sudo)
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabRepairError: If the server failed to perform the request
"""
path = '/geo_nodes/%s/repair' % self.get_id()
server_data = self.manager.gitlab.http_post(path, **kwargs)
self._update_attrs(server_data)
@cli.register_custom_action('GeoNode')
@exc.on_http_error(exc.GitlabGetError)
def status(self, **kwargs):
"""Get the status of the geo node.
Args:
**kwargs: Extra options to send to the server (e.g. sudo)
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabGetError: If the server failed to perform the request
Returns:
dict: The status of the geo node
"""
path = '/geo_nodes/%s/status' % self.get_id()
return self.manager.gitlab.http_get(path, **kwargs)
class GeoNodeManager(RetrieveMixin, UpdateMixin, DeleteMixin, RESTManager):
_path = '/geo_nodes'
_obj_cls = GeoNode
_update_attrs = (tuple(), ('enabled', 'url', 'files_max_capacity',
'repos_max_capacity'))
@cli.register_custom_action('GeoNodeManager')
@exc.on_http_error(exc.GitlabGetError)
def status(self, **kwargs):
"""Get the status of all the geo nodes.
Args:
**kwargs: Extra options to send to the server (e.g. sudo)
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabGetError: If the server failed to perform the request
Returns:
list: The status of all the geo nodes
"""
return self.gitlab.http_list('/geo_nodes/status', **kwargs)
@cli.register_custom_action('GeoNodeManager')
@exc.on_http_error(exc.GitlabGetError)
def current_failures(self, **kwargs):
"""Get the list of failures on the current geo node.
Args:
**kwargs: Extra options to send to the server (e.g. sudo)
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabGetError: If the server failed to perform the request
Returns:
list: The list of failures
"""
return self.gitlab.http_list('/geo_nodes/current/failures', **kwargs)