# -*- 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 . from types import ModuleType from typing import ( Any, Callable, Dict, List, Optional, Tuple, Type, TYPE_CHECKING, Union, ) import requests import gitlab from gitlab import base, cli from gitlab import exceptions as exc from gitlab import types as g_types from gitlab import utils __all__ = [ "GetMixin", "GetWithoutIdMixin", "RefreshMixin", "ListMixin", "RetrieveMixin", "CreateMixin", "UpdateMixin", "SetMixin", "DeleteMixin", "CRUDMixin", "NoUpdateMixin", "SaveMixin", "ObjectDeleteMixin", "UserAgentDetailMixin", "AccessRequestMixin", "DownloadMixin", "SubscribableMixin", "TodoMixin", "TimeTrackingMixin", "ParticipantsMixin", "BadgeRenderMixin", ] if TYPE_CHECKING: # When running mypy we use these as the base classes _RestManagerBase = base.RESTManager _RestObjectBase = base.RESTObject else: _RestManagerBase = object _RestObjectBase = object class GetMixin(_RestManagerBase): _computed_path: Optional[str] _from_parent_attrs: Dict[str, Any] _obj_cls: Optional[Type[base.RESTObject]] _optional_get_attrs: Tuple[str, ...] = () _parent: Optional[base.RESTObject] _parent_attrs: Dict[str, Any] _path: Optional[str] gitlab: gitlab.Gitlab @exc.on_http_error(exc.GitlabGetError) def get( self, id: Union[str, int], lazy: bool = False, **kwargs: Any ) -> base.RESTObject: """Retrieve a single object. Args: id (int or str): ID of the object to retrieve lazy: If True, don't request the server, but create a shallow object giving access to the managers. This is useful if you want to avoid useless calls to the API. **kwargs: Extra options to send to the server (e.g. sudo) Returns: The generated RESTObject. Raises: GitlabAuthenticationError: If authentication is not correct GitlabGetError: If the server cannot perform the request """ if not isinstance(id, int): id = utils.clean_str_id(id) path = f"{self.path}/{id}" if TYPE_CHECKING: assert self._obj_cls is not None if lazy is True: if TYPE_CHECKING: assert self._obj_cls._id_attr is not None return self._obj_cls(self, {self._obj_cls._id_attr: id}) server_data = self.gitlab.http_get(path, **kwargs) if TYPE_CHECKING: assert not isinstance(server_data, requests.Response) return self._obj_cls(self, server_data) class GetWithoutIdMixin(_RestManagerBase): _computed_path: Optional[str] _from_parent_attrs: Dict[str, Any] _obj_cls: Optional[Type[base.RESTObject]] _optional_get_attrs: Tuple[str, ...] = () _parent: Optional[base.RESTObject] _parent_attrs: Dict[str, Any] _path: Optional[str] gitlab: gitlab.Gitlab @exc.on_http_error(exc.GitlabGetError) def get( self, id: Optional[Union[int, str]] = None, **kwargs: Any ) -> Optional[base.RESTObject]: """Retrieve a single object. Args: **kwargs: Extra options to send to the server (e.g. sudo) Returns: The generated RESTObject Raises: GitlabAuthenticationError: If authentication is not correct GitlabGetError: If the server cannot perform the request """ if TYPE_CHECKING: assert self.path is not None server_data = self.gitlab.http_get(self.path, **kwargs) if server_data is None: return None if TYPE_CHECKING: assert not isinstance(server_data, requests.Response) assert self._obj_cls is not None return self._obj_cls(self, server_data) class RefreshMixin(_RestObjectBase): _id_attr: Optional[str] _attrs: Dict[str, Any] _module: ModuleType _parent_attrs: Dict[str, Any] _updated_attrs: Dict[str, Any] manager: base.RESTManager @exc.on_http_error(exc.GitlabGetError) def refresh(self, **kwargs: Any) -> None: """Refresh a single object from server. Args: **kwargs: Extra options to send to the server (e.g. sudo) Returns None (updates the object) Raises: GitlabAuthenticationError: If authentication is not correct GitlabGetError: If the server cannot perform the request """ if self._id_attr: path = f"{self.manager.path}/{self.id}" else: if TYPE_CHECKING: assert self.manager.path is not None path = self.manager.path server_data = self.manager.gitlab.http_get(path, **kwargs) if TYPE_CHECKING: assert not isinstance(server_data, requests.Response) self._update_attrs(server_data) class ListMixin(_RestManagerBase): _computed_path: Optional[str] _from_parent_attrs: Dict[str, Any] _list_filters: Tuple[str, ...] = () _obj_cls: Optional[Type[base.RESTObject]] _parent: Optional[base.RESTObject] _parent_attrs: Dict[str, Any] _path: Optional[str] gitlab: gitlab.Gitlab @exc.on_http_error(exc.GitlabListError) def list(self, **kwargs: Any) -> Union[base.RESTObjectList, List[base.RESTObject]]: """Retrieve a list of objects. Args: all: If True, return all the items, without pagination per_page: Number of items to retrieve per request page: ID of the page to return (starts with page 1) as_list: If set to False and no pagination option is defined, return a generator instead of a list **kwargs: Extra options to send to the server (e.g. sudo) Returns: The list of objects, or a generator if `as_list` is False Raises: GitlabAuthenticationError: If authentication is not correct GitlabListError: If the server cannot perform the request """ # Duplicate data to avoid messing with what the user sent us data = kwargs.copy() if self.gitlab.per_page: data.setdefault("per_page", self.gitlab.per_page) # global keyset pagination if self.gitlab.pagination: data.setdefault("pagination", self.gitlab.pagination) if self.gitlab.order_by: data.setdefault("order_by", self.gitlab.order_by) # We get the attributes that need some special transformation if self._types: for attr_name, type_cls in self._types.items(): if attr_name in data.keys(): type_obj = type_cls(data[attr_name]) data[attr_name] = type_obj.get_for_api() # Allow to overwrite the path, handy for custom listings path = data.pop("path", self.path) if TYPE_CHECKING: assert self._obj_cls is not None obj = self.gitlab.http_list(path, **data) if isinstance(obj, list): return [self._obj_cls(self, item) for item in obj] else: return base.RESTObjectList(self, self._obj_cls, obj) class RetrieveMixin(ListMixin, GetMixin): _computed_path: Optional[str] _from_parent_attrs: Dict[str, Any] _obj_cls: Optional[Type[base.RESTObject]] _parent: Optional[base.RESTObject] _parent_attrs: Dict[str, Any] _path: Optional[str] gitlab: gitlab.Gitlab pass class CreateMixin(_RestManagerBase): _computed_path: Optional[str] _from_parent_attrs: Dict[str, Any] _obj_cls: Optional[Type[base.RESTObject]] _parent: Optional[base.RESTObject] _parent_attrs: Dict[str, Any] _path: Optional[str] gitlab: gitlab.Gitlab def _check_missing_create_attrs(self, data: Dict[str, Any]) -> None: missing = [] for attr in self._create_attrs.required: if attr not in data: missing.append(attr) continue if missing: raise AttributeError(f"Missing attributes: {', '.join(missing)}") @exc.on_http_error(exc.GitlabCreateError) def create( self, data: Optional[Dict[str, Any]] = None, **kwargs: Any ) -> base.RESTObject: """Create a new object. Args: data: parameters to send to the server to create the resource **kwargs: Extra options to send to the server (e.g. sudo) Returns: a new instance of the managed object class built with the data sent by the server Raises: GitlabAuthenticationError: If authentication is not correct GitlabCreateError: If the server cannot perform the request """ if data is None: data = {} self._check_missing_create_attrs(data) files = {} # We get the attributes that need some special transformation if self._types: # Duplicate data to avoid messing with what the user sent us data = data.copy() for attr_name, type_cls in self._types.items(): if attr_name in data.keys(): type_obj = type_cls(data[attr_name]) # if the type if FileAttribute we need to pass the data as # file if isinstance(type_obj, g_types.FileAttribute): k = type_obj.get_file_name(attr_name) files[attr_name] = (k, data.pop(attr_name)) else: data[attr_name] = type_obj.get_for_api() # Handle specific URL for creation path = kwargs.pop("path", self.path) server_data = self.gitlab.http_post(path, post_data=data, files=files, **kwargs) if TYPE_CHECKING: assert not isinstance(server_data, requests.Response) assert self._obj_cls is not None return self._obj_cls(self, server_data) class UpdateMixin(_RestManagerBase): _computed_path: Optional[str] _from_parent_attrs: Dict[str, Any] _obj_cls: Optional[Type[base.RESTObject]] _parent: Optional[base.RESTObject] _parent_attrs: Dict[str, Any] _path: Optional[str] _update_uses_post: bool = False gitlab: gitlab.Gitlab def _check_missing_update_attrs(self, data: Dict[str, Any]) -> None: if TYPE_CHECKING: assert self._obj_cls is not None # Remove the id field from the required list as it was previously moved # to the http path. required = tuple( [k for k in self._update_attrs.required if k != self._obj_cls._id_attr] ) missing = [] for attr in required: if attr not in data: missing.append(attr) continue if missing: raise AttributeError(f"Missing attributes: {', '.join(missing)}") def _get_update_method( self, ) -> Callable[..., Union[Dict[str, Any], requests.Response]]: """Return the HTTP method to use. Returns: http_put (default) or http_post """ if self._update_uses_post: http_method = self.gitlab.http_post else: http_method = self.gitlab.http_put return http_method @exc.on_http_error(exc.GitlabUpdateError) def update( self, id: Optional[Union[str, int]] = None, new_data: Optional[Dict[str, Any]] = None, **kwargs: Any, ) -> Dict[str, Any]: """Update an object on the server. Args: id: ID of the object to update (can be None if not required) new_data: the update data for the object **kwargs: Extra options to send to the server (e.g. sudo) Returns: The new object data (*not* a RESTObject) Raises: GitlabAuthenticationError: If authentication is not correct GitlabUpdateError: If the server cannot perform the request """ new_data = new_data or {} if id is None: path = self.path else: path = f"{self.path}/{id}" self._check_missing_update_attrs(new_data) files = {} # We get the attributes that need some special transformation if self._types: # Duplicate data to avoid messing with what the user sent us new_data = new_data.copy() for attr_name, type_cls in self._types.items(): if attr_name in new_data.keys(): type_obj = type_cls(new_data[attr_name]) # if the type if FileAttribute we need to pass the data as # file if isinstance(type_obj, g_types.FileAttribute): k = type_obj.get_file_name(attr_name) files[attr_name] = (k, new_data.pop(attr_name)) else: new_data[attr_name] = type_obj.get_for_api() http_method = self._get_update_method() result = http_method(path, post_data=new_data, files=files, **kwargs) if TYPE_CHECKING: assert not isinstance(result, requests.Response) return result class SetMixin(_RestManagerBase): _computed_path: Optional[str] _from_parent_attrs: Dict[str, Any] _obj_cls: Optional[Type[base.RESTObject]] _parent: Optional[base.RESTObject] _parent_attrs: Dict[str, Any] _path: Optional[str] gitlab: gitlab.Gitlab @exc.on_http_error(exc.GitlabSetError) def set(self, key: str, value: str, **kwargs: Any) -> base.RESTObject: """Create or update the object. Args: key: The key of the object to create/update value: The value to set for the object **kwargs: Extra options to send to the server (e.g. sudo) Raises: GitlabAuthenticationError: If authentication is not correct GitlabSetError: If an error occurred Returns: The created/updated attribute """ path = f"{self.path}/{utils.clean_str_id(key)}" data = {"value": value} server_data = self.gitlab.http_put(path, post_data=data, **kwargs) if TYPE_CHECKING: assert not isinstance(server_data, requests.Response) assert self._obj_cls is not None return self._obj_cls(self, server_data) class DeleteMixin(_RestManagerBase): _computed_path: Optional[str] _from_parent_attrs: Dict[str, Any] _obj_cls: Optional[Type[base.RESTObject]] _parent: Optional[base.RESTObject] _parent_attrs: Dict[str, Any] _path: Optional[str] gitlab: gitlab.Gitlab @exc.on_http_error(exc.GitlabDeleteError) def delete(self, id: Union[str, int], **kwargs: Any) -> None: """Delete an object on the server. Args: id: ID of the object to delete **kwargs: Extra options to send to the server (e.g. sudo) Raises: GitlabAuthenticationError: If authentication is not correct GitlabDeleteError: If the server cannot perform the request """ if id is None: path = self.path else: if not isinstance(id, int): id = utils.clean_str_id(id) path = f"{self.path}/{id}" self.gitlab.http_delete(path, **kwargs) class CRUDMixin(GetMixin, ListMixin, CreateMixin, UpdateMixin, DeleteMixin): _computed_path: Optional[str] _from_parent_attrs: Dict[str, Any] _obj_cls: Optional[Type[base.RESTObject]] _parent: Optional[base.RESTObject] _parent_attrs: Dict[str, Any] _path: Optional[str] gitlab: gitlab.Gitlab pass class NoUpdateMixin(GetMixin, ListMixin, CreateMixin, DeleteMixin): _computed_path: Optional[str] _from_parent_attrs: Dict[str, Any] _obj_cls: Optional[Type[base.RESTObject]] _parent: Optional[base.RESTObject] _parent_attrs: Dict[str, Any] _path: Optional[str] gitlab: gitlab.Gitlab pass class SaveMixin(_RestObjectBase): """Mixin for RESTObject's that can be updated.""" _id_attr: Optional[str] _attrs: Dict[str, Any] _module: ModuleType _parent_attrs: Dict[str, Any] _updated_attrs: Dict[str, Any] manager: base.RESTManager def _get_updated_data(self) -> Dict[str, Any]: updated_data = {} for attr in self.manager._update_attrs.required: # Get everything required, no matter if it's been updated updated_data[attr] = getattr(self, attr) # Add the updated attributes updated_data.update(self._updated_attrs) return updated_data def save(self, **kwargs: Any) -> None: """Save the changes made to the object to the server. The object is updated to match what the server returns. Args: **kwargs: Extra options to send to the server (e.g. sudo) Raise: GitlabAuthenticationError: If authentication is not correct GitlabUpdateError: If the server cannot perform the request """ updated_data = self._get_updated_data() # Nothing to update. Server fails if sent an empty dict. if not updated_data: return # call the manager obj_id = self.get_id() if TYPE_CHECKING: assert isinstance(self.manager, UpdateMixin) server_data = self.manager.update(obj_id, updated_data, **kwargs) if server_data is not None: self._update_attrs(server_data) class ObjectDeleteMixin(_RestObjectBase): """Mixin for RESTObject's that can be deleted.""" _id_attr: Optional[str] _attrs: Dict[str, Any] _module: ModuleType _parent_attrs: Dict[str, Any] _updated_attrs: Dict[str, Any] manager: base.RESTManager def delete(self, **kwargs: Any) -> None: """Delete the object from the server. Args: **kwargs: Extra options to send to the server (e.g. sudo) Raises: GitlabAuthenticationError: If authentication is not correct GitlabDeleteError: If the server cannot perform the request """ if TYPE_CHECKING: assert isinstance(self.manager, DeleteMixin) self.manager.delete(self.get_id(), **kwargs) class UserAgentDetailMixin(_RestObjectBase): _id_attr: Optional[str] _attrs: Dict[str, Any] _module: ModuleType _parent_attrs: Dict[str, Any] _updated_attrs: Dict[str, Any] manager: base.RESTManager @cli.register_custom_action(("Snippet", "ProjectSnippet", "ProjectIssue")) @exc.on_http_error(exc.GitlabGetError) def user_agent_detail(self, **kwargs: Any) -> Dict[str, Any]: """Get the user agent detail. 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 """ path = f"{self.manager.path}/{self.get_id()}/user_agent_detail" result = self.manager.gitlab.http_get(path, **kwargs) if TYPE_CHECKING: assert not isinstance(result, requests.Response) return result class AccessRequestMixin(_RestObjectBase): _id_attr: Optional[str] _attrs: Dict[str, Any] _module: ModuleType _parent_attrs: Dict[str, Any] _updated_attrs: Dict[str, Any] manager: base.RESTManager @cli.register_custom_action( ("ProjectAccessRequest", "GroupAccessRequest"), tuple(), ("access_level",) ) @exc.on_http_error(exc.GitlabUpdateError) def approve( self, access_level: int = gitlab.DEVELOPER_ACCESS, **kwargs: Any ) -> None: """Approve an access request. Args: access_level: The access level for the user **kwargs: Extra options to send to the server (e.g. sudo) Raises: GitlabAuthenticationError: If authentication is not correct GitlabUpdateError: If the server fails to perform the request """ path = f"{self.manager.path}/{self.id}/approve" data = {"access_level": access_level} server_data = self.manager.gitlab.http_put(path, post_data=data, **kwargs) if TYPE_CHECKING: assert not isinstance(server_data, requests.Response) self._update_attrs(server_data) class DownloadMixin(_RestObjectBase): _id_attr: Optional[str] _attrs: Dict[str, Any] _module: ModuleType _parent_attrs: Dict[str, Any] _updated_attrs: Dict[str, Any] manager: base.RESTManager @cli.register_custom_action(("GroupExport", "ProjectExport")) @exc.on_http_error(exc.GitlabGetError) def download( self, streamed: bool = False, action: Optional[Callable] = None, chunk_size: int = 1024, **kwargs: Any, ) -> Optional[bytes]: """Download the archive of a resource export. Args: streamed: If True the data will be processed by chunks of `chunk_size` and each chunk is passed to `action` for treatment action (callable): Callable responsible of dealing with chunk of data chunk_size: Size of each chunk **kwargs: Extra options to send to the server (e.g. sudo) Raises: GitlabAuthenticationError: If authentication is not correct GitlabGetError: If the server failed to perform the request Returns: The blob content if streamed is False, None otherwise """ path = f"{self.manager.path}/download" result = self.manager.gitlab.http_get( path, streamed=streamed, raw=True, **kwargs ) if TYPE_CHECKING: assert isinstance(result, requests.Response) return utils.response_content(result, streamed, action, chunk_size) class SubscribableMixin(_RestObjectBase): _id_attr: Optional[str] _attrs: Dict[str, Any] _module: ModuleType _parent_attrs: Dict[str, Any] _updated_attrs: Dict[str, Any] manager: base.RESTManager @cli.register_custom_action( ("ProjectIssue", "ProjectMergeRequest", "ProjectLabel", "GroupLabel") ) @exc.on_http_error(exc.GitlabSubscribeError) def subscribe(self, **kwargs: Any) -> None: """Subscribe to the object notifications. Args: **kwargs: Extra options to send to the server (e.g. sudo) raises: GitlabAuthenticationError: If authentication is not correct GitlabSubscribeError: If the subscription cannot be done """ path = f"{self.manager.path}/{self.get_id()}/subscribe" server_data = self.manager.gitlab.http_post(path, **kwargs) if TYPE_CHECKING: assert not isinstance(server_data, requests.Response) self._update_attrs(server_data) @cli.register_custom_action( ("ProjectIssue", "ProjectMergeRequest", "ProjectLabel", "GroupLabel") ) @exc.on_http_error(exc.GitlabUnsubscribeError) def unsubscribe(self, **kwargs: Any) -> None: """Unsubscribe from the object notifications. Args: **kwargs: Extra options to send to the server (e.g. sudo) raises: GitlabAuthenticationError: If authentication is not correct GitlabUnsubscribeError: If the unsubscription cannot be done """ path = f"{self.manager.path}/{self.get_id()}/unsubscribe" server_data = self.manager.gitlab.http_post(path, **kwargs) if TYPE_CHECKING: assert not isinstance(server_data, requests.Response) self._update_attrs(server_data) class TodoMixin(_RestObjectBase): _id_attr: Optional[str] _attrs: Dict[str, Any] _module: ModuleType _parent_attrs: Dict[str, Any] _updated_attrs: Dict[str, Any] manager: base.RESTManager @cli.register_custom_action(("ProjectIssue", "ProjectMergeRequest")) @exc.on_http_error(exc.GitlabTodoError) def todo(self, **kwargs: Any) -> None: """Create a todo associated to the object. Args: **kwargs: Extra options to send to the server (e.g. sudo) Raises: GitlabAuthenticationError: If authentication is not correct GitlabTodoError: If the todo cannot be set """ path = f"{self.manager.path}/{self.get_id()}/todo" self.manager.gitlab.http_post(path, **kwargs) class TimeTrackingMixin(_RestObjectBase): _id_attr: Optional[str] _attrs: Dict[str, Any] _module: ModuleType _parent_attrs: Dict[str, Any] _updated_attrs: Dict[str, Any] manager: base.RESTManager @cli.register_custom_action(("ProjectIssue", "ProjectMergeRequest")) @exc.on_http_error(exc.GitlabTimeTrackingError) def time_stats(self, **kwargs: Any) -> Dict[str, Any]: """Get time stats for the object. Args: **kwargs: Extra options to send to the server (e.g. sudo) Raises: GitlabAuthenticationError: If authentication is not correct GitlabTimeTrackingError: If the time tracking update cannot be done """ # Use the existing time_stats attribute if it exist, otherwise make an # API call if "time_stats" in self.attributes: return self.attributes["time_stats"] path = f"{self.manager.path}/{self.get_id()}/time_stats" result = self.manager.gitlab.http_get(path, **kwargs) if TYPE_CHECKING: assert not isinstance(result, requests.Response) return result @cli.register_custom_action(("ProjectIssue", "ProjectMergeRequest"), ("duration",)) @exc.on_http_error(exc.GitlabTimeTrackingError) def time_estimate(self, duration: str, **kwargs: Any) -> Dict[str, Any]: """Set an estimated time of work for the object. Args: duration: Duration in human format (e.g. 3h30) **kwargs: Extra options to send to the server (e.g. sudo) Raises: GitlabAuthenticationError: If authentication is not correct GitlabTimeTrackingError: If the time tracking update cannot be done """ path = f"{self.manager.path}/{self.get_id()}/time_estimate" data = {"duration": duration} result = self.manager.gitlab.http_post(path, post_data=data, **kwargs) if TYPE_CHECKING: assert not isinstance(result, requests.Response) return result @cli.register_custom_action(("ProjectIssue", "ProjectMergeRequest")) @exc.on_http_error(exc.GitlabTimeTrackingError) def reset_time_estimate(self, **kwargs: Any) -> Dict[str, Any]: """Resets estimated time for the object to 0 seconds. Args: **kwargs: Extra options to send to the server (e.g. sudo) Raises: GitlabAuthenticationError: If authentication is not correct GitlabTimeTrackingError: If the time tracking update cannot be done """ path = f"{self.manager.path}/{self.get_id()}/reset_time_estimate" result = self.manager.gitlab.http_post(path, **kwargs) if TYPE_CHECKING: assert not isinstance(result, requests.Response) return result @cli.register_custom_action(("ProjectIssue", "ProjectMergeRequest"), ("duration",)) @exc.on_http_error(exc.GitlabTimeTrackingError) def add_spent_time(self, duration: str, **kwargs: Any) -> Dict[str, Any]: """Add time spent working on the object. Args: duration: Duration in human format (e.g. 3h30) **kwargs: Extra options to send to the server (e.g. sudo) Raises: GitlabAuthenticationError: If authentication is not correct GitlabTimeTrackingError: If the time tracking update cannot be done """ path = f"{self.manager.path}/{self.get_id()}/add_spent_time" data = {"duration": duration} result = self.manager.gitlab.http_post(path, post_data=data, **kwargs) if TYPE_CHECKING: assert not isinstance(result, requests.Response) return result @cli.register_custom_action(("ProjectIssue", "ProjectMergeRequest")) @exc.on_http_error(exc.GitlabTimeTrackingError) def reset_spent_time(self, **kwargs: Any) -> Dict[str, Any]: """Resets the time spent working on the object. Args: **kwargs: Extra options to send to the server (e.g. sudo) Raises: GitlabAuthenticationError: If authentication is not correct GitlabTimeTrackingError: If the time tracking update cannot be done """ path = f"{self.manager.path}/{self.get_id()}/reset_spent_time" result = self.manager.gitlab.http_post(path, **kwargs) if TYPE_CHECKING: assert not isinstance(result, requests.Response) return result class ParticipantsMixin(_RestObjectBase): _id_attr: Optional[str] _attrs: Dict[str, Any] _module: ModuleType _parent_attrs: Dict[str, Any] _updated_attrs: Dict[str, Any] manager: base.RESTManager @cli.register_custom_action(("ProjectMergeRequest", "ProjectIssue")) @exc.on_http_error(exc.GitlabListError) def participants(self, **kwargs: Any) -> Dict[str, Any]: """List the participants. Args: all: If True, return all the items, without pagination per_page: Number of items to retrieve per request page: ID of the page to return (starts with page 1) as_list: If set to False and no pagination option is defined, return a generator instead of a list **kwargs: Extra options to send to the server (e.g. sudo) Raises: GitlabAuthenticationError: If authentication is not correct GitlabListError: If the list could not be retrieved Returns: The list of participants """ path = f"{self.manager.path}/{self.get_id()}/participants" result = self.manager.gitlab.http_get(path, **kwargs) if TYPE_CHECKING: assert not isinstance(result, requests.Response) return result class BadgeRenderMixin(_RestManagerBase): @cli.register_custom_action( ("GroupBadgeManager", "ProjectBadgeManager"), ("link_url", "image_url") ) @exc.on_http_error(exc.GitlabRenderError) def render(self, link_url: str, image_url: str, **kwargs: Any) -> Dict[str, Any]: """Preview link_url and image_url after interpolation. Args: link_url: URL of the badge link image_url: URL of the badge image **kwargs: Extra options to send to the server (e.g. sudo) Raises: GitlabAuthenticationError: If authentication is not correct GitlabRenderError: If the rendering failed Returns: The rendering properties """ path = f"{self.path}/render" data = {"link_url": link_url, "image_url": image_url} result = self.gitlab.http_get(path, data, **kwargs) if TYPE_CHECKING: assert not isinstance(result, requests.Response) return result class PromoteMixin(_RestObjectBase): _id_attr: Optional[str] _attrs: Dict[str, Any] _module: ModuleType _parent_attrs: Dict[str, Any] _updated_attrs: Dict[str, Any] _update_uses_post: bool = False manager: base.RESTManager def _get_update_method( self, ) -> Callable[..., Union[Dict[str, Any], requests.Response]]: """Return the HTTP method to use. Returns: http_put (default) or http_post """ if self._update_uses_post: http_method = self.manager.gitlab.http_post else: http_method = self.manager.gitlab.http_put return http_method @exc.on_http_error(exc.GitlabPromoteError) def promote(self, **kwargs: Any) -> Dict[str, Any]: """Promote the item. Args: **kwargs: Extra options to send to the server (e.g. sudo) Raises: GitlabAuthenticationError: If authentication is not correct GitlabPromoteError: If the item could not be promoted GitlabParsingError: If the json data could not be parsed Returns: The updated object data (*not* a RESTObject) """ path = f"{self.manager.path}/{self.id}/promote" http_method = self._get_update_method() result = http_method(path, **kwargs) if TYPE_CHECKING: assert not isinstance(result, requests.Response) return result