diff options
author | Gauvain Pocentek <gauvain@pocentek.net> | 2017-05-23 21:11:17 +0200 |
---|---|---|
committer | Gauvain Pocentek <gauvain@pocentek.net> | 2017-05-23 21:11:24 +0200 |
commit | e853a30b0c083fa835513a82816b315cf147092c (patch) | |
tree | 5fde97d15437f9cb9a8c4ce286d58f9113061009 /gitlab/objects.py | |
parent | f3738854f0d010bade44edc60404dbab984d2adb (diff) | |
download | gitlab-e853a30b0c083fa835513a82816b315cf147092c.tar.gz |
Reorganise the code to handle v3 and v4 objects
Having objects managing both versions will only make the code more
complicated, with lots of tests everywhere. This solution might generate
some code duplication, but it should be maintainable.
Diffstat (limited to 'gitlab/objects.py')
-rw-r--r-- | gitlab/objects.py | 2851 |
1 files changed, 0 insertions, 2851 deletions
diff --git a/gitlab/objects.py b/gitlab/objects.py deleted file mode 100644 index 630d415..0000000 --- a/gitlab/objects.py +++ /dev/null @@ -1,2851 +0,0 @@ -# -*- coding: utf-8 -*- -# -# Copyright (C) 2013-2015 Gauvain Pocentek <gauvain@pocentek.net> -# -# This program is free software: you can redistribute it and/or modify -# it under the terms of the GNU Lesser General Public License as published by -# the Free Software Foundation, either version 3 of the License, or -# (at your option) any later version. -# -# This program is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU Lesser General Public License for more details. -# -# You should have received a copy of the GNU Lesser General Public License -# along with this program. If not, see <http://www.gnu.org/licenses/>. - -from __future__ import print_function -from __future__ import division -from __future__ import absolute_import -import base64 -import copy -import itertools -import json -import sys -import urllib -import warnings - -import six - -import gitlab -from gitlab.exceptions import * # noqa -from gitlab import utils - - -class jsonEncoder(json.JSONEncoder): - def default(self, obj): - if isinstance(obj, GitlabObject): - return obj.as_dict() - elif isinstance(obj, gitlab.Gitlab): - return {'url': obj._url} - return json.JSONEncoder.default(self, obj) - - -class BaseManager(object): - """Base manager class for API operations. - - Managers provide method to manage GitLab API objects, such as retrieval, - listing, creation. - - Inherited class must define the ``obj_cls`` attribute. - - Attributes: - obj_cls (class): class of objects wrapped by this manager. - """ - - obj_cls = None - - def __init__(self, gl, parent=None, args=[]): - """Constructs a manager. - - Args: - gl (gitlab.Gitlab): Gitlab object referencing the GitLab server. - parent (Optional[Manager]): A parent manager. - args (list): A list of tuples defining a link between the - parent/child attributes. - - Raises: - AttributeError: If `obj_cls` is None. - """ - self.gitlab = gl - self.args = args - self.parent = parent - - if self.obj_cls is None: - raise AttributeError("obj_cls must be defined") - - def _set_parent_args(self, **kwargs): - args = copy.copy(kwargs) - if self.parent is not None: - for attr, parent_attr in self.args: - args.setdefault(attr, getattr(self.parent, parent_attr)) - - return args - - def get(self, id=None, **kwargs): - """Get a GitLab object. - - Args: - id: ID of the object to retrieve. - **kwargs: Additional arguments to send to GitLab. - - Returns: - object: An object of class `obj_cls`. - - Raises: - NotImplementedError: If objects cannot be retrieved. - GitlabGetError: If the server fails to perform the request. - """ - args = self._set_parent_args(**kwargs) - if not self.obj_cls.canGet: - raise NotImplementedError - if id is None and self.obj_cls.getRequiresId is True: - raise ValueError('The id argument must be defined.') - return self.obj_cls.get(self.gitlab, id, **args) - - def list(self, **kwargs): - """Get a list of GitLab objects. - - Args: - **kwargs: Additional arguments to send to GitLab. - - Returns: - list[object]: A list of `obj_cls` objects. - - Raises: - NotImplementedError: If objects cannot be listed. - GitlabListError: If the server fails to perform the request. - """ - args = self._set_parent_args(**kwargs) - if not self.obj_cls.canList: - raise NotImplementedError - return self.obj_cls.list(self.gitlab, **args) - - def create(self, data, **kwargs): - """Create a new object of class `obj_cls`. - - Args: - data (dict): The parameters to send to the GitLab server to create - the object. Required and optional arguments are defined in the - `requiredCreateAttrs` and `optionalCreateAttrs` of the - `obj_cls` class. - **kwargs: Additional arguments to send to GitLab. - - Returns: - object: A newly create `obj_cls` object. - - Raises: - NotImplementedError: If objects cannot be created. - GitlabCreateError: If the server fails to perform the request. - """ - args = self._set_parent_args(**kwargs) - if not self.obj_cls.canCreate: - raise NotImplementedError - return self.obj_cls.create(self.gitlab, data, **args) - - def delete(self, id, **kwargs): - """Delete a GitLab object. - - Args: - id: ID of the object to delete. - - Raises: - NotImplementedError: If objects cannot be deleted. - GitlabDeleteError: If the server fails to perform the request. - """ - args = self._set_parent_args(**kwargs) - if not self.obj_cls.canDelete: - raise NotImplementedError - self.gitlab.delete(self.obj_cls, id, **args) - - -class GitlabObject(object): - """Base class for all classes that interface with GitLab.""" - #: Url to use in GitLab for this object - _url = None - # Some objects (e.g. merge requests) have different urls for singular and - # plural - _urlPlural = None - _id_in_delete_url = True - _id_in_update_url = True - _constructorTypes = None - - #: Tells if GitLab-api allows retrieving single objects. - canGet = True - #: Tells if GitLab-api allows listing of objects. - canList = True - #: Tells if GitLab-api allows creation of new objects. - canCreate = True - #: Tells if GitLab-api allows updating object. - canUpdate = True - #: Tells if GitLab-api allows deleting object. - canDelete = True - #: Attributes that are required for constructing url. - requiredUrlAttrs = [] - #: Attributes that are required when retrieving list of objects. - requiredListAttrs = [] - #: Attributes that are optional when retrieving list of objects. - optionalListAttrs = [] - #: Attributes that are optional when retrieving single object. - optionalGetAttrs = [] - #: Attributes that are required when retrieving single object. - requiredGetAttrs = [] - #: Attributes that are required when deleting object. - requiredDeleteAttrs = [] - #: Attributes that are required when creating a new object. - requiredCreateAttrs = [] - #: Attributes that are optional when creating a new object. - optionalCreateAttrs = [] - #: Attributes that are required when updating an object. - requiredUpdateAttrs = [] - #: Attributes that are optional when updating an object. - optionalUpdateAttrs = [] - #: Whether the object ID is required in the GET url. - getRequiresId = True - #: List of managers to create. - managers = [] - #: Name of the identifier of an object. - idAttr = 'id' - #: Attribute to use as ID when displaying the object. - shortPrintAttr = None - - def _data_for_gitlab(self, extra_parameters={}, update=False, - as_json=True): - data = {} - if update and (self.requiredUpdateAttrs or self.optionalUpdateAttrs): - attributes = itertools.chain(self.requiredUpdateAttrs, - self.optionalUpdateAttrs) - else: - attributes = itertools.chain(self.requiredCreateAttrs, - self.optionalCreateAttrs) - attributes = list(attributes) + ['sudo', 'page', 'per_page'] - for attribute in attributes: - if hasattr(self, attribute): - value = getattr(self, attribute) - # labels need to be sent as a comma-separated list - if attribute == 'labels' and isinstance(value, list): - value = ", ".join(value) - elif attribute == 'sudo': - value = str(value) - data[attribute] = value - - data.update(extra_parameters) - - return json.dumps(data) if as_json else data - - @classmethod - def list(cls, gl, **kwargs): - """Retrieve a list of objects from GitLab. - - Args: - gl (gitlab.Gitlab): Gitlab object referencing the GitLab server. - per_page (int): Maximum number of items to return. - page (int): ID of the page to return when using pagination. - - Returns: - list[object]: A list of objects. - - Raises: - NotImplementedError: If objects can't be listed. - GitlabListError: If the server cannot perform the request. - """ - if not cls.canList: - raise NotImplementedError - - if not cls._url: - raise NotImplementedError - - return gl.list(cls, **kwargs) - - @classmethod - def get(cls, gl, id, **kwargs): - """Retrieve a single object. - - Args: - gl (gitlab.Gitlab): Gitlab object referencing the GitLab server. - id (int or str): ID of the object to retrieve. - - Returns: - object: The found GitLab object. - - Raises: - NotImplementedError: If objects can't be retrieved. - GitlabGetError: If the server cannot perform the request. - """ - - if cls.canGet is False: - raise NotImplementedError - elif cls.canGet is True: - return cls(gl, id, **kwargs) - elif cls.canGet == 'from_list': - for obj in cls.list(gl, **kwargs): - obj_id = getattr(obj, obj.idAttr) - if str(obj_id) == str(id): - return obj - - raise GitlabGetError("Object not found") - - def _get_object(self, k, v, **kwargs): - if self._constructorTypes and k in self._constructorTypes: - return globals()[self._constructorTypes[k]](self.gitlab, v, - **kwargs) - else: - return v - - def _set_from_dict(self, data, **kwargs): - if not hasattr(data, 'items'): - return - - for k, v in data.items(): - # If a k attribute already exists and is a Manager, do nothing (see - # https://github.com/python-gitlab/python-gitlab/issues/209) - if isinstance(getattr(self, k, None), BaseManager): - continue - - if isinstance(v, list): - self.__dict__[k] = [] - for i in v: - self.__dict__[k].append(self._get_object(k, i, **kwargs)) - elif v is None: - self.__dict__[k] = None - else: - self.__dict__[k] = self._get_object(k, v, **kwargs) - - def _create(self, **kwargs): - if not self.canCreate: - raise NotImplementedError - - json = self.gitlab.create(self, **kwargs) - self._set_from_dict(json) - self._from_api = True - - def _update(self, **kwargs): - if not self.canUpdate: - raise NotImplementedError - - json = self.gitlab.update(self, **kwargs) - self._set_from_dict(json) - - def save(self, **kwargs): - if self._from_api: - self._update(**kwargs) - else: - self._create(**kwargs) - - def delete(self, **kwargs): - if not self.canDelete: - raise NotImplementedError - - if not self._from_api: - raise GitlabDeleteError("Object not yet created") - - return self.gitlab.delete(self, **kwargs) - - @classmethod - def create(cls, gl, data, **kwargs): - """Create an object. - - Args: - gl (gitlab.Gitlab): Gitlab object referencing the GitLab server. - data (dict): The data used to define the object. - - Returns: - object: The new object. - - Raises: - NotImplementedError: If objects can't be created. - GitlabCreateError: If the server cannot perform the request. - """ - if not cls.canCreate: - raise NotImplementedError - - obj = cls(gl, data, **kwargs) - obj.save() - - return obj - - def __init__(self, gl, data=None, **kwargs): - """Constructs a new object. - - Do not use this method. Use the `get` or `create` class methods - instead. - - Args: - gl (gitlab.Gitlab): Gitlab object referencing the GitLab server. - data: If `data` is a dict, create a new object using the - information. If it is an int or a string, get a GitLab object - from an API request. - **kwargs: Additional arguments to send to GitLab. - """ - self._from_api = False - #: (gitlab.Gitlab): Gitlab connection. - self.gitlab = gl - - if (data is None or isinstance(data, six.integer_types) or - isinstance(data, six.string_types)): - if not self.canGet: - raise NotImplementedError - data = self.gitlab.get(self.__class__, data, **kwargs) - self._from_api = True - - # the API returned a list because custom kwargs where used - # instead of the id to request an object. Usually parameters - # other than an id return ambiguous results. However in the - # gitlab universe iids together with a project_id are - # unambiguous for merge requests and issues, too. - # So if there is only one element we can use it as our data - # source. - if 'iid' in kwargs and isinstance(data, list): - if len(data) < 1: - raise GitlabGetError('Not found') - elif len(data) == 1: - data = data[0] - else: - raise GitlabGetError('Impossible! You found multiple' - ' elements with the same iid.') - - self._set_from_dict(data, **kwargs) - - if kwargs: - for k, v in kwargs.items(): - # Don't overwrite attributes returned by the server (#171) - if k not in self.__dict__ or not self.__dict__[k]: - self.__dict__[k] = v - - # Special handling for api-objects that don't have id-number in api - # responses. Currently only Labels and Files - if not hasattr(self, "id"): - self.id = None - - def _set_manager(self, var, cls, attrs): - manager = cls(self.gitlab, self, attrs) - setattr(self, var, manager) - - def __getattr__(self, name): - # build a manager if it doesn't exist yet - for var, cls, attrs in self.managers: - if var != name: - continue - self._set_manager(var, cls, attrs) - return getattr(self, var) - - raise AttributeError - - def __str__(self): - return '%s => %s' % (type(self), str(self.__dict__)) - - def __repr__(self): - return '<%s %s:%s>' % (self.__class__.__name__, - self.idAttr, - getattr(self, self.idAttr)) - - def display(self, pretty): - if pretty: - self.pretty_print() - else: - self.short_print() - - def short_print(self, depth=0): - """Print the object on the standard output (verbose). - - Args: - depth (int): Used internaly for recursive call. - """ - id = self.__dict__[self.idAttr] - print("%s%s: %s" % (" " * depth * 2, self.idAttr, id)) - if self.shortPrintAttr: - print("%s%s: %s" % (" " * depth * 2, - self.shortPrintAttr.replace('_', '-'), - self.__dict__[self.shortPrintAttr])) - - @staticmethod - def _get_display_encoding(): - return sys.stdout.encoding or sys.getdefaultencoding() - - @staticmethod - def _obj_to_str(obj): - if isinstance(obj, dict): - s = ", ".join(["%s: %s" % - (x, GitlabObject._obj_to_str(y)) - for (x, y) in obj.items()]) - return "{ %s }" % s - elif isinstance(obj, list): - s = ", ".join([GitlabObject._obj_to_str(x) for x in obj]) - return "[ %s ]" % s - elif six.PY2 and isinstance(obj, six.text_type): - return obj.encode(GitlabObject._get_display_encoding(), "replace") - else: - return str(obj) - - def pretty_print(self, depth=0): - """Print the object on the standard output (verbose). - - Args: - depth (int): Used internaly for recursive call. - """ - id = self.__dict__[self.idAttr] - print("%s%s: %s" % (" " * depth * 2, self.idAttr, id)) - for k in sorted(self.__dict__.keys()): - if k in (self.idAttr, 'id', 'gitlab'): - continue - if k[0] == '_': - continue - v = self.__dict__[k] - pretty_k = k.replace('_', '-') - if six.PY2: - pretty_k = pretty_k.encode( - GitlabObject._get_display_encoding(), "replace") - if isinstance(v, GitlabObject): - if depth == 0: - print("%s:" % pretty_k) - v.pretty_print(1) - else: - print("%s: %s" % (pretty_k, v.id)) - elif isinstance(v, BaseManager): - continue - else: - if hasattr(v, __name__) and v.__name__ == 'Gitlab': - continue - v = GitlabObject._obj_to_str(v) - print("%s%s: %s" % (" " * depth * 2, pretty_k, v)) - - def json(self): - """Dump the object as json. - - Returns: - str: The json string. - """ - return json.dumps(self, cls=jsonEncoder) - - def as_dict(self): - """Dump the object as a dict.""" - return {k: v for k, v in six.iteritems(self.__dict__) - if (not isinstance(v, BaseManager) and not k[0] == '_')} - - def __eq__(self, other): - if type(other) is type(self): - return self.as_dict() == other.as_dict() - return False - - def __ne__(self, other): - return not self.__eq__(other) - - -class SidekiqManager(object): - """Manager for the Sidekiq methods. - - This manager doesn't actually manage objects but provides helper fonction - for the sidekiq metrics API. - """ - def __init__(self, gl): - """Constructs a Sidekiq manager. - - Args: - gl (gitlab.Gitlab): Gitlab object referencing the GitLab server. - """ - self.gitlab = gl - - def _simple_get(self, url, **kwargs): - r = self.gitlab._raw_get(url, **kwargs) - raise_error_from_response(r, GitlabGetError) - return r.json() - - def queue_metrics(self, **kwargs): - """Returns the registred queues information.""" - return self._simple_get('/sidekiq/queue_metrics', **kwargs) - - def process_metrics(self, **kwargs): - """Returns the registred sidekiq workers.""" - return self._simple_get('/sidekiq/process_metrics', **kwargs) - - def job_stats(self, **kwargs): - """Returns statistics about the jobs performed.""" - return self._simple_get('/sidekiq/job_stats', **kwargs) - - def compound_metrics(self, **kwargs): - """Returns all available metrics and statistics.""" - return self._simple_get('/sidekiq/compound_metrics', **kwargs) - - -class UserEmail(GitlabObject): - _url = '/users/%(user_id)s/emails' - canUpdate = False - shortPrintAttr = 'email' - requiredUrlAttrs = ['user_id'] - requiredCreateAttrs = ['email'] - - -class UserEmailManager(BaseManager): - obj_cls = UserEmail - - -class UserKey(GitlabObject): - _url = '/users/%(user_id)s/keys' - canGet = 'from_list' - canUpdate = False - requiredUrlAttrs = ['user_id'] - requiredCreateAttrs = ['title', 'key'] - - -class UserKeyManager(BaseManager): - obj_cls = UserKey - - -class UserProject(GitlabObject): - _url = '/projects/user/%(user_id)s' - _constructorTypes = {'owner': 'User', 'namespace': 'Group'} - canUpdate = False - canDelete = False - canList = False - canGet = False - requiredUrlAttrs = ['user_id'] - requiredCreateAttrs = ['name'] - optionalCreateAttrs = ['default_branch', 'issues_enabled', 'wall_enabled', - 'merge_requests_enabled', 'wiki_enabled', - 'snippets_enabled', 'public', 'visibility_level', - 'description', 'builds_enabled', 'public_builds', - 'import_url', 'only_allow_merge_if_build_succeeds'] - - -class UserProjectManager(BaseManager): - obj_cls = UserProject - - -class User(GitlabObject): - _url = '/users' - shortPrintAttr = 'username' - requiredCreateAttrs = ['email', 'username', 'name'] - optionalCreateAttrs = ['password', 'reset_password', 'skype', 'linkedin', - 'twitter', 'projects_limit', 'extern_uid', - 'provider', 'bio', 'admin', 'can_create_group', - 'website_url', 'confirm', 'external', - 'organization', 'location'] - requiredUpdateAttrs = ['email', 'username', 'name'] - optionalUpdateAttrs = ['password', 'skype', 'linkedin', 'twitter', - 'projects_limit', 'extern_uid', 'provider', 'bio', - 'admin', 'can_create_group', 'website_url', - 'confirm', 'external', 'organization', 'location'] - managers = ( - ('emails', UserEmailManager, [('user_id', 'id')]), - ('keys', UserKeyManager, [('user_id', 'id')]), - ('projects', UserProjectManager, [('user_id', 'id')]), - ) - - def _data_for_gitlab(self, extra_parameters={}, update=False, - as_json=True): - if hasattr(self, 'confirm'): - self.confirm = str(self.confirm).lower() - return super(User, self)._data_for_gitlab(extra_parameters) - - def block(self, **kwargs): - """Blocks the user.""" - url = '/users/%s/block' % self.id - r = self.gitlab._raw_put(url, **kwargs) - raise_error_from_response(r, GitlabBlockError) - self.state = 'blocked' - - def unblock(self, **kwargs): - """Unblocks the user.""" - url = '/users/%s/unblock' % self.id - r = self.gitlab._raw_put(url, **kwargs) - raise_error_from_response(r, GitlabUnblockError) - self.state = 'active' - - def __eq__(self, other): - if type(other) is type(self): - selfdict = self.as_dict() - otherdict = other.as_dict() - selfdict.pop('password', None) - otherdict.pop('password', None) - return selfdict == otherdict - return False - - -class UserManager(BaseManager): - obj_cls = User - - def search(self, query, **kwargs): - """Search users. - - Args: - query (str): The query string to send to GitLab for the search. - all (bool): If True, return all the items, without pagination - **kwargs: Additional arguments to send to GitLab. - - Returns: - list(User): A list of matching users. - - Raises: - GitlabConnectionError: If the server cannot be reached. - GitlabListError: If the server fails to perform the request. - """ - url = self.obj_cls._url + '?search=' + query - return self.gitlab._raw_list(url, self.obj_cls, **kwargs) - - def get_by_username(self, username, **kwargs): - """Get a user by its username. - - Args: - username (str): The name of the user. - **kwargs: Additional arguments to send to GitLab. - - Returns: - User: The matching user. - - Raises: - GitlabConnectionError: If the server cannot be reached. - GitlabGetError: If the server fails to perform the request. - """ - url = self.obj_cls._url + '?username=' + username - results = self.gitlab._raw_list(url, self.obj_cls, **kwargs) - assert len(results) in (0, 1) - try: - return results[0] - except IndexError: - raise GitlabGetError('no such user: ' + username) - - -class CurrentUserEmail(GitlabObject): - _url = '/user/emails' - canUpdate = False - shortPrintAttr = 'email' - requiredCreateAttrs = ['email'] - - -class CurrentUserEmailManager(BaseManager): - obj_cls = CurrentUserEmail - - -class CurrentUserKey(GitlabObject): - _url = '/user/keys' - canUpdate = False - shortPrintAttr = 'title' - requiredCreateAttrs = ['title', 'key'] - - -class CurrentUserKeyManager(BaseManager): - obj_cls = CurrentUserKey - - -class CurrentUser(GitlabObject): - _url = '/user' - canList = False - canCreate = False - canUpdate = False - canDelete = False - shortPrintAttr = 'username' - managers = ( - ('emails', CurrentUserEmailManager, [('user_id', 'id')]), - ('keys', CurrentUserKeyManager, [('user_id', 'id')]), - ) - - -class ApplicationSettings(GitlabObject): - _url = '/application/settings' - _id_in_update_url = False - getRequiresId = False - optionalUpdateAttrs = ['after_sign_out_path', - 'container_registry_token_expire_delay', - 'default_branch_protection', - 'default_project_visibility', - 'default_projects_limit', - 'default_snippet_visibility', - 'domain_blacklist', - 'domain_blacklist_enabled', - 'domain_whitelist', - 'enabled_git_access_protocol', - 'gravatar_enabled', - 'home_page_url', - 'max_attachment_size', - 'repository_storage', - 'restricted_signup_domains', - 'restricted_visibility_levels', - 'session_expire_delay', - 'sign_in_text', - 'signin_enabled', - 'signup_enabled', - 'twitter_sharing_enabled', - 'user_oauth_applications'] - canList = False - canCreate = False - canDelete = False - - def _data_for_gitlab(self, extra_parameters={}, update=False, - as_json=True): - data = (super(ApplicationSettings, self) - ._data_for_gitlab(extra_parameters, update=update, - as_json=False)) - if not self.domain_whitelist: - data.pop('domain_whitelist', None) - return json.dumps(data) - - -class ApplicationSettingsManager(BaseManager): - obj_cls = ApplicationSettings - - -class BroadcastMessage(GitlabObject): - _url = '/broadcast_messages' - requiredCreateAttrs = ['message'] - optionalCreateAttrs = ['starts_at', 'ends_at', 'color', 'font'] - requiredUpdateAttrs = [] - optionalUpdateAttrs = ['message', 'starts_at', 'ends_at', 'color', 'font'] - - -class BroadcastMessageManager(BaseManager): - obj_cls = BroadcastMessage - - -class Key(GitlabObject): - _url = '/deploy_keys' - canGet = 'from_list' - canCreate = False - canUpdate = False - canDelete = False - - def __init__(self, *args, **kwargs): - warnings.warn("`Key` is deprecated, use `DeployKey` instead", - DeprecationWarning) - super(Key, self).__init__(*args, **kwargs) - - -class KeyManager(BaseManager): - obj_cls = Key - - -class DeployKey(GitlabObject): - _url = '/deploy_keys' - canGet = 'from_list' - canCreate = False - canUpdate = False - canDelete = False - - -class DeployKeyManager(BaseManager): - obj_cls = DeployKey - - -class NotificationSettings(GitlabObject): - _url = '/notification_settings' - _id_in_update_url = False - getRequiresId = False - optionalUpdateAttrs = ['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'] - canList = False - canCreate = False - canDelete = False - - -class NotificationSettingsManager(BaseManager): - obj_cls = NotificationSettings - - -class Gitignore(GitlabObject): - _url = '/templates/gitignores' - canDelete = False - canUpdate = False - canCreate = False - idAttr = 'name' - - -class GitignoreManager(BaseManager): - obj_cls = Gitignore - - -class Gitlabciyml(GitlabObject): - _url = '/templates/gitlab_ci_ymls' - canDelete = False - canUpdate = False - canCreate = False - idAttr = 'name' - - -class GitlabciymlManager(BaseManager): - obj_cls = Gitlabciyml - - -class GroupIssue(GitlabObject): - _url = '/groups/%(group_id)s/issues' - canGet = 'from_list' - canCreate = False - canUpdate = False - canDelete = False - requiredUrlAttrs = ['group_id'] - optionalListAttrs = ['state', 'labels', 'milestone', 'order_by', 'sort'] - - -class GroupIssueManager(BaseManager): - obj_cls = GroupIssue - - -class GroupMember(GitlabObject): - _url = '/groups/%(group_id)s/members' - canGet = 'from_list' - requiredUrlAttrs = ['group_id'] - requiredCreateAttrs = ['access_level', 'user_id'] - optionalCreateAttrs = ['expires_at'] - requiredUpdateAttrs = ['access_level'] - optionalCreateAttrs = ['expires_at'] - shortPrintAttr = 'username' - - def _update(self, **kwargs): - self.user_id = self.id - super(GroupMember, self)._update(**kwargs) - - -class GroupMemberManager(BaseManager): - obj_cls = GroupMember - - -class GroupNotificationSettings(NotificationSettings): - _url = '/groups/%(group_id)s/notification_settings' - requiredUrlAttrs = ['group_id'] - - -class GroupNotificationSettingsManager(BaseManager): - obj_cls = GroupNotificationSettings - - -class GroupAccessRequest(GitlabObject): - _url = '/groups/%(group_id)s/access_requests' - canGet = 'from_list' - canUpdate = False - - def approve(self, access_level=gitlab.DEVELOPER_ACCESS, **kwargs): - """Approve an access request. - - Attrs: - access_level (int): The access level for the user. - - Raises: - GitlabConnectionError: If the server cannot be reached. - GitlabUpdateError: If the server fails to perform the request. - """ - - url = ('/groups/%(group_id)s/access_requests/%(id)s/approve' % - {'group_id': self.group_id, 'id': self.id}) - data = {'access_level': access_level} - r = self.gitlab._raw_put(url, data=data, **kwargs) - raise_error_from_response(r, GitlabUpdateError, 201) - self._set_from_dict(r.json()) - - -class GroupAccessRequestManager(BaseManager): - obj_cls = GroupAccessRequest - - -class Hook(GitlabObject): - _url = '/hooks' - canUpdate = False - requiredCreateAttrs = ['url'] - shortPrintAttr = 'url' - - -class HookManager(BaseManager): - obj_cls = Hook - - -class Issue(GitlabObject): - _url = '/issues' - _constructorTypes = {'author': 'User', 'assignee': 'User', - 'milestone': 'ProjectMilestone'} - canGet = 'from_list' - canDelete = False - canUpdate = False - canCreate = False - shortPrintAttr = 'title' - optionalListAttrs = ['state', 'labels', 'order_by', 'sort'] - - -class IssueManager(BaseManager): - obj_cls = Issue - - -class License(GitlabObject): - _url = '/licenses' - canDelete = False - canUpdate = False - canCreate = False - idAttr = 'key' - - optionalListAttrs = ['popular'] - optionalGetAttrs = ['project', 'fullname'] - - -class LicenseManager(BaseManager): - obj_cls = License - - -class Snippet(GitlabObject): - _url = '/snippets' - _constructorTypes = {'author': 'User'} - requiredCreateAttrs = ['title', 'file_name', 'content'] - optionalCreateAttrs = ['lifetime', 'visibility_level'] - optionalUpdateAttrs = ['title', 'file_name', 'content', 'visibility_level'] - shortPrintAttr = 'title' - - def raw(self, streamed=False, action=None, chunk_size=1024, **kwargs): - """Return the raw 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. - - Returns: - str: The snippet content. - - Raises: - GitlabConnectionError: If the server cannot be reached. - GitlabGetError: If the server fails to perform the request. - """ - url = ("/snippets/%(snippet_id)s/raw" % {'snippet_id': self.id}) - r = self.gitlab._raw_get(url, **kwargs) - raise_error_from_response(r, GitlabGetError) - return utils.response_content(r, streamed, action, chunk_size) - - -class SnippetManager(BaseManager): - obj_cls = Snippet - - def public(self, **kwargs): - """List all the public snippets. - - Args: - all (bool): If True, return all the items, without pagination - **kwargs: Additional arguments to send to GitLab. - - Returns: - list(gitlab.Gitlab.Snippet): The list of snippets. - """ - return self.gitlab._raw_list("/snippets/public", Snippet, **kwargs) - - -class Namespace(GitlabObject): - _url = '/namespaces' - canGet = 'from_list' - canUpdate = False - canDelete = False - canCreate = False - optionalListAttrs = ['search'] - - -class NamespaceManager(BaseManager): - obj_cls = Namespace - - -class ProjectBoardList(GitlabObject): - _url = '/projects/%(project_id)s/boards/%(board_id)s/lists' - requiredUrlAttrs = ['project_id', 'board_id'] - _constructorTypes = {'label': 'ProjectLabel'} - requiredCreateAttrs = ['label_id'] - requiredUpdateAttrs = ['position'] - - -class ProjectBoardListManager(BaseManager): - obj_cls = ProjectBoardList - - -class ProjectBoard(GitlabObject): - _url = '/projects/%(project_id)s/boards' - requiredUrlAttrs = ['project_id'] - _constructorTypes = {'labels': 'ProjectBoardList'} - canGet = 'from_list' - canUpdate = False - canCreate = False - canDelete = False - managers = ( - ('lists', ProjectBoardListManager, - [('project_id', 'project_id'), ('board_id', 'id')]), - ) - - -class ProjectBoardManager(BaseManager): - obj_cls = ProjectBoard - - -class ProjectBranch(GitlabObject): - _url = '/projects/%(project_id)s/repository/branches' - _constructorTypes = {'author': 'User', "committer": "User"} - - idAttr = 'name' - canUpdate = False - requiredUrlAttrs = ['project_id'] - requiredCreateAttrs = ['branch_name', 'ref'] - - def protect(self, protect=True, **kwargs): - """Protects the branch.""" - url = self._url % {'project_id': self.project_id} - action = 'protect' if protect else 'unprotect' - url = "%s/%s/%s" % (url, self.name, action) - r = self.gitlab._raw_put(url, data=None, content_type=None, **kwargs) - raise_error_from_response(r, GitlabProtectError) - - if protect: - self.protected = protect - else: - del self.protected - - def unprotect(self, **kwargs): - """Unprotects the branch.""" - self.protect(False, **kwargs) - - -class ProjectBranchManager(BaseManager): - obj_cls = ProjectBranch - - -class ProjectBuild(GitlabObject): - _url = '/projects/%(project_id)s/builds' - _constructorTypes = {'user': 'User', - 'commit': 'ProjectCommit', - 'runner': 'Runner'} - requiredUrlAttrs = ['project_id'] - canDelete = False - canUpdate = False - canCreate = False - - def cancel(self, **kwargs): - """Cancel the build.""" - url = '/projects/%s/builds/%s/cancel' % (self.project_id, self.id) - r = self.gitlab._raw_post(url) - raise_error_from_response(r, GitlabBuildCancelError, 201) - - def retry(self, **kwargs): - """Retry the build.""" - url = '/projects/%s/builds/%s/retry' % (self.project_id, self.id) - r = self.gitlab._raw_post(url) - raise_error_from_response(r, GitlabBuildRetryError, 201) - - def play(self, **kwargs): - """Trigger a build explicitly.""" - url = '/projects/%s/builds/%s/play' % (self.project_id, self.id) - r = self.gitlab._raw_post(url) - raise_error_from_response(r, GitlabBuildPlayError) - - def erase(self, **kwargs): - """Erase the build (remove build artifacts and trace).""" - url = '/projects/%s/builds/%s/erase' % (self.project_id, self.id) - r = self.gitlab._raw_post(url) - raise_error_from_response(r, GitlabBuildEraseError, 201) - - def keep_artifacts(self, **kwargs): - """Prevent artifacts from being delete when expiration is set. - - Raises: - GitlabConnectionError: If the server cannot be reached. - GitlabCreateError: If the request failed. - """ - url = ('/projects/%s/builds/%s/artifacts/keep' % - (self.project_id, self.id)) - r = self.gitlab._raw_post(url) - raise_error_from_response(r, GitlabGetError, 200) - - def artifacts(self, streamed=False, action=None, chunk_size=1024, - **kwargs): - """Get the build 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. - - Returns: - str: The artifacts if `streamed` is False, None otherwise. - - Raises: - GitlabConnectionError: If the server cannot be reached. - GitlabGetError: If the artifacts are not available. - """ - url = '/projects/%s/builds/%s/artifacts' % (self.project_id, self.id) - r = self.gitlab._raw_get(url, streamed=streamed, **kwargs) - raise_error_from_response(r, GitlabGetError, 200) - return utils.response_content(r, streamed, action, chunk_size) - - def trace(self, streamed=False, action=None, chunk_size=1024, **kwargs): - """Get the build 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. - - Returns: - str: The trace. - - Raises: - GitlabConnectionError: If the server cannot be reached. - GitlabGetError: If the trace is not available. - """ - url = '/projects/%s/builds/%s/trace' % (self.project_id, self.id) - r = self.gitlab._raw_get(url, streamed=streamed, **kwargs) - raise_error_from_response(r, GitlabGetError, 200) - return utils.response_content(r, streamed, action, chunk_size) - - -class ProjectBuildManager(BaseManager): - obj_cls = ProjectBuild - - -class ProjectCommitStatus(GitlabObject): - _url = '/projects/%(project_id)s/repository/commits/%(commit_id)s/statuses' - _create_url = '/projects/%(project_id)s/statuses/%(commit_id)s' - canUpdate = False - canDelete = False - requiredUrlAttrs = ['project_id', 'commit_id'] - optionalGetAttrs = ['ref_name', 'stage', 'name', 'all'] - requiredCreateAttrs = ['state'] - optionalCreateAttrs = ['description', 'name', 'context', 'ref', - 'target_url'] - - -class ProjectCommitStatusManager(BaseManager): - obj_cls = ProjectCommitStatus - - -class ProjectCommitComment(GitlabObject): - _url = '/projects/%(project_id)s/repository/commits/%(commit_id)s/comments' - canUpdate = False - canGet = False - canDelete = False - requiredUrlAttrs = ['project_id', 'commit_id'] - requiredCreateAttrs = ['note'] - optionalCreateAttrs = ['path', 'line', 'line_type'] - - -class ProjectCommitCommentManager(BaseManager): - obj_cls = ProjectCommitComment - - -class ProjectCommit(GitlabObject): - _url = '/projects/%(project_id)s/repository/commits' - canDelete = False - canUpdate = False - requiredUrlAttrs = ['project_id'] - requiredCreateAttrs = ['branch_name', 'commit_message', 'actions'] - optionalCreateAttrs = ['author_email', 'author_name'] - shortPrintAttr = 'title' - managers = ( - ('comments', ProjectCommitCommentManager, - [('project_id', 'project_id'), ('commit_id', 'id')]), - ('statuses', ProjectCommitStatusManager, - [('project_id', 'project_id'), ('commit_id', 'id')]), - ) - - def diff(self, **kwargs): - """Generate the commit diff.""" - url = ('/projects/%(project_id)s/repository/commits/%(commit_id)s/diff' - % {'project_id': self.project_id, 'commit_id': self.id}) - r = self.gitlab._raw_get(url, **kwargs) - raise_error_from_response(r, GitlabGetError) - - return r.json() - - def blob(self, filepath, streamed=False, action=None, chunk_size=1024, - **kwargs): - """Generate the content of a file for this commit. - - Args: - filepath (str): Path of the file to request. - 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. - - Returns: - str: The content of the file - - Raises: - GitlabConnectionError: If the server cannot be reached. - GitlabGetError: If the server fails to perform the request. - """ - url = ('/projects/%(project_id)s/repository/blobs/%(commit_id)s' % - {'project_id': self.project_id, 'commit_id': self.id}) - url += '?filepath=%s' % filepath - r = self.gitlab._raw_get(url, streamed=streamed, **kwargs) - raise_error_from_response(r, GitlabGetError) - return utils.response_content(r, streamed, action, chunk_size) - - def builds(self, **kwargs): - """List the build for this commit. - - Returns: - list(ProjectBuild): A list of builds. - - Raises: - GitlabConnectionError: If the server cannot be reached. - GitlabListError: If the server fails to perform the request. - """ - url = '/projects/%s/repository/commits/%s/builds' % (self.project_id, - self.id) - return self.gitlab._raw_list(url, ProjectBuild, - {'project_id': self.project_id}, - **kwargs) - - def cherry_pick(self, branch, **kwargs): - """Cherry-pick a commit into a branch. - - Args: - branch (str): Name of target branch. - - Raises: - GitlabCherryPickError: If the cherry pick could not be applied. - """ - url = ('/projects/%s/repository/commits/%s/cherry_pick' % - (self.project_id, self.id)) - - r = self.gitlab._raw_post(url, data={'project_id': self.project_id, - 'branch': branch}, **kwargs) - errors = {400: GitlabCherryPickError} - raise_error_from_response(r, errors, expected_code=201) - - -class ProjectCommitManager(BaseManager): - obj_cls = ProjectCommit - - -class ProjectEnvironment(GitlabObject): - _url = '/projects/%(project_id)s/environments' - canGet = 'from_list' - requiredUrlAttrs = ['project_id'] - requiredCreateAttrs = ['name'] - optionalCreateAttrs = ['external_url'] - optionalUpdateAttrs = ['name', 'external_url'] - - -class ProjectEnvironmentManager(BaseManager): - obj_cls = ProjectEnvironment - - -class ProjectKey(GitlabObject): - _url = '/projects/%(project_id)s/keys' - canUpdate = False - requiredUrlAttrs = ['project_id'] - requiredCreateAttrs = ['title', 'key'] - - -class ProjectKeyManager(BaseManager): - obj_cls = ProjectKey - - def enable(self, key_id): - """Enable a deploy key for a project.""" - url = '/projects/%s/deploy_keys/%s/enable' % (self.parent.id, key_id) - r = self.gitlab._raw_post(url) - raise_error_from_response(r, GitlabProjectDeployKeyError, 201) - - def disable(self, key_id): - """Disable a deploy key for a project.""" - url = '/projects/%s/deploy_keys/%s/disable' % (self.parent.id, key_id) - r = self.gitlab._raw_delete(url) - raise_error_from_response(r, GitlabProjectDeployKeyError, 200) - - -class ProjectEvent(GitlabObject): - _url = '/projects/%(project_id)s/events' - canGet = 'from_list' - canDelete = False - canUpdate = False - canCreate = False - requiredUrlAttrs = ['project_id'] - shortPrintAttr = 'target_title' - - -class ProjectEventManager(BaseManager): - obj_cls = ProjectEvent - - -class ProjectFork(GitlabObject): - _url = '/projects/fork/%(project_id)s' - canUpdate = False - canDelete = False - canList = False - canGet = False - requiredUrlAttrs = ['project_id'] - optionalCreateAttrs = ['namespace'] - - -class ProjectForkManager(BaseManager): - obj_cls = ProjectFork - - -class ProjectHook(GitlabObject): - _url = '/projects/%(project_id)s/hooks' - requiredUrlAttrs = ['project_id'] - requiredCreateAttrs = ['url'] - optionalCreateAttrs = ['push_events', 'issues_events', 'note_events', - 'merge_requests_events', 'tag_push_events', - 'build_events', 'enable_ssl_verification', 'token', - 'pipeline_events'] - shortPrintAttr = 'url' - - -class ProjectHookManager(BaseManager): - obj_cls = ProjectHook - - -class ProjectIssueNote(GitlabObject): - _url = '/projects/%(project_id)s/issues/%(issue_id)s/notes' - _constructorTypes = {'author': 'User'} - canDelete = False - requiredUrlAttrs = ['project_id', 'issue_id'] - requiredCreateAttrs = ['body'] - optionalCreateAttrs = ['created_at'] - - -class ProjectIssueNoteManager(BaseManager): - obj_cls = ProjectIssueNote - - -class ProjectIssue(GitlabObject): - _url = '/projects/%(project_id)s/issues/' - _constructorTypes = {'author': 'User', 'assignee': 'User', - 'milestone': 'ProjectMilestone'} - optionalListAttrs = ['state', 'labels', 'milestone', 'iid', 'order_by', - 'sort'] - requiredUrlAttrs = ['project_id'] - requiredCreateAttrs = ['title'] - optionalCreateAttrs = ['description', 'assignee_id', 'milestone_id', - 'labels', 'created_at', 'due_date'] - optionalUpdateAttrs = ['title', 'description', 'assignee_id', - 'milestone_id', 'labels', 'created_at', - 'updated_at', 'state_event', 'due_date'] - shortPrintAttr = 'title' - managers = ( - ('notes', ProjectIssueNoteManager, - [('project_id', 'project_id'), ('issue_id', 'id')]), - ) - - def subscribe(self, **kwargs): - """Subscribe to an issue. - - Raises: - GitlabConnectionError: If the server cannot be reached. - GitlabSubscribeError: If the subscription cannot be done - """ - url = ('/projects/%(project_id)s/issues/%(issue_id)s/subscription' % - {'project_id': self.project_id, 'issue_id': self.id}) - - r = self.gitlab._raw_post(url, **kwargs) - raise_error_from_response(r, GitlabSubscribeError) - self._set_from_dict(r.json()) - - def unsubscribe(self, **kwargs): - """Unsubscribe an issue. - - Raises: - GitlabConnectionError: If the server cannot be reached. - GitlabUnsubscribeError: If the unsubscription cannot be done - """ - url = ('/projects/%(project_id)s/issues/%(issue_id)s/subscription' % - {'project_id': self.project_id, 'issue_id': self.id}) - - r = self.gitlab._raw_delete(url, **kwargs) - raise_error_from_response(r, GitlabUnsubscribeError) - self._set_from_dict(r.json()) - - def move(self, to_project_id, **kwargs): - """Move the issue to another project. - - Raises: - GitlabConnectionError: If the server cannot be reached. - """ - url = ('/projects/%(project_id)s/issues/%(issue_id)s/move' % - {'project_id': self.project_id, 'issue_id': self.id}) - - data = {'to_project_id': to_project_id} - data.update(**kwargs) - r = self.gitlab._raw_post(url, data=data) - raise_error_from_response(r, GitlabUpdateError, 201) - self._set_from_dict(r.json()) - - def todo(self, **kwargs): - """Create a todo for the issue. - - Raises: - GitlabConnectionError: If the server cannot be reached. - """ - url = ('/projects/%(project_id)s/issues/%(issue_id)s/todo' % - {'project_id': self.project_id, 'issue_id': self.id}) - r = self.gitlab._raw_post(url, **kwargs) - raise_error_from_response(r, GitlabTodoError, [201, 304]) - - def time_stats(self, **kwargs): - """Get time stats for the issue. - - Raises: - GitlabConnectionError: If the server cannot be reached. - """ - url = ('/projects/%(project_id)s/issues/%(issue_id)s/time_stats' % - {'project_id': self.project_id, 'issue_id': self.id}) - r = self.gitlab._raw_get(url, **kwargs) - raise_error_from_response(r, GitlabGetError) - return r.json() - - def time_estimate(self, **kwargs): - """Set an estimated time of work for the issue. - - Raises: - GitlabConnectionError: If the server cannot be reached. - """ - url = ('/projects/%(project_id)s/issues/%(issue_id)s/time_estimate' % - {'project_id': self.project_id, 'issue_id': self.id}) - r = self.gitlab._raw_post(url, **kwargs) - raise_error_from_response(r, GitlabTimeTrackingError, 201) - return r.json() - - def reset_time_estimate(self, **kwargs): - """Resets estimated time for the issue to 0 seconds. - - Raises: - GitlabConnectionError: If the server cannot be reached. - """ - url = ('/projects/%(project_id)s/issues/%(issue_id)s/' - 'reset_time_estimate' % - {'project_id': self.project_id, 'issue_id': self.id}) - r = self.gitlab._raw_post(url, **kwargs) - raise_error_from_response(r, GitlabTimeTrackingError, 200) - return r.json() - - def add_spent_time(self, **kwargs): - """Set an estimated time of work for the issue. - - Raises: - GitlabConnectionError: If the server cannot be reached. - """ - url = ('/projects/%(project_id)s/issues/%(issue_id)s/' - 'add_spent_time' % - {'project_id': self.project_id, 'issue_id': self.id}) - r = self.gitlab._raw_post(url, **kwargs) - raise_error_from_response(r, GitlabTimeTrackingError, 200) - return r.json() - - def reset_spent_time(self, **kwargs): - """Set an estimated time of work for the issue. - - Raises: - GitlabConnectionError: If the server cannot be reached. - """ - url = ('/projects/%(project_id)s/issues/%(issue_id)s/' - 'reset_spent_time' % - {'project_id': self.project_id, 'issue_id': self.id}) - r = self.gitlab._raw_post(url, **kwargs) - raise_error_from_response(r, GitlabTimeTrackingError, 200) - return r.json() - - -class ProjectIssueManager(BaseManager): - obj_cls = ProjectIssue - - -class ProjectMember(GitlabObject): - _url = '/projects/%(project_id)s/members' - requiredUrlAttrs = ['project_id'] - requiredCreateAttrs = ['access_level', 'user_id'] - optionalCreateAttrs = ['expires_at'] - requiredUpdateAttrs = ['access_level'] - optionalCreateAttrs = ['expires_at'] - shortPrintAttr = 'username' - - -class ProjectMemberManager(BaseManager): - obj_cls = ProjectMember - - -class ProjectNote(GitlabObject): - _url = '/projects/%(project_id)s/notes' - _constructorTypes = {'author': 'User'} - canUpdate = False - canDelete = False - requiredUrlAttrs = ['project_id'] - requiredCreateAttrs = ['body'] - - -class ProjectNoteManager(BaseManager): - obj_cls = ProjectNote - - -class ProjectNotificationSettings(NotificationSettings): - _url = '/projects/%(project_id)s/notification_settings' - requiredUrlAttrs = ['project_id'] - - -class ProjectNotificationSettingsManager(BaseManager): - obj_cls = ProjectNotificationSettings - - -class ProjectTagRelease(GitlabObject): - _url = '/projects/%(project_id)s/repository/tags/%(tag_name)/release' - canDelete = False - canList = False - requiredUrlAttrs = ['project_id', 'tag_name'] - requiredCreateAttrs = ['description'] - shortPrintAttr = 'description' - - -class ProjectTag(GitlabObject): - _url = '/projects/%(project_id)s/repository/tags' - _constructorTypes = {'release': 'ProjectTagRelease', - 'commit': 'ProjectCommit'} - idAttr = 'name' - canGet = 'from_list' - canUpdate = False - requiredUrlAttrs = ['project_id'] - requiredCreateAttrs = ['tag_name', 'ref'] - optionalCreateAttrs = ['message'] - shortPrintAttr = 'name' - - def set_release_description(self, description): - """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. - - Raises: - GitlabConnectionError: If the server cannot be reached. - GitlabCreateError: If the server fails to create the release. - GitlabUpdateError: If the server fails to update the release. - """ - url = '/projects/%s/repository/tags/%s/release' % (self.project_id, - self.name) - if self.release is None: - r = self.gitlab._raw_post(url, data={'description': description}) - raise_error_from_response(r, GitlabCreateError, 201) - else: - r = self.gitlab._raw_put(url, data={'description': description}) - raise_error_from_response(r, GitlabUpdateError, 200) - self.release = ProjectTagRelease(self, r.json()) - - -class ProjectTagManager(BaseManager): - obj_cls = ProjectTag - - -class ProjectMergeRequestDiff(GitlabObject): - _url = ('/projects/%(project_id)s/merge_requests/' - '%(merge_request_id)s/versions') - canCreate = False - canUpdate = False - canDelete = False - requiredUrlAttrs = ['project_id', 'merge_request_id'] - - -class ProjectMergeRequestDiffManager(BaseManager): - obj_cls = ProjectMergeRequestDiff - - -class ProjectMergeRequestNote(GitlabObject): - _url = '/projects/%(project_id)s/merge_requests/%(merge_request_id)s/notes' - _constructorTypes = {'author': 'User'} - requiredUrlAttrs = ['project_id', 'merge_request_id'] - requiredCreateAttrs = ['body'] - - -class ProjectMergeRequestNoteManager(BaseManager): - obj_cls = ProjectMergeRequestNote - - -class ProjectMergeRequest(GitlabObject): - _url = '/projects/%(project_id)s/merge_requests' - _constructorTypes = {'author': 'User', 'assignee': 'User'} - requiredUrlAttrs = ['project_id'] - requiredCreateAttrs = ['source_branch', 'target_branch', 'title'] - optionalCreateAttrs = ['assignee_id', 'description', 'target_project_id', - 'labels', 'milestone_id', 'remove_source_branch'] - optionalUpdateAttrs = ['target_branch', 'assignee_id', 'title', - 'description', 'state_event', 'labels', - 'milestone_id'] - optionalListAttrs = ['iid', 'state', 'order_by', 'sort'] - - managers = ( - ('notes', ProjectMergeRequestNoteManager, - [('project_id', 'project_id'), ('merge_request_id', 'id')]), - ('diffs', ProjectMergeRequestDiffManager, - [('project_id', 'project_id'), ('merge_request_id', 'id')]), - ) - - def _data_for_gitlab(self, extra_parameters={}, update=False, - as_json=True): - data = (super(ProjectMergeRequest, self) - ._data_for_gitlab(extra_parameters, update=update, - as_json=False)) - if update: - # Drop source_branch attribute as it is not accepted by the gitlab - # server (Issue #76) - data.pop('source_branch', None) - return json.dumps(data) - - def subscribe(self, **kwargs): - """Subscribe to a MR. - - Raises: - GitlabConnectionError: If the server cannot be reached. - GitlabSubscribeError: If the subscription cannot be done - """ - url = ('/projects/%(project_id)s/merge_requests/%(mr_id)s/' - 'subscription' % - {'project_id': self.project_id, 'mr_id': self.id}) - - r = self.gitlab._raw_post(url, **kwargs) - raise_error_from_response(r, GitlabSubscribeError, [201, 304]) - if r.status_code == 201: - self._set_from_dict(r.json()) - - def unsubscribe(self, **kwargs): - """Unsubscribe a MR. - - Raises: - GitlabConnectionError: If the server cannot be reached. - GitlabUnsubscribeError: If the unsubscription cannot be done - """ - url = ('/projects/%(project_id)s/merge_requests/%(mr_id)s/' - 'subscription' % - {'project_id': self.project_id, 'mr_id': self.id}) - - r = self.gitlab._raw_delete(url, **kwargs) - raise_error_from_response(r, GitlabUnsubscribeError, [200, 304]) - if r.status_code == 200: - self._set_from_dict(r.json()) - - def cancel_merge_when_build_succeeds(self, **kwargs): - """Cancel merge when build succeeds.""" - - u = ('/projects/%s/merge_requests/%s/cancel_merge_when_build_succeeds' - % (self.project_id, self.id)) - r = self.gitlab._raw_put(u, **kwargs) - errors = {401: GitlabMRForbiddenError, - 405: GitlabMRClosedError, - 406: GitlabMROnBuildSuccessError} - raise_error_from_response(r, errors) - return ProjectMergeRequest(self, r.json()) - - def closes_issues(self, **kwargs): - """List issues closed by the MR. - - Returns: - list (ProjectIssue): List of closed issues - - Raises: - GitlabConnectionError: If the server cannot be reached. - GitlabGetError: If the server fails to perform the request. - """ - url = ('/projects/%s/merge_requests/%s/closes_issues' % - (self.project_id, self.id)) - return self.gitlab._raw_list(url, ProjectIssue, - {'project_id': self.project_id}, - **kwargs) - - def commits(self, **kwargs): - """List the merge request commits. - - Returns: - list (ProjectCommit): List of commits - - Raises: - GitlabConnectionError: If the server cannot be reached. - GitlabListError: If the server fails to perform the request. - """ - url = ('/projects/%s/merge_requests/%s/commits' % - (self.project_id, self.id)) - return self.gitlab._raw_list(url, ProjectCommit, - {'project_id': self.project_id}, - **kwargs) - - def changes(self, **kwargs): - """List the merge request changes. - - Returns: - list (dict): List of changes - - Raises: - GitlabConnectionError: If the server cannot be reached. - GitlabListError: If the server fails to perform the request. - """ - url = ('/projects/%s/merge_requests/%s/changes' % - (self.project_id, self.id)) - r = self.gitlab._raw_get(url, **kwargs) - raise_error_from_response(r, GitlabListError) - return r.json() - - def merge(self, merge_commit_message=None, - should_remove_source_branch=False, - merged_when_build_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 - merged_when_build_succeeds (bool): Wait for the build to succeed, - then merge - - Returns: - ProjectMergeRequest: The updated MR - Raises: - GitlabConnectionError: If the server cannot be reached. - GitlabMRForbiddenError: If the user doesn't have permission to - close thr MR - GitlabMRClosedError: If the MR is already closed - """ - url = '/projects/%s/merge_requests/%s/merge' % (self.project_id, - self.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 merged_when_build_succeeds: - data['merged_when_build_succeeds'] = True - - r = self.gitlab._raw_put(url, data=data, **kwargs) - errors = {401: GitlabMRForbiddenError, - 405: GitlabMRClosedError} - raise_error_from_response(r, errors) - self._set_from_dict(r.json()) - - def todo(self, **kwargs): - """Create a todo for the merge request. - - Raises: - GitlabConnectionError: If the server cannot be reached. - """ - url = ('/projects/%(project_id)s/merge_requests/%(mr_id)s/todo' % - {'project_id': self.project_id, 'mr_id': self.id}) - r = self.gitlab._raw_post(url, **kwargs) - raise_error_from_response(r, GitlabTodoError, [201, 304]) - - def time_stats(self, **kwargs): - """Get time stats for the merge request. - - Raises: - GitlabConnectionError: If the server cannot be reached. - """ - url = ('/projects/%(project_id)s/merge_requests/%(mr_id)s/time_stats' % - {'project_id': self.project_id, 'mr_id': self.id}) - r = self.gitlab._raw_get(url, **kwargs) - raise_error_from_response(r, GitlabGetError) - return r.json() - - def time_estimate(self, **kwargs): - """Set an estimated time of work for the merge request. - - Raises: - GitlabConnectionError: If the server cannot be reached. - """ - url = ('/projects/%(project_id)s/merge_requests/%(mr_id)s/' - 'time_estimate' % - {'project_id': self.project_id, 'mr_id': self.id}) - r = self.gitlab._raw_post(url, **kwargs) - raise_error_from_response(r, GitlabTimeTrackingError, 201) - return r.json() - - def reset_time_estimate(self, **kwargs): - """Resets estimated time for the merge request to 0 seconds. - - Raises: - GitlabConnectionError: If the server cannot be reached. - """ - url = ('/projects/%(project_id)s/merge_requests/%(mr_id)s/' - 'reset_time_estimate' % - {'project_id': self.project_id, 'mr_id': self.id}) - r = self.gitlab._raw_post(url, **kwargs) - raise_error_from_response(r, GitlabTimeTrackingError, 200) - return r.json() - - def add_spent_time(self, **kwargs): - """Set an estimated time of work for the merge request. - - Raises: - GitlabConnectionError: If the server cannot be reached. - """ - url = ('/projects/%(project_id)s/merge_requests/%(mr_id)s/' - 'add_spent_time' % - {'project_id': self.project_id, 'mr_id': self.id}) - r = self.gitlab._raw_post(url, **kwargs) - raise_error_from_response(r, GitlabTimeTrackingError, 200) - return r.json() - - def reset_spent_time(self, **kwargs): - """Set an estimated time of work for the merge request. - - Raises: - GitlabConnectionError: If the server cannot be reached. - """ - url = ('/projects/%(project_id)s/merge_requests/%(mr_id)s/' - 'reset_spent_time' % - {'project_id': self.project_id, 'mr_id': self.id}) - r = self.gitlab._raw_post(url, **kwargs) - raise_error_from_response(r, GitlabTimeTrackingError, 200) - return r.json() - - -class ProjectMergeRequestManager(BaseManager): - obj_cls = ProjectMergeRequest - - -class ProjectMilestone(GitlabObject): - _url = '/projects/%(project_id)s/milestones' - canDelete = False - requiredUrlAttrs = ['project_id'] - optionalListAttrs = ['iid', 'state'] - requiredCreateAttrs = ['title'] - optionalCreateAttrs = ['description', 'due_date', 'start_date', - 'state_event'] - optionalUpdateAttrs = requiredCreateAttrs + optionalCreateAttrs - shortPrintAttr = 'title' - - def issues(self, **kwargs): - url = "/projects/%s/milestones/%s/issues" % (self.project_id, self.id) - return self.gitlab._raw_list(url, ProjectIssue, - {'project_id': self.project_id}, - **kwargs) - - def merge_requests(self, **kwargs): - """List the merge requests related to this milestone - - Returns: - list (ProjectMergeRequest): List of merge requests - - Raises: - GitlabConnectionError: If the server cannot be reached. - GitlabListError: If the server fails to perform the request. - """ - url = ('/projects/%s/milestones/%s/merge_requests' % - (self.project_id, self.id)) - return self.gitlab._raw_list(url, ProjectMergeRequest, - {'project_id': self.project_id}, - **kwargs) - - -class ProjectMilestoneManager(BaseManager): - obj_cls = ProjectMilestone - - -class ProjectLabel(GitlabObject): - _url = '/projects/%(project_id)s/labels' - _id_in_delete_url = False - _id_in_update_url = False - canGet = 'from_list' - requiredUrlAttrs = ['project_id'] - idAttr = 'name' - requiredDeleteAttrs = ['name'] - requiredCreateAttrs = ['name', 'color'] - optionalCreateAttrs = ['description', 'priority'] - requiredUpdateAttrs = ['name'] - optionalUpdateAttrs = ['new_name', 'color', 'description', 'priority'] - - def subscribe(self, **kwargs): - """Subscribe to a label. - - Raises: - GitlabConnectionError: If the server cannot be reached. - GitlabSubscribeError: If the subscription cannot be done - """ - url = ('/projects/%(project_id)s/labels/%(label_id)s/subscription' % - {'project_id': self.project_id, 'label_id': self.name}) - - r = self.gitlab._raw_post(url, **kwargs) - raise_error_from_response(r, GitlabSubscribeError, [201, 304]) - self._set_from_dict(r.json()) - - def unsubscribe(self, **kwargs): - """Unsubscribe a label. - - Raises: - GitlabConnectionError: If the server cannot be reached. - GitlabUnsubscribeError: If the unsubscription cannot be done - """ - url = ('/projects/%(project_id)s/labels/%(label_id)s/subscription' % - {'project_id': self.project_id, 'label_id': self.name}) - - r = self.gitlab._raw_delete(url, **kwargs) - raise_error_from_response(r, GitlabUnsubscribeError, [200, 304]) - self._set_from_dict(r.json()) - - -class ProjectLabelManager(BaseManager): - obj_cls = ProjectLabel - - -class ProjectFile(GitlabObject): - _url = '/projects/%(project_id)s/repository/files' - canList = False - requiredUrlAttrs = ['project_id'] - requiredGetAttrs = ['file_path', 'ref'] - requiredCreateAttrs = ['file_path', 'branch_name', 'content', - 'commit_message'] - optionalCreateAttrs = ['encoding'] - requiredDeleteAttrs = ['branch_name', 'commit_message', 'file_path'] - shortPrintAttr = 'file_path' - getRequiresId = False - - def decode(self): - """Returns the decoded content of the file. - - Returns: - (str): the decoded content. - """ - return base64.b64decode(self.content) - - -class ProjectFileManager(BaseManager): - obj_cls = ProjectFile - - -class ProjectPipeline(GitlabObject): - _url = '/projects/%(project_id)s/pipelines' - _create_url = '/projects/%(project_id)s/pipeline' - - canUpdate = False - canDelete = False - - requiredUrlAttrs = ['project_id'] - requiredCreateAttrs = ['ref'] - - def retry(self, **kwargs): - """Retries failed builds in a pipeline. - - Raises: - GitlabConnectionError: If the server cannot be reached. - GitlabPipelineRetryError: If the retry cannot be done. - """ - url = ('/projects/%(project_id)s/pipelines/%(id)s/retry' % - {'project_id': self.project_id, 'id': self.id}) - r = self.gitlab._raw_post(url, data=None, content_type=None, **kwargs) - raise_error_from_response(r, GitlabPipelineRetryError, 201) - self._set_from_dict(r.json()) - - def cancel(self, **kwargs): - """Cancel builds in a pipeline. - - Raises: - GitlabConnectionError: If the server cannot be reached. - GitlabPipelineCancelError: If the retry cannot be done. - """ - url = ('/projects/%(project_id)s/pipelines/%(id)s/cancel' % - {'project_id': self.project_id, 'id': self.id}) - r = self.gitlab._raw_post(url, data=None, content_type=None, **kwargs) - raise_error_from_response(r, GitlabPipelineRetryError, 200) - self._set_from_dict(r.json()) - - -class ProjectPipelineManager(BaseManager): - obj_cls = ProjectPipeline - - -class ProjectSnippetNote(GitlabObject): - _url = '/projects/%(project_id)s/snippets/%(snippet_id)s/notes' - _constructorTypes = {'author': 'User'} - canUpdate = False - canDelete = False - requiredUrlAttrs = ['project_id', 'snippet_id'] - requiredCreateAttrs = ['body'] - - -class ProjectSnippetNoteManager(BaseManager): - obj_cls = ProjectSnippetNote - - -class ProjectSnippet(GitlabObject): - _url = '/projects/%(project_id)s/snippets' - _constructorTypes = {'author': 'User'} - requiredUrlAttrs = ['project_id'] - requiredCreateAttrs = ['title', 'file_name', 'code'] - optionalCreateAttrs = ['lifetime', 'visibility_level'] - optionalUpdateAttrs = ['title', 'file_name', 'code', 'visibility_level'] - shortPrintAttr = 'title' - managers = ( - ('notes', ProjectSnippetNoteManager, - [('project_id', 'project_id'), ('snippet_id', 'id')]), - ) - - def content(self, streamed=False, action=None, chunk_size=1024, **kwargs): - """Return the raw 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. - - Returns: - str: The snippet content - - Raises: - GitlabConnectionError: If the server cannot be reached. - GitlabGetError: If the server fails to perform the request. - """ - url = ("/projects/%(project_id)s/snippets/%(snippet_id)s/raw" % - {'project_id': self.project_id, 'snippet_id': self.id}) - r = self.gitlab._raw_get(url, **kwargs) - raise_error_from_response(r, GitlabGetError) - return utils.response_content(r, streamed, action, chunk_size) - - -class ProjectSnippetManager(BaseManager): - obj_cls = ProjectSnippet - - -class ProjectTrigger(GitlabObject): - _url = '/projects/%(project_id)s/triggers' - canUpdate = False - idAttr = 'token' - requiredUrlAttrs = ['project_id'] - - -class ProjectTriggerManager(BaseManager): - obj_cls = ProjectTrigger - - -class ProjectVariable(GitlabObject): - _url = '/projects/%(project_id)s/variables' - idAttr = 'key' - requiredUrlAttrs = ['project_id'] - requiredCreateAttrs = ['key', 'value'] - - -class ProjectVariableManager(BaseManager): - obj_cls = ProjectVariable - - -class ProjectService(GitlabObject): - _url = '/projects/%(project_id)s/services/%(service_name)s' - canList = False - canCreate = False - _id_in_update_url = False - _id_in_delete_url = False - getRequiresId = False - requiredUrlAttrs = ['project_id', 'service_name'] - - _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': (tuple(), ( - # Required fields in GitLab >= 8.14 - 'url', 'project_key', - - # Required fields in GitLab < 8.14 - 'new_issue_url', 'project_url', 'issues_url', 'api_url', - 'description', - - # Optional fields - 'username', 'password', 'jira_issue_transition_id')), - '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 _data_for_gitlab(self, extra_parameters={}, update=False, - as_json=True): - data = (super(ProjectService, self) - ._data_for_gitlab(extra_parameters, update=update, - as_json=False)) - missing = [] - # Mandatory args - for attr in self._service_attrs[self.service_name][0]: - if not hasattr(self, attr): - missing.append(attr) - else: - data[attr] = getattr(self, attr) - - if missing: - raise GitlabUpdateError('Missing attribute(s): %s' % - ", ".join(missing)) - - # Optional args - for attr in self._service_attrs[self.service_name][1]: - if hasattr(self, attr): - data[attr] = getattr(self, attr) - - return json.dumps(data) - - -class ProjectServiceManager(BaseManager): - obj_cls = ProjectService - - def available(self, **kwargs): - """List the services known by python-gitlab. - - Returns: - list (str): The list of service code names. - """ - return list(ProjectService._service_attrs.keys()) - - -class ProjectAccessRequest(GitlabObject): - _url = '/projects/%(project_id)s/access_requests' - canGet = 'from_list' - canUpdate = False - - def approve(self, access_level=gitlab.DEVELOPER_ACCESS, **kwargs): - """Approve an access request. - - Attrs: - access_level (int): The access level for the user. - - Raises: - GitlabConnectionError: If the server cannot be reached. - GitlabUpdateError: If the server fails to perform the request. - """ - - url = ('/projects/%(project_id)s/access_requests/%(id)s/approve' % - {'project_id': self.project_id, 'id': self.id}) - data = {'access_level': access_level} - r = self.gitlab._raw_put(url, data=data, **kwargs) - raise_error_from_response(r, GitlabUpdateError, 201) - self._set_from_dict(r.json()) - - -class ProjectAccessRequestManager(BaseManager): - obj_cls = ProjectAccessRequest - - -class ProjectDeployment(GitlabObject): - _url = '/projects/%(project_id)s/deployments' - canCreate = False - canUpdate = False - canDelete = False - - -class ProjectDeploymentManager(BaseManager): - obj_cls = ProjectDeployment - - -class ProjectRunner(GitlabObject): - _url = '/projects/%(project_id)s/runners' - canUpdate = False - requiredCreateAttrs = ['runner_id'] - - -class ProjectRunnerManager(BaseManager): - obj_cls = ProjectRunner - - -class Project(GitlabObject): - _url = '/projects' - _constructorTypes = {'owner': 'User', 'namespace': 'Group'} - optionalListAttrs = ['search'] - requiredCreateAttrs = ['name'] - optionalListAttrs = ['search'] - optionalCreateAttrs = ['path', 'namespace_id', 'description', - 'issues_enabled', 'merge_requests_enabled', - 'builds_enabled', 'wiki_enabled', - 'snippets_enabled', 'container_registry_enabled', - 'shared_runners_enabled', 'public', - 'visibility_level', 'import_url', 'public_builds', - 'only_allow_merge_if_build_succeeds', - 'only_allow_merge_if_all_discussions_are_resolved', - 'lfs_enabled', 'request_access_enabled'] - optionalUpdateAttrs = ['name', 'path', 'default_branch', 'description', - 'issues_enabled', 'merge_requests_enabled', - 'builds_enabled', 'wiki_enabled', - 'snippets_enabled', 'container_registry_enabled', - 'shared_runners_enabled', 'public', - 'visibility_level', 'import_url', 'public_builds', - 'only_allow_merge_if_build_succeeds', - 'only_allow_merge_if_all_discussions_are_resolved', - 'lfs_enabled', 'request_access_enabled'] - shortPrintAttr = 'path' - managers = ( - ('accessrequests', ProjectAccessRequestManager, - [('project_id', 'id')]), - ('boards', ProjectBoardManager, [('project_id', 'id')]), - ('board_lists', ProjectBoardListManager, [('project_id', 'id')]), - ('branches', ProjectBranchManager, [('project_id', 'id')]), - ('builds', ProjectBuildManager, [('project_id', 'id')]), - ('commits', ProjectCommitManager, [('project_id', 'id')]), - ('deployments', ProjectDeploymentManager, [('project_id', 'id')]), - ('environments', ProjectEnvironmentManager, [('project_id', 'id')]), - ('events', ProjectEventManager, [('project_id', 'id')]), - ('files', ProjectFileManager, [('project_id', 'id')]), - ('forks', ProjectForkManager, [('project_id', 'id')]), - ('hooks', ProjectHookManager, [('project_id', 'id')]), - ('keys', ProjectKeyManager, [('project_id', 'id')]), - ('issues', ProjectIssueManager, [('project_id', 'id')]), - ('labels', ProjectLabelManager, [('project_id', 'id')]), - ('members', ProjectMemberManager, [('project_id', 'id')]), - ('mergerequests', ProjectMergeRequestManager, [('project_id', 'id')]), - ('milestones', ProjectMilestoneManager, [('project_id', 'id')]), - ('notes', ProjectNoteManager, [('project_id', 'id')]), - ('notificationsettings', ProjectNotificationSettingsManager, - [('project_id', 'id')]), - ('pipelines', ProjectPipelineManager, [('project_id', 'id')]), - ('runners', ProjectRunnerManager, [('project_id', 'id')]), - ('services', ProjectServiceManager, [('project_id', 'id')]), - ('snippets', ProjectSnippetManager, [('project_id', 'id')]), - ('tags', ProjectTagManager, [('project_id', 'id')]), - ('triggers', ProjectTriggerManager, [('project_id', 'id')]), - ('variables', ProjectVariableManager, [('project_id', 'id')]), - ) - - VISIBILITY_PRIVATE = gitlab.VISIBILITY_PRIVATE - VISIBILITY_INTERNAL = gitlab.VISIBILITY_INTERNAL - VISIBILITY_PUBLIC = gitlab.VISIBILITY_PUBLIC - - def repository_tree(self, path='', ref_name='', **kwargs): - """Return a list of files in the repository. - - Args: - path (str): Path of the top folder (/ by default) - ref_name (str): Reference to a commit or branch - - Returns: - str: The json representation of the tree. - - Raises: - GitlabConnectionError: If the server cannot be reached. - GitlabGetError: If the server fails to perform the request. - """ - url = "/projects/%s/repository/tree" % (self.id) - params = [] - if path: - params.append(urllib.urlencode({'path': path})) - if ref_name: - params.append("ref_name=%s" % ref_name) - if params: - url += '?' + "&".join(params) - r = self.gitlab._raw_get(url, **kwargs) - raise_error_from_response(r, GitlabGetError) - return r.json() - - def repository_blob(self, sha, filepath, streamed=False, action=None, - chunk_size=1024, **kwargs): - """Return the content of a file for a commit. - - Args: - sha (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. - - Returns: - str: The file content - - Raises: - GitlabConnectionError: If the server cannot be reached. - GitlabGetError: If the server fails to perform the request. - """ - url = "/projects/%s/repository/blobs/%s" % (self.id, sha) - url += '?%s' % (urllib.urlencode({'filepath': filepath})) - r = self.gitlab._raw_get(url, streamed=streamed, **kwargs) - raise_error_from_response(r, GitlabGetError) - return utils.response_content(r, streamed, action, chunk_size) - - def repository_raw_blob(self, sha, streamed=False, action=None, - chunk_size=1024, **kwargs): - """Returns the raw file contents for a blob by blob SHA. - - 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. - - Returns: - str: The blob content - - Raises: - GitlabConnectionError: If the server cannot be reached. - GitlabGetError: If the server fails to perform the request. - """ - url = "/projects/%s/repository/raw_blobs/%s" % (self.id, sha) - r = self.gitlab._raw_get(url, streamed=streamed, **kwargs) - raise_error_from_response(r, GitlabGetError) - return utils.response_content(r, streamed, action, chunk_size) - - def repository_compare(self, from_, to, **kwargs): - """Returns a diff between two branches/commits. - - Args: - from_(str): orig branch/SHA - to(str): dest branch/SHA - - Returns: - str: The diff - - Raises: - GitlabConnectionError: If the server cannot be reached. - GitlabGetError: If the server fails to perform the request. - """ - url = "/projects/%s/repository/compare" % self.id - url = "%s?from=%s&to=%s" % (url, from_, to) - r = self.gitlab._raw_get(url, **kwargs) - raise_error_from_response(r, GitlabGetError) - return r.json() - - def repository_contributors(self): - """Returns a list of contributors for the project. - - Returns: - list: The contibutors - - Raises: - GitlabConnectionError: If the server cannot be reached. - GitlabGetError: If the server fails to perform the request. - """ - url = "/projects/%s/repository/contributors" % self.id - r = self.gitlab._raw_get(url) - raise_error_from_response(r, GitlabListError) - return r.json() - - 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. - - Returns: - str: The binary data of the archive. - - Raises: - GitlabConnectionError: If the server cannot be reached. - GitlabGetError: If the server fails to perform the request. - """ - url = '/projects/%s/repository/archive' % self.id - if sha: - url += '?sha=%s' % sha - r = self.gitlab._raw_get(url, streamed=streamed, **kwargs) - raise_error_from_response(r, GitlabGetError) - return utils.response_content(r, streamed, action, chunk_size) - - def create_fork_relation(self, forked_from_id): - """Create a forked from/to relation between existing projects. - - Args: - forked_from_id (int): The ID of the project that was forked from - - Raises: - GitlabConnectionError: If the server cannot be reached. - GitlabCreateError: If the server fails to perform the request. - """ - url = "/projects/%s/fork/%s" % (self.id, forked_from_id) - r = self.gitlab._raw_post(url) - raise_error_from_response(r, GitlabCreateError, 201) - - def delete_fork_relation(self): - """Delete a forked relation between existing projects. - - Raises: - GitlabConnectionError: If the server cannot be reached. - GitlabDeleteError: If the server fails to perform the request. - """ - url = "/projects/%s/fork" % self.id - r = self.gitlab._raw_delete(url) - raise_error_from_response(r, GitlabDeleteError) - - def star(self, **kwargs): - """Star a project. - - Returns: - Project: the updated Project - - Raises: - GitlabCreateError: If the action cannot be done - GitlabConnectionError: If the server cannot be reached. - """ - url = "/projects/%s/star" % self.id - r = self.gitlab._raw_post(url, **kwargs) - raise_error_from_response(r, GitlabCreateError, [201, 304]) - return Project(self.gitlab, r.json()) if r.status_code == 201 else self - - def unstar(self, **kwargs): - """Unstar a project. - - Returns: - Project: the updated Project - - Raises: - GitlabDeleteError: If the action cannot be done - GitlabConnectionError: If the server cannot be reached. - """ - url = "/projects/%s/star" % self.id - r = self.gitlab._raw_delete(url, **kwargs) - raise_error_from_response(r, GitlabDeleteError, [200, 304]) - return Project(self.gitlab, r.json()) if r.status_code == 200 else self - - def archive(self, **kwargs): - """Archive a project. - - Returns: - Project: the updated Project - - Raises: - GitlabCreateError: If the action cannot be done - GitlabConnectionError: If the server cannot be reached. - """ - url = "/projects/%s/archive" % self.id - r = self.gitlab._raw_post(url, **kwargs) - raise_error_from_response(r, GitlabCreateError, 201) - return Project(self.gitlab, r.json()) if r.status_code == 201 else self - - def archive_(self, **kwargs): - warnings.warn("`archive_()` is deprecated, use `archive()` instead", - DeprecationWarning) - return self.archive(**kwargs) - - def unarchive(self, **kwargs): - """Unarchive a project. - - Returns: - Project: the updated Project - - Raises: - GitlabDeleteError: If the action cannot be done - GitlabConnectionError: If the server cannot be reached. - """ - url = "/projects/%s/unarchive" % self.id - r = self.gitlab._raw_delete(url, **kwargs) - raise_error_from_response(r, GitlabCreateError, 201) - return Project(self.gitlab, r.json()) if r.status_code == 201 else self - - def unarchive_(self, **kwargs): - warnings.warn("`unarchive_()` is deprecated, " - "use `unarchive()` instead", - DeprecationWarning) - return self.unarchive(**kwargs) - - def share(self, group_id, group_access, **kwargs): - """Share the project with a group. - - Args: - group_id (int): ID of the group. - group_access (int): Access level for the group. - - Raises: - GitlabConnectionError: If the server cannot be reached. - GitlabCreateError: If the server fails to perform the request. - """ - url = "/projects/%s/share" % self.id - data = {'group_id': group_id, 'group_access': group_access} - r = self.gitlab._raw_post(url, data=data, **kwargs) - raise_error_from_response(r, GitlabCreateError, 201) - - def trigger_build(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 commit SHA, a branch name, ... - token (str): The trigger token - variables (dict): Variables passed to the build script - - Raises: - GitlabConnectionError: If the server cannot be reached. - GitlabCreateError: If the server fails to perform the request. - """ - url = "/projects/%s/trigger/builds" % self.id - form = {r'variables[%s]' % k: v for k, v in six.iteritems(variables)} - data = {'ref': ref, 'token': token} - data.update(form) - r = self.gitlab._raw_post(url, data=data, **kwargs) - raise_error_from_response(r, GitlabCreateError, 201) - - -class Runner(GitlabObject): - _url = '/runners' - canCreate = False - optionalUpdateAttrs = ['description', 'active', 'tag_list'] - optionalListAttrs = ['scope'] - - -class RunnerManager(BaseManager): - obj_cls = Runner - - 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 - - Returns: - list(Runner): a list of runners matching the scope. - - Raises: - GitlabConnectionError: If the server cannot be reached. - GitlabListError: If the resource cannot be found - """ - url = '/runners/all' - if scope is not None: - url += '?scope=' + scope - return self.gitlab._raw_list(url, self.obj_cls, **kwargs) - - -class TeamMember(GitlabObject): - _url = '/user_teams/%(team_id)s/members' - canUpdate = False - requiredUrlAttrs = ['teamd_id'] - requiredCreateAttrs = ['access_level'] - shortPrintAttr = 'username' - - -class Todo(GitlabObject): - _url = '/todos' - canGet = 'from_list' - canUpdate = False - canCreate = False - optionalListAttrs = ['action', 'author_id', 'project_id', 'state', 'type'] - - -class TodoManager(BaseManager): - obj_cls = Todo - - def delete_all(self, **kwargs): - """Mark all the todos as done. - - Raises: - GitlabConnectionError: If the server cannot be reached. - GitlabDeleteError: If the resource cannot be found - - Returns: - The number of todos maked done. - """ - url = '/todos' - r = self.gitlab._raw_delete(url, **kwargs) - raise_error_from_response(r, GitlabDeleteError) - return int(r.text) - - -class ProjectManager(BaseManager): - obj_cls = Project - - def search(self, query, **kwargs): - """Search projects by name. - - API v3 only. - - .. note:: - - The search is only performed on the project name (not on the - namespace or the description). To perform a smarter search, use the - ``search`` argument of the ``list()`` method: - - .. code-block:: python - - gl.projects.list(search=your_search_string) - - Args: - query (str): The query string to send to GitLab for the search. - all (bool): If True, return all the items, without pagination - **kwargs: Additional arguments to send to GitLab. - - Returns: - list(gitlab.Gitlab.Project): A list of matching projects. - """ - if self.gitlab.api_version == '4': - raise NotImplementedError("Not supported by v4 API") - - return self.gitlab._raw_list("/projects/search/" + query, Project, - **kwargs) - - def all(self, **kwargs): - """List all the projects (need admin rights). - - Args: - all (bool): If True, return all the items, without pagination - **kwargs: Additional arguments to send to GitLab. - - Returns: - list(gitlab.Gitlab.Project): The list of projects. - """ - return self.gitlab._raw_list("/projects/all", Project, **kwargs) - - def owned(self, **kwargs): - """List owned projects. - - Args: - all (bool): If True, return all the items, without pagination - **kwargs: Additional arguments to send to GitLab. - - Returns: - list(gitlab.Gitlab.Project): The list of owned projects. - """ - return self.gitlab._raw_list("/projects/owned", Project, **kwargs) - - def starred(self, **kwargs): - """List starred projects. - - Args: - all (bool): If True, return all the items, without pagination - **kwargs: Additional arguments to send to GitLab. - - Returns: - list(gitlab.Gitlab.Project): The list of starred projects. - """ - return self.gitlab._raw_list("/projects/starred", Project, **kwargs) - - -class GroupProject(Project): - _url = '/groups/%(group_id)s/projects' - canGet = 'from_list' - canCreate = False - canDelete = False - canUpdate = False - optionalListAttrs = ['archived', 'visibility', 'order_by', 'sort', - 'search', 'ci_enabled_first'] - - def __init__(self, *args, **kwargs): - Project.__init__(self, *args, **kwargs) - - -class GroupProjectManager(ProjectManager): - obj_cls = GroupProject - - -class Group(GitlabObject): - _url = '/groups' - requiredCreateAttrs = ['name', 'path'] - optionalCreateAttrs = ['description', 'visibility_level', 'parent_id', - 'lfs_enabled', 'request_access_enabled'] - optionalUpdateAttrs = ['name', 'path', 'description', 'visibility_level', - 'lfs_enabled', 'request_access_enabled'] - shortPrintAttr = 'name' - managers = ( - ('accessrequests', GroupAccessRequestManager, [('group_id', 'id')]), - ('members', GroupMemberManager, [('group_id', 'id')]), - ('notificationsettings', GroupNotificationSettingsManager, - [('group_id', 'id')]), - ('projects', GroupProjectManager, [('group_id', 'id')]), - ('issues', GroupIssueManager, [('group_id', 'id')]), - ) - - GUEST_ACCESS = gitlab.GUEST_ACCESS - REPORTER_ACCESS = gitlab.REPORTER_ACCESS - DEVELOPER_ACCESS = gitlab.DEVELOPER_ACCESS - MASTER_ACCESS = gitlab.MASTER_ACCESS - OWNER_ACCESS = gitlab.OWNER_ACCESS - - VISIBILITY_PRIVATE = gitlab.VISIBILITY_PRIVATE - VISIBILITY_INTERNAL = gitlab.VISIBILITY_INTERNAL - VISIBILITY_PUBLIC = gitlab.VISIBILITY_PUBLIC - - def transfer_project(self, id, **kwargs): - """Transfers a project to this new groups. - - Attrs: - id (int): ID of the project to transfer. - - Raises: - GitlabConnectionError: If the server cannot be reached. - GitlabTransferProjectError: If the server fails to perform the - request. - """ - url = '/groups/%d/projects/%d' % (self.id, id) - r = self.gitlab._raw_post(url, None, **kwargs) - raise_error_from_response(r, GitlabTransferProjectError, 201) - - -class GroupManager(BaseManager): - obj_cls = Group - - def search(self, query, **kwargs): - """Searches groups by name. - - Args: - query (str): The search string - all (bool): If True, return all the items, without pagination - - Returns: - list(Group): a list of matching groups. - """ - url = '/groups?search=' + query - return self.gitlab._raw_list(url, self.obj_cls, **kwargs) - - -class TeamMemberManager(BaseManager): - obj_cls = TeamMember - - -class TeamProject(GitlabObject): - _url = '/user_teams/%(team_id)s/projects' - _constructorTypes = {'owner': 'User', 'namespace': 'Group'} - canUpdate = False - requiredCreateAttrs = ['greatest_access_level'] - requiredUrlAttrs = ['team_id'] - shortPrintAttr = 'name' - - -class TeamProjectManager(BaseManager): - obj_cls = TeamProject - - -class Team(GitlabObject): - _url = '/user_teams' - shortPrintAttr = 'name' - requiredCreateAttrs = ['name', 'path'] - canUpdate = False - managers = ( - ('members', TeamMemberManager, [('team_id', 'id')]), - ('projects', TeamProjectManager, [('team_id', 'id')]), - ) - - -class TeamManager(BaseManager): - obj_cls = Team |