def default(self, obj):

if isinstance(obj, GitlabObject):

return obj.__dict__

elif isinstance(obj, Gitlab):

return {'url': obj._url}

"""Represents a GitLab server connection"""

def __init__(self, url, private_token=None,

email=None, password=None, ssl_verify=True):

"""Stores informations about the server

self._url = '%s/api/v3' % url

self.setToken(private_token)

self.email = email

self.password = password

self.ssl_verify = ssl_verify

def auth(self):

"""Performs an authentication using either the private token, or the

if self.private_token:

self.token_auth()

else:

self.credentials_auth()

def credentials_auth(self):

if not self.email or not self.password:

raise GitlabAuthenticationError("Missing email/password")

r = self.rawPost('/session',

{'email': self.email, 'password': self.password})

if r.status_code == 201:

self.user = CurrentUser(self, r.json())

else:

raise GitlabAuthenticationError(r.json()['message'])

self.setToken(self.user.private_token)

def token_auth(self):

self.user = CurrentUser(self)

def setUrl(self, url):

"""Updates the gitlab URL"""

self._url = '%s/api/v3' % url

def setToken(self, token):

"""Sets the private token for authentication"""

self.private_token = token if token else None

self.headers = {"PRIVATE-TOKEN": token} if token else None

def setCredentials(self, email, password):

"""Sets the email/login and password for authentication"""

self.email = email

self.password = password

def rawGet(self, path, **kwargs):

url = '%s%s' % (self._url, path)

if kwargs:

url += "?%s" % ("&".join(

["%s=%s" % (k, v) for k, v in kwargs.items()]))

try:

return requests.get(url,

headers=self.headers,

verify=self.ssl_verify)

except:

raise GitlabConnectionError(

"Can't connect to GitLab server (%s)" % self._url)

def rawPost(self, path, data=None):

url = '%s%s' % (self._url, path)

try:

return requests.post(url, data,

headers=self.headers,

verify=self.ssl_verify)

except:

raise GitlabConnectionError(

"Can't connect to GitLab server (%s)" % self._url)

def rawPut(self, path):

url = '%s%s' % (self._url, path)

try:

return requests.put(url,

headers=self.headers,

verify=self.ssl_verify)

except:

raise GitlabConnectionError(

"Can't connect to GitLab server (%s)" % self._url)

def rawDelete(self, path):

url = '%s%s' % (self._url, path)

try:

return requests.delete(url,

headers=self.headers,

verify=self.ssl_verify)

except:

raise GitlabConnectionError(

"Can't connect to GitLab server (%s)" % self._url)

def list(self, obj_class, **kwargs):

missing = []

for k in obj_class.requiredListAttrs:

if k not in kwargs:

missing.append(k)

if missing:

raise GitlabListError('Missing attribute(s): %s' %

", ".join(missing))

url = obj_class._url % kwargs

url = '%s%s' % (self._url, url)

if kwargs:

url += "?%s" % ("&".join(

["%s=%s" % (k, v) for k, v in kwargs.items()]))

try:

r = requests.get(url, headers=self.headers, verify=self.ssl_verify)

except:

raise GitlabConnectionError(

"Can't connect to GitLab server (%s)" % self._url)

if r.status_code == 200:

cls = obj_class

if obj_class._returnClass:

cls = obj_class._returnClass

l = [cls(self, item) for item in r.json() if item is not None]

if kwargs:

for k, v in kwargs.items():

if k in ('page', 'per_page'):

continue

for obj in l:

obj.__dict__[k] = str(v)

return l

elif r.status_code == 401:

raise GitlabAuthenticationError(r.json()['message'])

else:

raise GitlabGetError('%d: %s' % (r.status_code, r.text))

def get(self, obj_class, id=None, **kwargs):

missing = []

for k in obj_class.requiredGetAttrs:

if k not in kwargs:

missing.append(k)

if missing:

raise GitlabListError('Missing attribute(s): %s' %

", ".join(missing))

url = obj_class._url % kwargs

if id is not None:

url = '%s%s/%s' % (self._url, url, str(id))

else:

url = '%s%s' % (self._url, url)

try:

