def default(self, obj):

if isinstance(obj, GitlabObject):

return obj.__dict__

elif isinstance(obj, Gitlab):

return {'url': obj._url}

def __init__(self, error_message="", response_code=None,

response_body=None):

Exception.__init__(self, error_message)

# Http status code

self.response_code = response_code

# Full http response

self.response_body = response_body

# Parsed error message from gitlab

self.error_message = error_message

def __str__(self):

if self.response_code is not None:

return "{0}: {1}".format(self.response_code, self.error_message)

else:

""" Tries to parse gitlab error message from response and raises error.

try:

message = response.json()['message']

except (KeyError, ValueError):

message = response.content

if response.status_code == 401:

error = GitlabAuthenticationError

raise error(error_message=message,

response_code=response.status_code,

""" Represents a GitLab server connection

def __init__(self, url, private_token=None,

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

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

#: Timeout to use for requests to gitlab server

self.timeout = timeout

#: Headers that will be used in request to GitLab

self.headers = {}

self.setToken(private_token)

#: the user email

self.email = email

#: the user password (associated with email)

self.password = password

#: (Passed to requests-library)

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

data = json.dumps({'email': self.email, 'password': self.password})

r = self.rawPost('/session', data, content_type='application/json')

if r.status_code == 201:

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

else:

_raiseErrorFromResponse(r, GitlabAuthenticationError)

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 constructUrl(self, id_, obj, parameters):

args = _sanitize_dict(parameters)

url = obj._url % args

if id_ is not None:

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

else:

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

return url

def _createHeaders(self, content_type=None, headers={} ):

request_headers = self.headers.copy()

request_headers.update(headers)

if content_type is not None:

request_headers['Content-type'] = content_type

return request_headers

def setToken(self, token):

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

self.private_token = token if token else None

if token:

self.headers["PRIVATE-TOKEN"] = token

elif "PRIVATE-TOKEN" in self.headers:

del self.headers["PRIVATE-TOKEN"]

def setCredentials(self, email, password):

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

self.email = email

self.password = password

def rawGet(self, path, content_type=None, **kwargs):

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

headers = self._createHeaders(content_type)

try:

return requests.get(url,

params=kwargs,

headers=headers,

verify=self.ssl_verify,

timeout=self.timeout)

except:

raise GitlabConnectionError(

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

def rawPost(self, path, data=None, content_type=None, **kwargs):

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

headers = self._createHeaders(content_type)

try:

return requests.post(url, params=kwargs, data=data,

headers=headers,

verify=self.ssl_verify,

timeout=self.timeout)

except:

raise GitlabConnectionError(

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

def rawPut(self, path, data=None, content_type=None, **kwargs):

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

headers = self._createHeaders(content_type)

try:

return requests.put(url, data=data, params=kwargs,

headers=headers,

verify=self.ssl_verify,

timeout=self.timeout)

except:

raise GitlabConnectionError(

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

def rawDelete(self, path, content_type=None, **kwargs):

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

headers = self._createHeaders(content_type)

try:

return requests.delete(url,

params=kwargs,

headers=headers,

verify=self.ssl_verify,

timeout=self.timeout)

except:

raise GitlabConnectionError(

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

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

missing = []

for k in chain(obj_class.requiredUrlAttrs,

obj_class.requiredListAttrs):

if k not in kwargs:

missing.append(k)

if missing:

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

", ".join(missing))

url = self.constructUrl(id_=None, obj=obj_class, parameters=kwargs)

headers = self._createHeaders()

# Remove attributes that are used in url so that there is only

# url-parameters left

params = kwargs.copy()

for attribute in obj_class.requiredUrlAttrs:

del params[attribute]

try:

r = requests.get(url, params=params, headers=headers,

verify=self.ssl_verify,

timeout=self.timeout)

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

cls_kwargs = kwargs.copy()

# Add _created manually, because we are not creating objects

# through normal path

cls_kwargs['_created'] = True

# Remove parameters from kwargs before passing it to constructor

for key in ['page', 'per_page', 'sudo']:

if key in cls_kwargs:

del cls_kwargs[key]

return [cls(self, item, **cls_kwargs) for item in r.json() if item is not None]

else:

_raiseErrorFromResponse(r, GitlabListError)

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

missing = []

for k in chain(obj_class.requiredUrlAttrs,

obj_class.requiredGetAttrs):

if k not in kwargs:

missing.append(k)

if missing:

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

", ".join(missing))

url = self.constructUrl(id_=id, obj=obj_class, parameters=kwargs)

headers = self._createHeaders()

# Remove attributes that are used in url so that there is only

# url-parameters left

params = kwargs.copy()

for attribute in obj_class.requiredUrlAttrs:

del params[attribute]

try:

r = requests.get(url, params=params, headers=headers,

verify=self.ssl_verify, timeout=self.timeout)

except:

raise GitlabConnectionError(

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

if r.status_code == 200:

return r.json()

else:

_raiseErrorFromResponse(r, GitlabGetError)

def delete(self, obj, **kwargs):

params = obj.__dict__.copy()

params.update(kwargs)

missing = []

for k in chain(obj.requiredUrlAttrs, obj.requiredDeleteAttrs):

if k not in params:

missing.append(k)

if missing:

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

", ".join(missing))

url = self.constructUrl(id_=obj.id, obj=obj, parameters=params)

headers = self._createHeaders()

# Remove attributes that are used in url so that there is only

# url-parameters left

for attribute in obj.requiredUrlAttrs:

del params[attribute]

try:

r = requests.delete(url,

params=params,

headers=headers,

verify=self.ssl_verify,

timeout=self.timeout)

except:

raise GitlabConnectionError(

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

if r.status_code == 200:

return True

else:

_raiseErrorFromResponse(r, GitlabDeleteError)

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

params = obj.__dict__.copy()

params.update(kwargs)

missing = []

for k in chain(obj.requiredUrlAttrs, obj.requiredCreateAttrs):

if k not in params:

missing.append(k)

if missing:

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

", ".join(missing))

url = self.constructUrl(id_=None, obj=obj, parameters=params)

headers = self._createHeaders(content_type="application/json")

# build data that can really be sent to server

data = obj._dataForGitlab(extra_parameters=kwargs)

try:

r = requests.post(url, data=data,

headers=headers,

verify=self.ssl_verify,

timeout=self.timeout)

except:

raise GitlabConnectionError(

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

if r.status_code == 201:

return r.json()

else:

_raiseErrorFromResponse(r, GitlabCreateError)

def update(self, obj, **kwargs):

params = obj.__dict__.copy()

params.update(kwargs)

missing = []

for k in chain(obj.requiredUrlAttrs, obj.requiredCreateAttrs):

if k not in params:

missing.append(k)

if missing:

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

", ".join(missing))

url = self.constructUrl(id_=obj.id, obj=obj, parameters=params)

headers = self._createHeaders(content_type="application/json")

# build data that can really be sent to server

data = obj._dataForGitlab(extra_parameters=kwargs)

try:

r = requests.put(url, data=data,

headers=headers,

verify=self.ssl_verify,

timeout=self.timeout)

except:

raise GitlabConnectionError(

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

if r.status_code == 200:

return r.json()

else:

_raiseErrorFromResponse(r, GitlabUpdateError)

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

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

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

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

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

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

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

"""Creates a project for a user.

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

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

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

if r.status_code != 200:

_raiseErrorFromResponse(r, GitlabListError)

l = []

for o in r.json():

l.append(Project(self, o))

return l

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

"""Searches projects by name.

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

def all_projects(self, **kwargs):

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

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

def owned_projects(self, **kwargs):

"""Lists owned projects."""

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

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

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

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

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

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

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

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

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

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

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

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

""" Baseclass for all classes that interface with GitLab

#: Url to use in GitLab for this object

_url = None

_returnClass = None

_constructorTypes = None

#: Tells if _getListOrObject should return list or object when id is None

getListWhenNoId = True

#: 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 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 = []

idAttr = 'id'

shortPrintAttr = None

def _dataForGitlab(self, extra_parameters={}):

data = {}

for attribute in chain(self.requiredCreateAttrs,

self.optionalCreateAttrs):

if hasattr(self, attribute):

data[attribute] = getattr(self,attribute)

data.update(extra_parameters)

return json.dumps(data)

@classmethod

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

if not cls.canList:

raise NotImplementedError

if not cls._url:

raise NotImplementedError

return gl.list(cls, **kwargs)

@classmethod

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

if id is None and cls.getListWhenNoId:

return cls.list(gl, **kwargs)

else:

return cls(gl, 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 is None:

self.__dict__[k] = None

else:

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

def _create(self, **kwargs):

if not self.canCreate:

raise NotImplementedError

json = self.gitlab.create(self, **kwargs)

self._setFromDict(json)

self._created = True

def _update(self, **kwargs):

if not self.canUpdate:

raise NotImplementedError

json = self.gitlab.update(self, **kwargs)

self._setFromDict(json)

def save(self, **kwargs):

if self._created:

self._update(**kwargs)

else:

self._create(**kwargs)

def delete(self, **kwargs):

if not self.canDelete:

raise NotImplementedError

if not self._created:

raise GitlabDeleteError("Object not yet created")

return self.gitlab.delete(self, **kwargs)

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

self._created = False

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)

# Object is created because we got it from api

self._created = True

self._setFromDict(data)

if kwargs:

for k, v in kwargs.items():

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 __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 six.PY2 and isinstance(obj, six.text_type):

return obj.encode(_get_display_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 or k == 'id':

continue

if k[0] == '_':

continue

v = self.__dict__[k]

pretty_k = k.replace('_', '-')

if six.PY2:

pretty_k = pretty_k.encode(_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))

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

canUpdate = False

requiredUrlAttrs = ['user_id']

_url = '/users'

shortPrintAttr = 'username'

# FIXME: password is required for create but not for update

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

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

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

'bio', 'admin', 'can_create_group', 'website_url']

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

return UserKey._getListOrObject(self.gitlab, 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):

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

canGet = False

canUpdate = False

requiredUrlAttrs = ['group_id']

requiredCreateAttrs = ['access_level', 'user_id']

_url = '/groups'

canUpdate = False

_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 GroupMember._getListOrObject(self.gitlab, id,

group_id=self.id,

**kwargs)

def transfer_project(self, id, **kwargs):

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

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

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'

_constructorTypes = {'author': 'User', "committer": "User"}

idAttr = 'name'

canUpdate = False

requiredUrlAttrs = ['project_id']

requiredCreateAttrs = ['branch_name', 'ref']

_constructorTypes = {'commit': 'ProjectCommit'}

def protect(self, protect=True, **kwargs):

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, data=None, content_type=None, **kwargs)

if r.status_code == 200:

if protect:

self.protected = protect

else:

del self.protected

else:

_raiseErrorFromResponse(r, GitlabProtectError)

def unprotect(self, **kwargs):

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

canDelete = False

canUpdate = False

canCreate = False

requiredUrlAttrs = ['project_id']

shortPrintAttr = 'title'

def diff(self, **kwargs):

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, **kwargs)

if r.status_code == 200:

return r.json()

else:

_raiseErrorFromResponse(r, GitlabGetError)

def blob(self, filepath, **kwargs):

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, **kwargs)

if r.status_code == 200:

return r.content

else:

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

canUpdate = False

requiredUrlAttrs = ['project_id']

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

canGet = False

canDelete = False

canUpdate = False

canCreate = False

requiredUrlAttrs = ['project_id']

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

requiredUrlAttrs = ['project_id']

requiredCreateAttrs = ['url']

optionalCreateAttrs = ['push_events', 'issues_events',

'merge_requests_events', 'tag_push_events']

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

_constructorTypes = {'author': 'User'}

canUpdate = False

canDelete = False

requiredUrlAttrs = ['project_id', 'issue_id']

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

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

'milestone': 'ProjectMilestone'}

canDelete = False

requiredUrlAttrs = ['project_id']

requiredCreateAttrs = ['title']

# FIXME: state_event is only valid with update

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

'labels', 'state_event']

shortPrintAttr = 'title'

def _dataForGitlab(self, extra_parameters={}):

# Gitlab-api returns labels in a json list and takes them in a

# comma separated list.

if hasattr(self, "labels"):

if self.labels is not None and not isinstance(self.labels, six.string_types):

labels = ", ".join(self.labels)

extra_parameters['labels'] = labels

return super(ProjectIssue, self)._dataForGitlab(extra_parameters)

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

return ProjectIssueNote._getListOrObject(self.gitlab, id,

project_id=self.project_id,

issue_id=self.id,

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

requiredUrlAttrs = ['project_id']

requiredCreateAttrs = ['access_level', 'user_id']

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

_constructorTypes = {'author': 'User'}

canUpdate = False

canDelete = False

requiredUrlAttrs = ['project_id']