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

"""Represents a GitLab server connection.

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='4',

session=None):

self._api_version = str(api_version)

self._server_version = self._server_revision = None

self._base_url = url

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.events = objects.EventManager(self)

self.features = objects.FeatureManager(self)

self.pagesdomains = objects.PagesDomainManager(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 __enter__(self):

return self

def __exit__(self, *args):

self.session.close()

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 url(self):

"""The user-provided server URL."""

return self._base_url

@property

def api_url(self):

"""The computed API base URL."""

return self._url

@property

def api_version(self):

"""The API version used (3 or 4)."""

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.

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.

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.

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.

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.

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.

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.

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.

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

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)

def copy_dict(dest, src):

for k, v in src.items():

if isinstance(v, dict):

# Transform dict values in new attributes. For example:

# custom_attributes: {'foo', 'bar'} =>

# custom_attributes['foo']: 'bar'

for dict_k, dict_v in v.items():

dest['%s[%s]' % (k, dict_k)] = dict_v

else:

dest[k] = v

params = {}

copy_dict(params, query_data)

copy_dict(params, 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)

settings = self.session.merge_environment_settings(

prepped.url, {}, streamed, verify, None)

# obey the rate limit by default

obey_rate_limit = kwargs.get("obey_rate_limit", True)

while True:

result = self.session.send(prepped, timeout=timeout, **settings)

if 200 <= result.status_code < 300:

return result

if 429 == result.status_code and obey_rate_limit:

wait_time = int(result.headers["Retry-After"])

time.sleep(wait_time)

continue

try:

error_message = result.json()['message']

except (KeyError, ValueError, TypeError):

error_message = result.content

if result.status_code == 401:

raise GitlabAuthenticationError(

response_code=result.status_code,

error_message=error_message,

response_body=result.content)

raise GitlabHttpError(response_code=result.status_code,

error_message=error_message,

response_body=result.content)

def http_get(self, path, query_data={}, streamed=False, **kwargs):

"""Make a GET request to the Gitlab server.

result = self.http_request('get', path, query_data=query_data,

streamed=streamed, **kwargs)

if (result.headers['Content-Type'] == 'application/json' and

not streamed):

try:

return result.json()

except Exception:

raise GitlabParsingError(

error_message="Failed to parse the server message")

else:

return result

def http_list(self, path, query_data={}, as_list=None, **kwargs):

"""Make a GET request to the Gitlab server for list-oriented queries.

# In case we want to change the default behavior at some point

as_list = True if as_list is None else as_list

get_all = kwargs.get('all', False)

url = self._build_url(path)

if get_all is True:

return list(GitlabList(self, url, query_data, **kwargs))

if 'page' in kwargs or as_list is True:

# pagination requested, we return a list

return list(GitlabList(self, url, query_data, get_next=False,

**kwargs))

# No pagination, generator requested

return GitlabList(self, url, query_data, **kwargs)

def http_post(self, path, query_data={}, post_data={}, files=None,

**kwargs):

"""Make a POST request to the Gitlab server.

result = self.http_request('post', path, query_data=query_data,

post_data=post_data, files=files, **kwargs)

try:

if result.headers.get('Content-Type', None) == 'application/json':

return result.json()

except Exception:

raise GitlabParsingError(

error_message="Failed to parse the server message")

return result

def http_put(self, path, query_data={}, post_data={}, **kwargs):

"""Make a PUT request to the Gitlab server.

result = self.http_request('put', path, query_data=query_data,

post_data=post_data, **kwargs)

try:

return result.json()

except Exception:

raise GitlabParsingError(

error_message="Failed to parse the server message")

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

"""Make a PUT request to the Gitlab server.

"""Generator representing a list of remote objects.

def __init__(self, gl, url, query_data, get_next=True, **kwargs):

self._gl = gl

self._query(url, query_data, **kwargs)

self._get_next = get_next

def _query(self, url, query_data={}, **kwargs):

result = self._gl.http_request('get', url, query_data=query_data,

**kwargs)

try:

self._next_url = result.links['next']['url']

except KeyError:

self._next_url = None

self._current_page = result.headers.get('X-Page')

self._prev_page = result.headers.get('X-Prev-Page')

self._next_page = result.headers.get('X-Next-Page')

self._per_page = result.headers.get('X-Per-Page')

self._total_pages = result.headers.get('X-Total-Pages')

self._total = result.headers.get('X-Total')

try:

self._data = result.json()

except Exception:

raise GitlabParsingError(

error_message="Failed to parse the server message")

self._current = 0

@property

def current_page(self):

"""The current page number."""

return int(self._current_page)

@property

def prev_page(self):

"""The next page number.

return int(self._prev_page) if self._prev_page else None

@property

def next_page(self):

"""The next page number.

return int(self._next_page) if self._next_page else None

@property

def per_page(self):

"""The number of items per page."""

return int(self._per_page)

@property

def total_pages(self):

"""The total number of pages."""

return int(self._total_pages)

@property

def total(self):

"""The total number of items."""

return int(self._total)

def __iter__(self):

return self

def __len__(self):

return int(self._total)

def __next__(self):

return self.next()

def next(self):

try:

item = self._data[self._current]

self._current += 1

return item

except IndexError:

if self._next_url and self._get_next is True:

self._query(self._next_url)

return self.next()