r = requests.get(url, headers=self.headers, verify=self.ssl_verify)

except:

raise GitlabConnectionError(

"Can't connect to GitLab server (%s)" % self._url)

if r.status_code == 200:

return r.json()

elif r.status_code == 401:

raise GitlabAuthenticationError(r.json()['message'])

elif r.status_code == 404:

raise GitlabGetError("Object doesn't exist")

else:

raise GitlabGetError('%d: %s' % (r.status_code, r.text))

def delete(self, obj):

url = obj._url % obj.__dict__

url = '%s%s/%s' % (self._url, url, str(obj.id))

try:

r = requests.delete(url,

headers=self.headers,

verify=self.ssl_verify)

except:

raise GitlabConnectionError(

"Can't connect to GitLab server (%s)" % self._url)

if r.status_code == 200:

return True

elif r.status_code == 401:

raise GitlabAuthenticationError(r.json()['message'])

else:

raise GitlabDeleteError(r.json()['message'])

return False

def create(self, obj):

missing = []

for k in obj.requiredCreateAttrs:

if k not in obj.__dict__:

missing.append(k)

if missing:

raise GitlabCreateError('Missing attribute(s): %s' %

", ".join(missing))

url = obj._url % obj.__dict__

url = '%s%s' % (self._url, url)

try:

r = requests.post(url, obj.__dict__,

headers=self.headers,

verify=self.ssl_verify)

except:

raise GitlabConnectionError(

"Can't connect to GitLab server (%s)" % self._url)

if r.status_code == 201:

return r.json()

elif r.status_code == 401:

raise GitlabAuthenticationError(r.json()['message'])

else:

raise GitlabCreateError('%d: %s' % (r.status_code, r.text))

def update(self, obj):

url = obj._url % obj.__dict__

url = '%s%s/%s' % (self._url, url, str(obj.id))

# build a dict of data that can really be sent to server

d = {}

for k, v in obj.__dict__.items():

if type(v) in (int, str, bool):

d[k] = str(v)

elif type(v) == unicode:

d[k] = str(v.encode(sys.stdout.encoding, "replace"))

try:

r = requests.put(url, d,

headers=self.headers,

verify=self.ssl_verify)

except:

raise GitlabConnectionError(

"Can't connect to GitLab server (%s)" % self._url)

if r.status_code == 200:

return r.json()

elif r.status_code == 401:

raise GitlabAuthenticationError(r.json()['message'])

else:

raise GitlabUpdateError('%d: %s' % (r.status_code, r.text))

def _getListOrObject(self, cls, id, **kwargs):

if id is None:

return cls.list(self, **kwargs)

else:

return cls(self, id, **kwargs)

def Hook(self, id=None, **kwargs):

"""Creates/tests/lists system hook(s) known by the GitLab server.

return self._getListOrObject(Hook, id, **kwargs)

def Project(self, id=None, **kwargs):

"""Creates/gets/lists project(s) known by the GitLab server.

return self._getListOrObject(Project, id, **kwargs)

def UserProject(self, id=None, **kwargs):

"""Creates a project for a user.

return self._getListOrObject(UserProject, id, **kwargs)

def _list_projects(self, url, **kwargs):

r = self.rawGet(url, **kwargs)

if r.status_code != 200:

raise GitlabListError

l = []

for o in r.json():

l.append(Project(self, o))

return l

def search_projects(self, query):

"""Searches projects by name.

return self._list_projects("/projects/search/" + query)

def all_projects(self, page=None, per_page=None):

"""Lists all the projects (need admin rights)."""

d = {}

if page is not None:

d['page'] = page

if per_page is not None:

d['per_page'] = per_page

return self._list_projects("/projects/all", **d)

def owned_projects(self, page=None, per_page=None):

"""Lists owned projects."""

d = {}

if page is not None:

d['page'] = page

if per_page is not None:

d['per_page'] = per_page

return self._list_projects("/projects/owned", **d)

def Group(self, id=None, **kwargs):

"""Creates/gets/lists group(s) known by the GitLab server.

return self._getListOrObject(Group, id, **kwargs)

def Issue(self, id=None, **kwargs):

"""Lists issues(s) known by the GitLab server.

