# -*- 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 time
import warnings
import requests
import six
import gitlab.config
from gitlab.const import * # noqa
from gitlab.exceptions import * # noqa
from gitlab import utils # noqa
__title__ = 'python-gitlab'
__version__ = '1.7.0'
__author__ = 'Gauvain Pocentek'
__email__ = 'gauvain@pocentek.net'
__license__ = 'LGPL3'
__copyright__ = 'Copyright 2013-2018 Gauvain Pocentek'
warnings.filterwarnings('default', category=DeprecationWarning,
module='^gitlab')
REDIRECT_MSG = ('python-gitlab detected an http to https redirection. You '
'must update your GitLab URL to use https:// to avoid issues.')
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 (support for 4 only)
"""
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, per_page=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()
self.per_page = per_page
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.geonodes = objects.GeoNodeManager(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.ldapgroups = objects.LDAPGroupManager(self)
self.licenses = objects.LicenseManager(self)
self.namespaces = objects.NamespaceManager(self)
self.mergerequests = objects.MergeRequestManager(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)
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)
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 (4 only)."""
return self._api_version
@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,
per_page=config.per_page)
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}
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):
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.
('unknown', 'unknwown') if the server doesn't
perform as expected.
"""
if self._server_version is None:
try:
data = self.http_get('/version')
self._server_version = data['version']
self._server_revision = data['revision']
except Exception:
self._server_version = self._server_revision = 'unknown'
return self._server_version, self._server_revision
@on_http_error(GitlabVerifyError)
def lint(self, content, **kwargs):
"""Validate a gitlab CI configuration.
Args:
content (txt): The .gitlab-ci.yml content
**kwargs: Extra options to send to the server (e.g. sudo)
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabVerifyError: If the validation could not be done
Returns:
tuple: (True, []) if the file is valid, (False, errors(list))
otherwise
"""
post_data = {'content': content}
data = self.http_post('/ci/lint', post_data=post_data, **kwargs)
return (data['status'] == 'valid', data['errors'])
@on_http_error(GitlabMarkdownError)
def markdown(self, text, gfm=False, project=None, **kwargs):
"""Render an arbitrary Markdown document.
Args:
text (str): The markdown text to render
gfm (bool): Render text using GitLab Flavored Markdown. Default is
False
project (str): Full path of a project used a context when `gfm` is
True
**kwargs: Extra options to send to the server (e.g. sudo)
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabMarkdownError: If the server cannot perform the request
Returns:
str: The HTML rendering of the markdown text.
"""
post_data = {'text': text, 'gfm': gfm}
if project is not None:
post_data['project'] = project
data = self.http_post('/markdown', post_data=post_data, **kwargs)
return data['html']
@on_http_error(GitlabLicenseError)
def get_license(self, **kwargs):
"""Retrieve information about the current license.
Args:
**kwargs: Extra options to send to the server (e.g. sudo)
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabGetError: If the server cannot perform the request
Returns:
dict: The current license information
"""
return self.http_get('/license', **kwargs)
@on_http_error(GitlabLicenseError)
def set_license(self, license, **kwargs):
"""Add a new license.
Args:
license (str): The license string
**kwargs: Extra options to send to the server (e.g. sudo)
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabPostError: If the server cannot perform the request
Returns:
dict: The new license information
"""
data = {'license': license}
return self.http_post('/license', post_data=data, **kwargs)
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)
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 _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.
Returns:
str: The full URL
"""
if path.startswith('http://') or path.startswith('https://'):
return path
else:
return '%s%s' % (self._url, path)
def _check_redirects(self, result):
# Check the requests history to detect http to https redirections.
# If the initial verb is POST, the next request will use a GET request,
# leading to an unwanted behaviour.
# If the initial verb is PUT, the data will not be send with the next
# request.
# If we detect a redirection to https with a POST or a PUT request, we
# raise an exception with a useful error message.
if result.history and self._base_url.startswith('http:'):
for item in result.history:
if item.status_code not in (301, 302):
continue
# GET methods can be redirected without issue
if item.request.method == 'GET':
continue
# Did we end-up with an https:// URL?
location = item.headers.get('Location', None)
if location and location.startswith('https://'):
raise RedirectError(REDIRECT_MSG)
def http_request(self, verb, path, query_data={}, post_data=None,
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
files (dict): The files to send to the server
**kwargs: Extra options to send to the server (e.g. sudo)
Returns:
A requests result object.
Raises:
GitlabHttpError: When the return code is not 2xx
"""
url = self._build_url(path)
params = {}
utils.copy_dict(params, query_data)
utils.copy_dict(params, kwargs)
opts = self._get_session_opts(content_type='application/json')
verify = opts.pop('verify')
timeout = opts.pop('timeout')
# We need to deal with json vs. data when uploading files
if files:
data = post_data
json = None
del opts["headers"]["Content-type"]
else:
json = post_data
data = None
# 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=json, data=data, params=params,
files=files, **opts)
prepped = self.session.prepare_request(req)
prepped.url = utils.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)
self._check_redirects(result)
if 200