# -*- coding: utf-8 -*- # # Copyright (C) 2013-2015 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 . """Package for interfacing with GitLab-api """ from __future__ import print_function from __future__ import division from __future__ import absolute_import import itertools import json import warnings import requests import six import gitlab.config from gitlab.exceptions import * # noqa from gitlab.objects import * # noqa __title__ = 'python-gitlab' __version__ = '0.10' __author__ = 'Gauvain Pocentek' __email__ = 'gauvain@pocentek.net' __license__ = 'LGPL3' __copyright__ = 'Copyright 2013-2015 Gauvain Pocentek' warnings.simplefilter('always', DeprecationWarning) def _sanitize(value): if isinstance(value, six.string_types): return value.replace('/', '%2F') return value def _sanitize_dict(src): return dict((k, _sanitize(v)) for k, v in src.items()) class Gitlab(object): """Represents a GitLab server connection Args: url (str): the URL of the Gitlab server private_token (str): the user private token email (str): the user email/login password (str): the user password (associated with email) ssl_verify (bool): (Passed to requests-library) timeout (float or tuple(float,float)): (Passed to requests-library). Timeout to use for requests to gitlab server """ 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.set_token(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 self.user_keys = UserKeyManager(self) self.users = UserManager(self) self.group_members = GroupMemberManager(self) self.groups = GroupManager(self) self.hooks = HookManager(self) self.issues = IssueManager(self) self.project_branches = ProjectBranchManager(self) self.project_commits = ProjectCommitManager(self) self.project_keys = ProjectKeyManager(self) self.project_events = ProjectEventManager(self) self.project_forks = ProjectForkManager(self) self.project_hooks = ProjectHookManager(self) self.project_issue_notes = ProjectIssueNoteManager(self) self.project_issues = ProjectIssueManager(self) self.project_members = ProjectMemberManager(self) self.project_notes = ProjectNoteManager(self) self.project_tags = ProjectTagManager(self) self.project_mergerequest_notes = ProjectMergeRequestNoteManager(self) self.project_mergerequests = ProjectMergeRequestManager(self) self.project_milestones = ProjectMilestoneManager(self) self.project_labels = ProjectLabelManager(self) self.project_files = ProjectFileManager(self) self.project_snippet_notes = ProjectSnippetNoteManager(self) self.project_snippets = ProjectSnippetManager(self) self.user_projects = UserProjectManager(self) self.projects = ProjectManager(self) self.team_members = TeamMemberManager(self) self.team_projects = TeamProjectManager(self) self.teams = TeamManager(self) @staticmethod def from_config(gitlab_id=None, config_files=None): config = gitlab.config.GitlabConfigParser(gitlab_id=gitlab_id, config_files=config_files) return Gitlab(config.url, private_token=config.token, ssl_verify=config.ssl_verify, timeout=config.timeout) def auth(self): """Performs an authentication. Uses either the private token, or the email/password pair. The user attribute will hold a CurrentUser object on success. """ 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._raw_post('/session', data, content_type='application/json') raise_error_from_response(r, GitlabAuthenticationError, 201) self.user = CurrentUser(self, r.json()) self.set_token(self.user.private_token) def token_auth(self): self.user = CurrentUser(self) def set_url(self, url): """Updates the gitlab URL.""" self._url = '%s/api/v3' % url def _construct_url(self, id_, obj, parameters): if 'next_url' in parameters: return parameters['next_url'] args = _sanitize_dict(parameters) if id_ is None and obj._urlPlural is not None: url = obj._urlPlural % args else: 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 _create_headers(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 set_token(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 set_credentials(self, email, password): """Sets the email/login and password for authentication.""" self.email = email self.password = password def _raw_get(self, path, content_type=None, **kwargs): url = '%s%s' % (self._url, path) headers = self._create_headers(content_type) try: return requests.get(url, params=kwargs, headers=headers, verify=self.ssl_verify, timeout=self.timeout) except Exception: raise GitlabConnectionError( "Can't connect to GitLab server (%s)" % self._url) def _raw_post(self, path, data=None, content_type=None, **kwargs): url = '%s%s' % (self._url, path) headers = self._create_headers(content_type) try: return requests.post(url, params=kwargs, data=data, headers=headers, verify=self.ssl_verify, timeout=self.timeout) except Exception: raise GitlabConnectionError( "Can't connect to GitLab server (%s)" % self._url) def _raw_put(self, path, data=None, content_type=None, **kwargs): url = '%s%s' % (self._url, path) headers = self._create_headers(content_type) try: return requests.put(url, data=data, params=kwargs, headers=headers, verify=self.ssl_verify, timeout=self.timeout) except Exception: raise GitlabConnectionError( "Can't connect to GitLab server (%s)" % self._url) def _raw_delete(self, path, content_type=None, **kwargs): url = '%s%s' % (self._url, path) headers = self._create_headers(content_type) try: return requests.delete(url, params=kwargs, headers=headers, verify=self.ssl_verify, timeout=self.timeout) except Exception: raise GitlabConnectionError( "Can't connect to GitLab server (%s)" % self._url) def list(self, obj_class, **kwargs): 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) headers = self._create_headers() # 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 Exception: raise GitlabConnectionError( "Can't connect to GitLab server (%s)" % self._url) raise_error_from_response(r, GitlabListError) cls = obj_class if obj_class._returnClass: cls = obj_class._returnClass cls_kwargs = kwargs.copy() # Add _from_api manually, because we are not creating objects # through normal path cls_kwargs['_from_api'] = True get_all_results = params.get('all', False) # Remove parameters from kwargs before passing it to constructor for key in ['all', 'page', 'per_page', 'sudo']: if key in cls_kwargs: del cls_kwargs[key] results = [cls(self, item, **cls_kwargs) for item in r.json() if item is not None] if ('next' in r.links and 'url' in r.links['next'] and get_all_results is True): args = kwargs.copy() args['next_url'] = r.links['next']['url'] results.extend(self.list(obj_class, **args)) return results def get(self, obj_class, id=None, **kwargs): 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)) sanitized_id = _sanitize(id) url = self._construct_url(id_=sanitized_id, obj=obj_class, parameters=kwargs) headers = self._create_headers() # 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 Exception: raise GitlabConnectionError( "Can't connect to GitLab server (%s)" % self._url) raise_error_from_response(r, GitlabGetError) return r.json() def delete(self, obj, **kwargs): params = obj.__dict__.copy() params.update(kwargs) missing = [] for k in itertools.chain(obj.requiredUrlAttrs, obj.requiredDeleteAttrs): if k not in params: missing.append(k) if missing: raise GitlabDeleteError('Missing attribute(s): %s' % ", ".join(missing)) obj_id = getattr(obj, obj.idAttr) url = self._construct_url(id_=obj_id, obj=obj, parameters=params) headers = self._create_headers() # 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 Exception: raise GitlabConnectionError( "Can't connect to GitLab server (%s)" % self._url) raise_error_from_response(r, GitlabDeleteError) return True def create(self, obj, **kwargs): 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) headers = self._create_headers(content_type="application/json") # build data that can really be sent to server data = obj._data_for_gitlab(extra_parameters=kwargs) try: r = requests.post(url, data=data, headers=headers, verify=self.ssl_verify, timeout=self.timeout) except Exception: raise GitlabConnectionError( "Can't connect to GitLab server (%s)" % self._url) raise_error_from_response(r, GitlabCreateError, 201) return r.json() def update(self, obj, **kwargs): 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 GitlabUpdateError('Missing attribute(s): %s' % ", ".join(missing)) url = self._construct_url(id_=obj.id, obj=obj, parameters=params) headers = self._create_headers(content_type="application/json") # build data that can really be sent to server data = obj._data_for_gitlab(extra_parameters=kwargs) try: r = requests.put(url, data=data, headers=headers, verify=self.ssl_verify, timeout=self.timeout) except Exception: raise GitlabConnectionError( "Can't connect to GitLab server (%s)" % self._url) raise_error_from_response(r, GitlabUpdateError) return r.json() def Hook(self, id=None, **kwargs): """Creates/tests/lists system hook(s) known by the GitLab server. If id is None, returns a list of hooks. If id is an integer, tests the matching hook. If id is a dict, creates a new object using attributes provided. The object is NOT saved on the server. Use the save() method on the object to write it on the server. """ warnings.warn("`Hook` is deprecated, use `hooks` instead", DeprecationWarning) return Hook._get_list_or_object(self, id, **kwargs) def Project(self, id=None, **kwargs): """Creates/gets/lists project(s) known by the GitLab server. If id is None, returns a list of projects. If id is an integer, returns the matching project (or raises a GitlabGetError if not found) If id is a dict, creates a new object using attributes provided. The object is NOT saved on the server. Use the save() method on the object to write it on the server. """ warnings.warn("`Project` is deprecated, use `projects` instead", DeprecationWarning) return Project._get_list_or_object(self, id, **kwargs) def UserProject(self, id=None, **kwargs): """Creates a project for a user. id must be a dict. """ warnings.warn("`UserProject` is deprecated, " "use `user_projects` instead", DeprecationWarning) return UserProject._get_list_or_object(self, id, **kwargs) def _list_projects(self, url, **kwargs): r = self._raw_get(url, **kwargs) raise_error_from_response(r, GitlabListError) l = [] for o in r.json(): p = Project(self, o) p._from_api = True l.append(p) return l def search_projects(self, query, **kwargs): """Searches projects by name. Returns a list of matching projects. """ 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 Args: id: If id is None, returns a list of groups. id: If id is an integer, returns the matching group (or raises a GitlabGetError if not found). id: If id is a dict, creates a new object using attributes provided. The object is NOT saved on the server. Use the save() method on the object to write it on the server. kwargs: Arbitrary keyword arguments """ warnings.warn("`Group` is deprecated, use `groups` instead", DeprecationWarning) return Group._get_list_or_object(self, id, **kwargs) def Issue(self, id=None, **kwargs): """Lists issues(s) known by the GitLab server. Does not support creation or getting a single issue unlike other methods in this class yet. """ warnings.warn("`Issue` is deprecated, use `issues` instead", DeprecationWarning) return Issue._get_list_or_object(self, id, **kwargs) def User(self, id=None, **kwargs): """Creates/gets/lists users(s) known by the GitLab server. If id is None, returns a list of users. If id is an integer, returns the matching user (or raises a GitlabGetError if not found) If id is a dict, creates a new object using attributes provided. The object is NOT saved on the server. Use the save() method on the object to write it on the server. """ warnings.warn("`User` is deprecated, use `users` instead", DeprecationWarning) return User._get_list_or_object(self, id, **kwargs) def Team(self, id=None, **kwargs): """Creates/gets/lists team(s) known by the GitLab server. If id is None, returns a list of teams. If id is an integer, returns the matching team (or raises a GitlabGetError if not found) If id is a dict, create a new object using attributes provided. The object is NOT saved on the server. Use the save() method on the object to write it on the server. """ warnings.warn("`Team` is deprecated, use `teams` instead", DeprecationWarning) return Team._get_list_or_object(self, id, **kwargs)