return self._getListOrObject(Issue, id, **kwargs)

def User(self, id=None, **kwargs):

"""Creates/gets/lists users(s) known by the GitLab server.

return self._getListOrObject(User, id, **kwargs)

def Team(self, id=None, **kwargs):

"""Creates/gets/lists team(s) known by the GitLab server.

_url = None

_returnClass = None

_constructorTypes = None

canGet = True

canList = True

canCreate = True

canUpdate = True

canDelete = True

requiredListAttrs = []

requiredGetAttrs = []

requiredCreateAttrs = []

optionalCreateAttrs = []

idAttr = 'id'

shortPrintAttr = None

@classmethod

def list(cls, gl, **kwargs):

if not cls.canList:

raise NotImplementedError

if not cls._url:

raise NotImplementedError

return gl.list(cls, **kwargs)

def _getListOrObject(self, cls, id, **kwargs):

if id is None:

if not cls.canList:

raise GitlabGetError

return cls.list(self.gitlab, **kwargs)

elif isinstance(id, dict):

if not cls.canCreate:

raise GitlabCreateError

return cls(self.gitlab, id, **kwargs)

else:

if not cls.canGet:

raise GitlabGetError

return cls(self.gitlab, id, **kwargs)

def _getObject(self, k, v):

if self._constructorTypes and k in self._constructorTypes:

return globals()[self._constructorTypes[k]](self.gitlab, v)

else:

return v

def _setFromDict(self, data):

for k, v in data.items():

if isinstance(v, list):

self.__dict__[k] = []

for i in v:

self.__dict__[k].append(self._getObject(k, i))

elif v:

self.__dict__[k] = self._getObject(k, v)

else: # None object

self.__dict__[k] = None

def _create(self):

if not self.canCreate:

raise NotImplementedError

json = self.gitlab.create(self)

self._setFromDict(json)

def _update(self):

if not self.canUpdate:

raise NotImplementedError

json = self.gitlab.update(self)

self._setFromDict(json)

def save(self):

if hasattr(self, 'id'):

self._update()

else:

self._create()

def delete(self):

if not self.canDelete:

raise NotImplementedError

if not hasattr(self, 'id'):

raise GitlabDeleteError

return self.gitlab.delete(self)

def __init__(self, gl, data=None, **kwargs):

self.gitlab = gl

if data is None or type(data) in [int, str, unicode]:

data = self.gitlab.get(self.__class__, data, **kwargs)

self._setFromDict(data)

if kwargs:

for k, v in kwargs.items():

self.__dict__[k] = v

def __str__(self):

return '%s => %s' % (type(self), str(self.__dict__))

def display(self, pretty):

if pretty:

self.pretty_print()

else:

self.short_print()

def short_print(self, depth=0):

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 _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 isinstance(obj, unicode):

return obj.encode(sys.stdout.encoding, "replace")

else:

return str(obj)

def pretty_print(self, depth=0):

id = self.__dict__[self.idAttr]

print("%s%s: %s" % (" " * depth * 2, self.idAttr, id))

for k in sorted(self.__dict__.keys()):

if k == self.idAttr:

continue

v = self.__dict__[k]

pretty_k = k.replace('_', '-').encode(sys.stdout.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))

else:

if isinstance(v, Gitlab):

continue

v = GitlabObject._obj_to_str(v)

print("%s%s: %s" % (" " * depth * 2, pretty_k, v))

def json(self):

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

canGet = False

canList = False

canUpdate = False

canDelete = False

_url = '/users'

shortPrintAttr = 'username'

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

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

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

'can_create_group']

def Key(self, id=None, **kwargs):

