def default(self, obj):

if isinstance(obj, GitlabObject):

return obj.as_dict()

elif isinstance(obj, gitlab.Gitlab):

return {'url': obj._url}

"""Base manager class for API operations.

obj_cls = None

def __init__(self, gl, parent=None, args=[]):

"""Constructs a manager.

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 = self._set_parent_args(**kwargs)

if not self.obj_cls.canGet:

raise NotImplementedError

return self.obj_cls.get(self.gitlab, id, **args)

def list(self, **kwargs):

"""Get a list of GitLab objects.

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 = 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 = self._set_parent_args(**kwargs)

if not self.obj_cls.canDelete:

raise NotImplementedError

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

if isinstance(value, list):

value = ",".join(value)

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

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.

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

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.

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

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

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.

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

"""Manager for the Sidekiq methods.

def __init__(self, gl):

"""Constructs a Sidekiq manager.

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

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

canUpdate = False

shortPrintAttr = 'email'

requiredUrlAttrs = ['user_id']

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

canGet = 'from_list'

canUpdate = False

requiredUrlAttrs = ['user_id']

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

_url = '/users'

shortPrintAttr = 'username'

requiredCreateAttrs = ['email', 'username', 'name', 'password']

optionalCreateAttrs = ['skype', 'linkedin', 'twitter', 'projects_limit',

'extern_uid', 'provider', 'bio', 'admin',

'can_create_group', 'website_url', 'confirm',

'external']

requiredUpdateAttrs = ['email', 'username', 'name']

optionalUpdateAttrs = ['password', 'skype', 'linkedin', 'twitter',

'projects_limit', 'extern_uid', 'provider', 'bio',

'admin', 'can_create_group', 'website_url',

'confirm', 'external']

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

obj_cls = User

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

"""Search users.

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.

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:

_url = '/user/emails'

canUpdate = False

shortPrintAttr = 'email'

_url = '/user/keys'

canUpdate = False

shortPrintAttr = 'title'

_url = '/user'

canList = False

canCreate = False

canUpdate = False

canDelete = False

shortPrintAttr = 'username'

managers = (

('emails', CurrentUserEmailManager, [('user_id', 'id')]),

('keys', CurrentUserKeyManager, [('user_id', 'id')]),

_url = '/application/settings'

_id_in_update_url = 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

_url = '/broadcast_messages'

requiredCreateAttrs = ['message']

optionalCreateAttrs = ['starts_at', 'ends_at', 'color', 'font']

requiredUpdateAttrs = []

_url = '/deploy_keys'

canGet = 'from_list'

canCreate = False

canUpdate = False

_url = '/notification_settings'

_id_in_update_url = 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

_url = '/templates/gitignores'

canDelete = False

canUpdate = False

canCreate = False

_url = '/templates/gitlab_ci_ymls'

canDelete = False

canUpdate = False

canCreate = False

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

canGet = 'from_list'

canCreate = False

canUpdate = False

canDelete = False

requiredUrlAttrs = ['group_id']

_url = '/groups/%(group_id)s/members'

canGet = 'from_list'

requiredUrlAttrs = ['group_id']

requiredCreateAttrs = ['access_level', 'user_id']

requiredUpdateAttrs = ['access_level']

shortPrintAttr = 'username'

def _update(self, **kwargs):

self.user_id = self.id

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

canGet = 'from_list'

canCreate = False

canDelete = False

canUpdate = False

optionalListAttrs = ['archived', 'visibility', 'order_by', 'sort',

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

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)

_url = '/groups'

_constructorTypes = {'projects': 'Project'}

requiredCreateAttrs = ['name', 'path']

optionalCreateAttrs = ['description', 'visibility_level']

optionalUpdateAttrs = ['name', 'path', 'description', 'visibility_level']

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.

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

r = self.gitlab._raw_post(url, None, **kwargs)

obj_cls = Group

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

"""Searches groups by name.

url = '/groups?search=' + query

_url = '/hooks'

canUpdate = False

requiredCreateAttrs = ['url']

_url = '/issues'

_constructorTypes = {'author': 'User', 'assignee': 'User',

'milestone': 'ProjectMilestone'}

canGet = 'from_list'

canDelete = False

canUpdate = False

canCreate = False

shortPrintAttr = 'title'