# -*- 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 . import copy import importlib import itertools import json import sys import six import gitlab from gitlab.exceptions import * # noqa class jsonEncoder(json.JSONEncoder): def default(self, obj): if isinstance(obj, GitlabObject): return obj.as_dict() elif isinstance(obj, gitlab.Gitlab): return {'url': obj._url} return json.JSONEncoder.default(self, obj) class BaseManager(object): """Base manager class for API operations. Managers provide method to manage GitLab API objects, such as retrieval, listing, creation. Inherited class must define the ``obj_cls`` attribute. Attributes: obj_cls (class): class of objects wrapped by this manager. """ obj_cls = None def __init__(self, gl, parent=None, args=[]): """Constructs a manager. Args: gl (gitlab.Gitlab): Gitlab object referencing the GitLab server. parent (Optional[Manager]): A parent manager. args (list): A list of tuples defining a link between the parent/child attributes. Raises: AttributeError: If `obj_cls` is None. """ self.gitlab = gl self.args = args self.parent = parent if self.obj_cls is None: raise AttributeError("obj_cls must be defined") def _set_parent_args(self, **kwargs): args = copy.copy(kwargs) if self.parent is not None: for attr, parent_attr in self.args: args.setdefault(attr, getattr(self.parent, parent_attr)) return args def get(self, id=None, **kwargs): """Get a GitLab object. Args: id: ID of the object to retrieve. **kwargs: Additional arguments to send to GitLab. Returns: object: An object of class `obj_cls`. Raises: NotImplementedError: If objects cannot be retrieved. GitlabGetError: If the server fails to perform the request. """ args = self._set_parent_args(**kwargs) if not self.obj_cls.canGet: raise NotImplementedError if id is None and self.obj_cls.getRequiresId is True: raise ValueError('The id argument must be defined.') return self.obj_cls.get(self.gitlab, id, **args) def list(self, **kwargs): """Get a list of GitLab objects. Args: **kwargs: Additional arguments to send to GitLab. Returns: list[object]: A list of `obj_cls` objects. Raises: NotImplementedError: If objects cannot be listed. GitlabListError: If the server fails to perform the request. """ args = self._set_parent_args(**kwargs) if not self.obj_cls.canList: raise NotImplementedError return self.obj_cls.list(self.gitlab, **args) def create(self, data, **kwargs): """Create a new object of class `obj_cls`. Args: data (dict): The parameters to send to the GitLab server to create the object. Required and optional arguments are defined in the `requiredCreateAttrs` and `optionalCreateAttrs` of the `obj_cls` class. **kwargs: Additional arguments to send to GitLab. Returns: object: A newly create `obj_cls` object. Raises: NotImplementedError: If objects cannot be created. GitlabCreateError: If the server fails to perform the request. """ args = self._set_parent_args(**kwargs) if not self.obj_cls.canCreate: raise NotImplementedError return self.obj_cls.create(self.gitlab, data, **args) def delete(self, id, **kwargs): """Delete a GitLab object. Args: id: ID of the object to delete. Raises: NotImplementedError: If objects cannot be deleted. GitlabDeleteError: If the server fails to perform the request. """ args = self._set_parent_args(**kwargs) if not self.obj_cls.canDelete: raise NotImplementedError self.gitlab.delete(self.obj_cls, id, **args) class GitlabObject(object): """Base class for all classes that interface with GitLab.""" #: Url to use in GitLab for this object _url = None # Some objects (e.g. merge requests) have different urls for singular and # plural _urlPlural = None _id_in_delete_url = True _id_in_update_url = True _constructorTypes = None #: 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 optional when retrieving list of objects. optionalListAttrs = [] #: Attributes that are optional when retrieving single object. optionalGetAttrs = [] #: 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 = [] #: Attributes that are required when updating an object. requiredUpdateAttrs = [] #: Attributes that are optional when updating an object. optionalUpdateAttrs = [] #: Whether the object ID is required in the GET url. getRequiresId = True #: List of managers to create. managers = [] #: Name of the identifier of an object. idAttr = 'id' #: Attribute to use as ID when displaying the object. shortPrintAttr = None def _data_for_gitlab(self, extra_parameters={}, update=False, as_json=True): data = {} if update and (self.requiredUpdateAttrs or self.optionalUpdateAttrs): attributes = itertools.chain(self.requiredUpdateAttrs, self.optionalUpdateAttrs) else: attributes = itertools.chain(self.requiredCreateAttrs, self.optionalCreateAttrs) attributes = list(attributes) + ['sudo', 'page', 'per_page'] for attribute in attributes: if hasattr(self, attribute): value = getattr(self, attribute) # labels need to be sent as a comma-separated list if attribute == 'labels' and isinstance(value, list): value = ", ".join(value) elif attribute == 'sudo': value = str(value) data[attribute] = value data.update(extra_parameters) return json.dumps(data) if as_json else data @classmethod def list(cls, gl, **kwargs): """Retrieve a list of objects from GitLab. Args: gl (gitlab.Gitlab): Gitlab object referencing the GitLab server. per_page (int): Maximum number of items to return. page (int): ID of the page to return when using pagination. Returns: list[object]: A list of objects. Raises: NotImplementedError: If objects can't be listed. GitlabListError: If the server cannot perform the request. """ if not cls.canList: raise NotImplementedError if not cls._url: raise NotImplementedError return gl.list(cls, **kwargs) @classmethod def get(cls, gl, id, **kwargs): """Retrieve a single object. Args: gl (gitlab.Gitlab): Gitlab object referencing the GitLab server. id (int or str): ID of the object to retrieve. Returns: object: The found GitLab object. Raises: NotImplementedError: If objects can't be retrieved. GitlabGetError: If the server cannot perform the request. """ if cls.canGet is False: raise NotImplementedError elif cls.canGet is True: return cls(gl, id, **kwargs) elif cls.canGet == 'from_list': for obj in cls.list(gl, **kwargs): obj_id = getattr(obj, obj.idAttr) if str(obj_id) == str(id): return obj raise GitlabGetError("Object not found") def _get_object(self, k, v, **kwargs): if self._constructorTypes and k in self._constructorTypes: cls = getattr(self._module, self._constructorTypes[k]) return cls(self.gitlab, v, **kwargs) else: return v def _set_from_dict(self, data, **kwargs): if not hasattr(data, 'items'): return for k, v in data.items(): # If a k attribute already exists and is a Manager, do nothing (see # https://github.com/python-gitlab/python-gitlab/issues/209) if isinstance(getattr(self, k, None), BaseManager): continue if isinstance(v, list): self.__dict__[k] = [] for i in v: self.__dict__[k].append(self._get_object(k, i, **kwargs)) elif v is None: self.__dict__[k] = None else: self.__dict__[k] = self._get_object(k, v, **kwargs) def _create(self, **kwargs): if not self.canCreate: raise NotImplementedError json = self.gitlab.create(self, **kwargs) self._set_from_dict(json) self._from_api = True def _update(self, **kwargs): if not self.canUpdate: raise NotImplementedError json = self.gitlab.update(self, **kwargs) self._set_from_dict(json) def save(self, **kwargs): if self._from_api: self._update(**kwargs) else: self._create(**kwargs) def delete(self, **kwargs): if not self.canDelete: raise NotImplementedError if not self._from_api: raise GitlabDeleteError("Object not yet created") return self.gitlab.delete(self, **kwargs) @classmethod def create(cls, gl, data, **kwargs): """Create an object. Args: gl (gitlab.Gitlab): Gitlab object referencing the GitLab server. data (dict): The data used to define the object. Returns: object: The new object. Raises: NotImplementedError: If objects can't be created. GitlabCreateError: If the server cannot perform the request. """ if not cls.canCreate: raise NotImplementedError obj = cls(gl, data, **kwargs) obj.save() return obj def __init__(self, gl, data=None, **kwargs): """Constructs a new object. Do not use this method. Use the `get` or `create` class methods instead. Args: gl (gitlab.Gitlab): Gitlab object referencing the GitLab server. data: If `data` is a dict, create a new object using the information. If it is an int or a string, get a GitLab object from an API request. **kwargs: Additional arguments to send to GitLab. """ self._from_api = False #: (gitlab.Gitlab): Gitlab connection. self.gitlab = gl # store the module in which the object has been created (v3/v4) to be # able to reference other objects from the same module self._module = importlib.import_module(self.__module__) 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) self._from_api = True # the API returned a list because custom kwargs where used # instead of the id to request an object. Usually parameters # other than an id return ambiguous results. However in the # gitlab universe iids together with a project_id are # unambiguous for merge requests and issues, too. # So if there is only one element we can use it as our data # source. if 'iid' in kwargs and isinstance(data, list): if len(data) < 1: raise GitlabGetError('Not found') elif len(data) == 1: data = data[0] else: raise GitlabGetError('Impossible! You found multiple' ' elements with the same iid.') self._set_from_dict(data, **kwargs) if kwargs: for k, v in kwargs.items(): # Don't overwrite attributes returned by the server (#171) if k not in self.__dict__ or not self.__dict__[k]: 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 _set_manager(self, var, cls, attrs): manager = cls(self.gitlab, self, attrs) setattr(self, var, manager) def __getattr__(self, name): # build a manager if it doesn't exist yet for var, cls, attrs in self.managers: if var != name: continue # Build the full class path if needed if isinstance(cls, six.string_types): cls = getattr(self._module, cls) self._set_manager(var, cls, attrs) return getattr(self, var) raise AttributeError(name) def __str__(self): return '%s => %s' % (type(self), str(self.__dict__)) def __repr__(self): return '<%s %s:%s>' % (self.__class__.__name__, self.idAttr, getattr(self, self.idAttr)) def display(self, pretty): if pretty: self.pretty_print() else: self.short_print() def short_print(self, depth=0): """Print the object on the standard output (verbose). Args: depth (int): Used internaly for recursive call. """ 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 _get_display_encoding(): return sys.stdout.encoding or sys.getdefaultencoding() @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(GitlabObject._get_display_encoding(), "replace") else: return str(obj) def pretty_print(self, depth=0): """Print the object on the standard output (verbose). Args: depth (int): Used internaly for recursive call. """ id = self.__dict__[self.idAttr] print("%s%s: %s" % (" " * depth * 2, self.idAttr, id)) for k in sorted(self.__dict__.keys()): if k in (self.idAttr, 'id', 'gitlab'): continue if k[0] == '_': continue v = self.__dict__[k] pretty_k = k.replace('_', '-') if six.PY2: pretty_k = pretty_k.encode( GitlabObject._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)) elif isinstance(v, BaseManager): continue else: if hasattr(v, __name__) and v.__name__ == 'Gitlab': continue v = GitlabObject._obj_to_str(v) print("%s%s: %s" % (" " * depth * 2, pretty_k, v)) def json(self): """Dump the object as json. Returns: str: The json string. """ return json.dumps(self, cls=jsonEncoder) def as_dict(self): """Dump the object as a dict.""" return {k: v for k, v in six.iteritems(self.__dict__) if (not isinstance(v, BaseManager) and not k[0] == '_')} def __eq__(self, other): if type(other) is type(self): return self.as_dict() == other.as_dict() return False def __ne__(self, other): return not self.__eq__(other)