return self._getListOrObject(UserKey, id,

user_id=self.id,

_url = '/user/keys'

canUpdate = False

shortPrintAttr = 'title'

_url = '/user'

canList = False

canCreate = False

canUpdate = False

canDelete = False

shortPrintAttr = 'username'

def Key(self, id=None, **kwargs):

if id is None:

return CurrentUserKey.list(self.gitlab, **kwargs)

else:

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

canGet = False

canUpdate = False

requiredCreateAttrs = ['group_id', 'user_id', 'access_level']

requiredDeleteAttrs = ['group_id', 'user_id']

_url = '/groups'

_constructorTypes = {'projects': 'Project'}

requiredCreateAttrs = ['name', 'path']

shortPrintAttr = 'name'

GUEST_ACCESS = 10

REPORTER_ACCESS = 20

DEVELOPER_ACCESS = 30

MASTER_ACCESS = 40

OWNER_ACCESS = 50

def Member(self, id=None, **kwargs):

return self._getListOrObject(GroupMember, id,

group_id=self.id,

**kwargs)

def transfer_project(self, id):

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

r = self.gitlab.rawPost(url, None)

if r.status_code != 201:

_url = '/hooks'

canUpdate = False

requiredCreateAttrs = ['url']

_url = '/issues'

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

'milestone': 'ProjectMilestone'}

canGet = False

canDelete = False

canUpdate = False

canCreate = False

_url = '/projects/%(project_id)s/repository/branches'

idAttr = 'name'

canDelete = False

canUpdate = False

canCreate = False

requiredGetAttrs = ['project_id']

requiredListAttrs = ['project_id']

_constructorTypes = {'commit': 'ProjectCommit'}

def protect(self, protect=True):

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.rawPut(url)

if r.status_code == 200:

if protect:

self.protected = protect

else:

del self.protected

else:

raise GitlabProtectError

def unprotect(self):

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

canDelete = False

canUpdate = False

canCreate = False

requiredListAttrs = ['project_id']

shortPrintAttr = 'title'

def diff(self):

url = '/projects/%(project_id)s/repository/commits/%(commit_id)s/diff' % \

{'project_id': self.project_id, 'commit_id': self.id}

r = self.gitlab.rawGet(url)

if r.status_code == 200:

return r.json()

raise GitlabGetError

def blob(self, filepath):

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.rawGet(url)

if r.status_code == 200:

return r.content

_url = '/projects/%(project_id)s/keys'

canUpdate = False

requiredListAttrs = ['project_id']

requiredGetAttrs = ['project_id']

_url = '/projects/%(project_id)s/events'

canGet = False

canDelete = False

canUpdate = False

canCreate = False

requiredListAttrs = ['project_id']

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

requiredListAttrs = ['project_id']

requiredGetAttrs = ['project_id']

requiredCreateAttrs = ['project_id', 'url']

_url = '/projects/%(project_id)s/issues/%(issue_id)s/notes'

_constructorTypes = {'author': 'User'}

canUpdate = False

canDelete = False

requiredListAttrs = ['project_id', 'issue_id']

requiredGetAttrs = ['project_id', 'issue_id']

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

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

'milestone': 'ProjectMilestone'}

canDelete = False

requiredListAttrs = ['project_id']

requiredGetAttrs = ['project_id']

requiredCreateAttrs = ['project_id', 'title']

optionalCreateAttrs = ['description', 'assignee_id', 'milestone_id',

'labels']

shortPrintAttr = 'title'

def Note(self, id=None, **kwargs):

