# -*- coding: utf-8 -*-
#
# Copyright (C) 2013-2017 Gauvain Pocentek
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with this program. If not, see .
"""Wrapper for the GitLab API."""
from __future__ import print_function
from __future__ import absolute_import
import importlib
import inspect
import itertools
import json
import re
import warnings
import requests
import six
import gitlab.config
from gitlab.const import * # noqa
from gitlab.exceptions import * # noqa
from gitlab.v3.objects import * # noqa
__title__ = 'python-gitlab'
__version__ = '1.1.0'
__author__ = 'Gauvain Pocentek'
__email__ = 'gauvain@pocentek.net'
__license__ = 'LGPL3'
__copyright__ = 'Copyright 2013-2017 Gauvain Pocentek'
warnings.filterwarnings('default', category=DeprecationWarning,
module='^gitlab')
def _sanitize(value):
if isinstance(value, dict):
return dict((k, _sanitize(v))
for k, v in six.iteritems(value))
if isinstance(value, six.string_types):
return value.replace('/', '%2F')
return value
class Gitlab(object):
"""Represents a GitLab server connection.
Args:
url (str): The URL of the GitLab server.
private_token (str): The user private token
oauth_token (str): An oauth token
email (str): The user email or login.
password (str): The user password (associated with email).
ssl_verify (bool|str): Whether SSL certificates should be validated. If
the value is a string, it is the path to a CA file used for
certificate validation.
timeout (float): Timeout to use for requests to the GitLab server.
http_username (str): Username for HTTP authentication
http_password (str): Password for HTTP authentication
api_version (str): Gitlab API version to use (3 or 4)
"""
def __init__(self, url, private_token=None, oauth_token=None, email=None,
password=None, ssl_verify=True, http_username=None,
http_password=None, timeout=None, api_version='3',
session=None):
self._api_version = str(api_version)
self._server_version = self._server_revision = None
self._url = '%s/api/v%s' % (url, api_version)
#: Timeout to use for requests to gitlab server
self.timeout = timeout
#: Headers that will be used in request to GitLab
self.headers = {}
#: The user email
self.email = email
#: The user password (associated with email)
self.password = password
#: Whether SSL certificates should be validated
self.ssl_verify = ssl_verify
self.private_token = private_token
self.http_username = http_username
self.http_password = http_password
self.oauth_token = oauth_token
self._set_auth_info()
#: Create a session object for requests
self.session = session or requests.Session()
objects = importlib.import_module('gitlab.v%s.objects' %
self._api_version)
self._objects = objects
self.broadcastmessages = objects.BroadcastMessageManager(self)
self.deploykeys = objects.DeployKeyManager(self)
self.gitlabciymls = objects.GitlabciymlManager(self)
self.gitignores = objects.GitignoreManager(self)
self.groups = objects.GroupManager(self)
self.hooks = objects.HookManager(self)
self.issues = objects.IssueManager(self)
self.licenses = objects.LicenseManager(self)
self.namespaces = objects.NamespaceManager(self)
self.notificationsettings = objects.NotificationSettingsManager(self)
self.projects = objects.ProjectManager(self)
self.runners = objects.RunnerManager(self)
self.settings = objects.ApplicationSettingsManager(self)
self.sidekiq = objects.SidekiqManager(self)
self.snippets = objects.SnippetManager(self)
self.users = objects.UserManager(self)
self.todos = objects.TodoManager(self)
if self._api_version == '3':
self.teams = objects.TeamManager(self)
else:
self.dockerfiles = objects.DockerfileManager(self)
self.user_activities = objects.UserActivitiesManager(self)
if self._api_version == '3':
# build the "submanagers"
for parent_cls in six.itervalues(vars(objects)):
if (not inspect.isclass(parent_cls)
or not issubclass(parent_cls, objects.GitlabObject)
or parent_cls == objects.CurrentUser):
continue
if not parent_cls.managers:
continue
for var, cls_name, attrs in parent_cls.managers:
prefix = self._cls_to_manager_prefix(parent_cls)
var_name = '%s_%s' % (prefix, var)
manager = getattr(objects, cls_name)(self)
setattr(self, var_name, manager)
def __getstate__(self):
state = self.__dict__.copy()
state.pop('_objects')
return state
def __setstate__(self, state):
self.__dict__.update(state)
objects = importlib.import_module('gitlab.v%s.objects' %
self._api_version)
self._objects = objects
@property
def api_version(self):
return self._api_version
def _cls_to_manager_prefix(self, cls):
# Manage bad naming decisions
camel_case = (cls.__name__
.replace('NotificationSettings', 'Notificationsettings')
.replace('MergeRequest', 'Mergerequest')
.replace('AccessRequest', 'Accessrequest'))
return re.sub(r'(.)([A-Z])', r'\1_\2', camel_case).lower()
@staticmethod
def from_config(gitlab_id=None, config_files=None):
"""Create a Gitlab connection from configuration files.
Args:
gitlab_id (str): ID of the configuration section.
config_files list[str]: List of paths to configuration files.
Returns:
(gitlab.Gitlab): A Gitlab connection.
Raises:
gitlab.config.GitlabDataError: If the configuration is not correct.
"""
config = gitlab.config.GitlabConfigParser(gitlab_id=gitlab_id,
config_files=config_files)
return Gitlab(config.url, private_token=config.private_token,
oauth_token=config.oauth_token,
ssl_verify=config.ssl_verify, timeout=config.timeout,
http_username=config.http_username,
http_password=config.http_password,
api_version=config.api_version)
def auth(self):
"""Performs an authentication.
Uses either the private token, or the email/password pair.
The `user` attribute will hold a `gitlab.objects.CurrentUser` object on
success.
"""
if self.private_token or self.oauth_token:
self._token_auth()
else:
self._credentials_auth()
def _credentials_auth(self):
data = {'email': self.email, 'password': self.password}
if self.api_version == '3':
r = self._raw_post('/session', json.dumps(data),
content_type='application/json')
raise_error_from_response(r, GitlabAuthenticationError, 201)
self.user = self._objects.CurrentUser(self, r.json())
else:
r = self.http_post('/session', data)
manager = self._objects.CurrentUserManager(self)
self.user = self._objects.CurrentUser(manager, r)
self.private_token = self.user.private_token
self._set_auth_info()
def _token_auth(self):
if self.api_version == '3':
self.user = self._objects.CurrentUser(self)
else:
self.user = self._objects.CurrentUserManager(self).get()
def version(self):
"""Returns the version and revision of the gitlab server.
Note that self.version and self.revision will be set on the gitlab
object.
Returns:
tuple (str, str): The server version and server revision, or
('unknown', 'unknwown') if the server doesn't
support this API call (gitlab < 8.13.0)
"""
if self._server_version is None:
r = self._raw_get('/version')
try:
raise_error_from_response(r, GitlabGetError, 200)
data = r.json()
self._server_version = data['version']
self._server_revision = data['revision']
except GitlabGetError:
self._server_version = self._server_revision = 'unknown'
return self._server_version, self._server_revision
def _construct_url(self, id_, obj, parameters, action=None):
if 'next_url' in parameters:
return parameters['next_url']
args = _sanitize(parameters)
url_attr = '_url'
if action is not None:
attr = '_%s_url' % action
if hasattr(obj, attr):
url_attr = attr
obj_url = getattr(obj, url_attr)
# TODO(gpocentek): the following will need an update when we have
# object with both urlPlural and _ACTION_url attributes
if id_ is None and obj._urlPlural is not None:
url = obj._urlPlural % args
else:
url = obj_url % args
if id_ is not None:
return '%s/%s' % (url, str(id_))
else:
return url
def _set_auth_info(self):
if self.private_token and self.oauth_token:
raise ValueError("Only one of private_token or oauth_token should "
"be defined")
if ((self.http_username and not self.http_password)
or (not self.http_username and self.http_password)):
raise ValueError("Both http_username and http_password should "
"be defined")
if self.oauth_token and self.http_username:
raise ValueError("Only one of oauth authentication or http "
"authentication should be defined")
self._http_auth = None
if self.private_token:
self.headers['PRIVATE-TOKEN'] = self.private_token
self.headers.pop('Authorization', None)
if self.oauth_token:
self.headers['Authorization'] = "Bearer %s" % self.oauth_token
self.headers.pop('PRIVATE-TOKEN', None)
if self.http_username:
self._http_auth = requests.auth.HTTPBasicAuth(self.http_username,
self.http_password)
def enable_debug(self):
import logging
try:
from http.client import HTTPConnection
except ImportError:
from httplib import HTTPConnection # noqa
HTTPConnection.debuglevel = 1
logging.basicConfig()
logging.getLogger().setLevel(logging.DEBUG)
requests_log = logging.getLogger("requests.packages.urllib3")
requests_log.setLevel(logging.DEBUG)
requests_log.propagate = True
def _create_headers(self, content_type=None):
request_headers = self.headers.copy()
if content_type is not None:
request_headers['Content-type'] = content_type
return request_headers
def _get_session_opts(self, content_type):
return {
'headers': self._create_headers(content_type),
'auth': self._http_auth,
'timeout': self.timeout,
'verify': self.ssl_verify
}
def _raw_get(self, path_, content_type=None, streamed=False, **kwargs):
if path_.startswith('http://') or path_.startswith('https://'):
url = path_
else:
url = '%s%s' % (self._url, path_)
opts = self._get_session_opts(content_type)
try:
return self.session.get(url, params=kwargs, stream=streamed,
**opts)
except Exception as e:
raise GitlabConnectionError(
"Can't connect to GitLab server (%s)" % e)
def _raw_list(self, path_, cls, **kwargs):
params = kwargs.copy()
catch_recursion_limit = kwargs.get('safe_all', False)
get_all_results = (kwargs.get('all', False) is True
or catch_recursion_limit)
# Remove these keys to avoid breaking the listing (urls will get too
# long otherwise)
for key in ['all', 'next_url', 'safe_all']:
if key in params:
del params[key]
r = self._raw_get(path_, **params)
raise_error_from_response(r, GitlabListError)
# These attributes are not needed in the object
for key in ['page', 'per_page', 'sudo']:
if key in params:
del params[key]
# Add _from_api manually, because we are not creating objects
# through normal path_
params['_from_api'] = True
results = [cls(self, item, **params) for item in r.json()
if item is not None]
try:
if ('next' in r.links and 'url' in r.links['next']
and get_all_results):
args = kwargs.copy()
args['next_url'] = r.links['next']['url']
results.extend(self.list(cls, **args))
except Exception as e:
# Catch the recursion limit exception if the 'safe_all'
# kwarg was provided
if not (catch_recursion_limit and
"maximum recursion depth exceeded" in str(e)):
raise e
return results
def _raw_post(self, path_, data=None, content_type=None,
files=None, **kwargs):
url = '%s%s' % (self._url, path_)
opts = self._get_session_opts(content_type)
try:
return self.session.post(url, params=kwargs, data=data,
files=files, **opts)
except Exception as e:
raise GitlabConnectionError(
"Can't connect to GitLab server (%s)" % e)
def _raw_put(self, path_, data=None, content_type=None, **kwargs):
url = '%s%s' % (self._url, path_)
opts = self._get_session_opts(content_type)
try:
return self.session.put(url, data=data, params=kwargs, **opts)
except Exception as e:
raise GitlabConnectionError(
"Can't connect to GitLab server (%s)" % e)
def _raw_delete(self, path_, content_type=None, **kwargs):
url = '%s%s' % (self._url, path_)
opts = self._get_session_opts(content_type)
try:
return self.session.delete(url, params=kwargs, **opts)
except Exception as e:
raise GitlabConnectionError(
"Can't connect to GitLab server (%s)" % e)
def list(self, obj_class, **kwargs):
"""Request the listing of GitLab resources.
Args:
obj_class (object): The class of resource to request.
**kwargs: Additional arguments to send to GitLab.
Returns:
list(obj_class): A list of objects of class `obj_class`.
Raises:
GitlabConnectionError: If the server cannot be reached.
GitlabListError: If the server fails to perform the request.
"""
missing = []
for k in itertools.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._construct_url(id_=None, obj=obj_class, parameters=kwargs)
return self._raw_list(url, obj_class, **kwargs)
def get(self, obj_class, id=None, **kwargs):
"""Request a GitLab resources.
Args:
obj_class (object): The class of resource to request.
id: The object ID.
**kwargs: Additional arguments to send to GitLab.
Returns:
obj_class: An object of class `obj_class`.
Raises:
GitlabConnectionError: If the server cannot be reached.
GitlabGetError: If the server fails to perform the request.
"""
missing = []
for k in itertools.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._construct_url(id_=_sanitize(id), obj=obj_class,
parameters=kwargs)
r = self._raw_get(url, **kwargs)
raise_error_from_response(r, GitlabGetError)
return r.json()
def delete(self, obj, id=None, **kwargs):
"""Delete an object on the GitLab server.
Args:
obj (object or id): The object, or the class of the object to
delete. If it is the class, the id of the object must be
specified as the `id` arguments.
id: ID of the object to remove. Required if `obj` is a class.
**kwargs: Additional arguments to send to GitLab.
Returns:
bool: True if the operation succeeds.
Raises:
GitlabConnectionError: If the server cannot be reached.
GitlabDeleteError: If the server fails to perform the request.
"""
if inspect.isclass(obj):
if not issubclass(obj, GitlabObject):
raise GitlabError("Invalid class: %s" % obj)
params = {obj.idAttr: id if id else getattr(obj, obj.idAttr)}
params.update(kwargs)
missing = []
for k in itertools.chain(obj.requiredUrlAttrs,
obj.requiredDeleteAttrs):
if k not in params:
try:
params[k] = getattr(obj, k)
except KeyError:
missing.append(k)
if missing:
raise GitlabDeleteError('Missing attribute(s): %s' %
", ".join(missing))
obj_id = params[obj.idAttr] if obj._id_in_delete_url else None
url = self._construct_url(id_=obj_id, obj=obj, parameters=params)
if obj._id_in_delete_url:
# The ID is already built, no need to add it as extra key in query
# string
params.pop(obj.idAttr)
r = self._raw_delete(url, **params)
raise_error_from_response(r, GitlabDeleteError,
expected_code=[200, 202, 204])
return True
def create(self, obj, **kwargs):
"""Create an object on the GitLab server.
The object class and attributes define the request to be made on the
GitLab server.
Args:
obj (object): The object to create.
**kwargs: Additional arguments to send to GitLab.
Returns:
str: A json representation of the object as returned by the GitLab
server
Raises:
GitlabConnectionError: If the server cannot be reached.
GitlabCreateError: If the server fails to perform the request.
"""
params = obj.__dict__.copy()
params.update(kwargs)
missing = []
for k in itertools.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._construct_url(id_=None, obj=obj, parameters=params,
action='create')
# build data that can really be sent to server
data = obj._data_for_gitlab(extra_parameters=kwargs)
r = self._raw_post(url, data=data, content_type='application/json')
raise_error_from_response(r, GitlabCreateError, 201)
return r.json()
def update(self, obj, **kwargs):
"""Update an object on the GitLab server.
The object class and attributes define the request to be made on the
GitLab server.
Args:
obj (object): The object to create.
**kwargs: Additional arguments to send to GitLab.
Returns:
str: A json representation of the object as returned by the GitLab
server
Raises:
GitlabConnectionError: If the server cannot be reached.
GitlabUpdateError: If the server fails to perform the request.
"""
params = obj.__dict__.copy()
params.update(kwargs)
missing = []
if obj.requiredUpdateAttrs or obj.optionalUpdateAttrs:
required_attrs = obj.requiredUpdateAttrs
else:
required_attrs = obj.requiredCreateAttrs
for k in itertools.chain(obj.requiredUrlAttrs, required_attrs):
if k not in params:
missing.append(k)
if missing:
raise GitlabUpdateError('Missing attribute(s): %s' %
", ".join(missing))
obj_id = params[obj.idAttr] if obj._id_in_update_url else None
url = self._construct_url(id_=obj_id, obj=obj, parameters=params)
# build data that can really be sent to server
data = obj._data_for_gitlab(extra_parameters=kwargs, update=True)
r = self._raw_put(url, data=data, content_type='application/json')
raise_error_from_response(r, GitlabUpdateError)
return r.json()
def _build_url(self, path):
"""Returns the full url from path.
If path is already a url, return it unchanged. If it's a path, append
it to the stored url.
This is a low-level method, different from _construct_url _build_url
have no knowledge of GitlabObject's.
Returns:
str: The full URL
"""
if path.startswith('http://') or path.startswith('https://'):
return path
else:
return '%s%s' % (self._url, path)
def http_request(self, verb, path, query_data={}, post_data={},
streamed=False, files=None, **kwargs):
"""Make an HTTP request to the Gitlab server.
Args:
verb (str): The HTTP method to call ('get', 'post', 'put',
'delete')
path (str): Path or full URL to query ('/projects' or
'http://whatever/v4/api/projecs')
query_data (dict): Data to send as query parameters
post_data (dict): Data to send in the body (will be converted to
json)
streamed (bool): Whether the data should be streamed
**kwargs: Extra data to make the query (e.g. sudo, per_page, page)
Returns:
A requests result object.
Raises:
GitlabHttpError: When the return code is not 2xx
"""
def sanitized_url(url):
parsed = six.moves.urllib.parse.urlparse(url)
new_path = parsed.path.replace('.', '%2E')
return parsed._replace(path=new_path).geturl()
url = self._build_url(path)
params = query_data.copy()
params.update(kwargs)
opts = self._get_session_opts(content_type='application/json')
# don't set the content-type header when uploading files
if files is not None:
del opts["headers"]["Content-type"]
verify = opts.pop('verify')
timeout = opts.pop('timeout')
# Requests assumes that `.` should not be encoded as %2E and will make
# changes to urls using this encoding. Using a prepped request we can
# get the desired behavior.
# The Requests behavior is right but it seems that web servers don't
# always agree with this decision (this is the case with a default
# gitlab installation)
req = requests.Request(verb, url, json=post_data, params=params,
files=files, **opts)
prepped = self.session.prepare_request(req)
prepped.url = sanitized_url(prepped.url)
result = self.session.send(prepped, stream=streamed, verify=verify,
timeout=timeout)
if 200