return self._getListOrObject(ProjectIssueNote, id,

project_id=self.project_id,

issue_id=self.id,

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

requiredListAttrs = ['project_id']

requiredGetAttrs = ['project_id']

requiredCreateAttrs = ['project_id', 'user_id', 'access_level']

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

_constructorTypes = {'author': 'User'}

canUpdate = False

canDelete = False

requiredListAttrs = ['project_id']

requiredGetAttrs = ['project_id']

_url = '/projects/%(project_id)s/repository/tags'

idAttr = 'name'

canGet = False

canDelete = False

canUpdate = False

canCreate = False

requiredListAttrs = ['project_id']

_url = '/projects/%(project_id)s/merge_requests/%(merge_request_id)s/notes'

_constructorTypes = {'author': 'User'}

canGet = False

canCreate = False

canUpdate = False

canDelete = False

_url = '/projects/%(project_id)s/merge_requests'

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

canDelete = False

requiredListAttrs = ['project_id']

requiredGetAttrs = ['project_id']

requiredCreateAttrs = ['project_id', 'source_branch',

'target_branch', 'title']

optionalCreateAttrs = ['assignee_id']

def Note(self, id=None, **kwargs):

return self._getListOrObject(ProjectMergeRequestNote, id,

project_id=self.project_id,

merge_request_id=self.id,

_url = '/projects/%(project_id)s/milestones'

canDelete = False

requiredListAttrs = ['project_id']

requiredGetAttrs = ['project_id']

requiredCreateAttrs = ['project_id', 'title']

optionalCreateAttrs = ['description', 'due_date']

_url = '/projects/%(project_id)s/snippets/%(snippet_id)s/notes'

_constructorTypes = {'author': 'User'}

canUpdate = False

canDelete = False

requiredListAttrs = ['project_id', 'snippet_id']

requiredGetAttrs = ['project_id', 'snippet_id']

_url = '/projects/%(project_id)s/snippets'

_constructorTypes = {'author': 'User'}

requiredListAttrs = ['project_id']

requiredGetAttrs = ['project_id']

requiredCreateAttrs = ['project_id', 'title', 'file_name', 'code']

optionalCreateAttrs = ['lifetime']

shortPrintAttr = 'title'

def Content(self):

url = "/projects/%(project_id)s/snippets/%(snippet_id)s/raw" % \

{'project_id': self.project_id, 'snippet_id': self.id}

r = self.gitlab.rawGet(url)

if r.status_code == 200:

return r.content

else:

raise GitlabGetError

def Note(self, id=None, **kwargs):

return self._getListOrObject(ProjectSnippetNote, id,

project_id=self.project_id,

snippet_id=self.id,

_url = '/projects/user/%(user_id)s'

_constructorTypes = {'owner': 'User', 'namespace': 'Group'}

canUpdate = False

canDelete = False

canList = False

canGet = False

requiredCreateAttrs = ['name', 'user_id']

optionalCreateAttrs = ['default_branch', 'issues_enabled', 'wall_enabled',

'merge_requests_enabled', 'wiki_enabled',

'snippets_enabled', 'public', 'visibility_level',

_url = '/projects'

_constructorTypes = {'owner': 'User', 'namespace': 'Group'}

canUpdate = False

canDelete = False

requiredCreateAttrs = ['name']

optionalCreateAttrs = ['default_branch', 'issues_enabled', 'wall_enabled',

'merge_requests_enabled', 'wiki_enabled',

'snippets_enabled', 'public', 'visibility_level',

'namespace_id', 'description']

shortPrintAttr = 'path'

def Branch(self, id=None, **kwargs):

return self._getListOrObject(ProjectBranch, id,

project_id=self.id,

**kwargs)

def Commit(self, id=None, **kwargs):

return self._getListOrObject(ProjectCommit, id,

project_id=self.id,

**kwargs)

def Event(self, id=None, **kwargs):

return self._getListOrObject(ProjectEvent, id,

project_id=self.id,

**kwargs)

def File(self, id=None, **kwargs):

return self._getListOrObject(ProjectFile, id,

project_id=self.id,

**kwargs)

def Hook(self, id=None, **kwargs):

return self._getListOrObject(ProjectHook, id,

project_id=self.id,

**kwargs)

def Key(self, id=None, **kwargs):

return self._getListOrObject(ProjectKey, id,

project_id=self.id,

**kwargs)

def Issue(self, id=None, **kwargs):

return self._getListOrObject(ProjectIssue, id,

project_id=self.id,

**kwargs)

def Member(self, id=None, **kwargs):

return self._getListOrObject(ProjectMember, id,

project_id=self.id,

**kwargs)

def MergeRequest(self, id=None, **kwargs):

return self._getListOrObject(ProjectMergeRequest, id,

project_id=self.id,

**kwargs)

def Milestone(self, id=None, **kwargs):

return self._getListOrObject(ProjectMilestone, id,

project_id=self.id,

**kwargs)

def Note(self, id=None, **kwargs):

return self._getListOrObject(ProjectNote, id,

project_id=self.id,

**kwargs)

def Snippet(self, id=None, **kwargs):

return self._getListOrObject(ProjectSnippet, id,

project_id=self.id,

**kwargs)