# -*- coding: utf-8 -*-

#

# Copyright (C) 2013-2017 Gauvain Pocentek <gauvain@pocentek.net>

#

# 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 <http://www.gnu.org/licenses/>.

from __future__ import print_function

from __future__ import absolute_import

import base64

from gitlab.base import * # noqa

from gitlab import cli

from gitlab.exceptions import * # noqa

from gitlab.mixins import * # noqa

from gitlab import types

from gitlab import utils

VISIBILITY_PRIVATE = "private"

VISIBILITY_INTERNAL = "internal"

VISIBILITY_PUBLIC = "public"

ACCESS_GUEST = 10

ACCESS_REPORTER = 20

ACCESS_DEVELOPER = 30

ACCESS_MASTER = 40

ACCESS_OWNER = 50

class SidekiqManager(RESTManager):

"""Manager for the Sidekiq methods.

This manager doesn't actually manage objects but provides helper fonction

for the sidekiq metrics API.

"""

@cli.register_custom_action("SidekiqManager")

@exc.on_http_error(exc.GitlabGetError)

def queue_metrics(self, **kwargs):

"""Return the registred queues information.

Args:

**kwargs: Extra options to send to the server (e.g. sudo)

Raises:

GitlabAuthenticationError: If authentication is not correct

GitlabGetError: If the information couldn't be retrieved

Returns:

dict: Information about the Sidekiq queues

"""

return self.gitlab.http_get("/sidekiq/queue_metrics", **kwargs)

@cli.register_custom_action("SidekiqManager")

@exc.on_http_error(exc.GitlabGetError)

def process_metrics(self, **kwargs):

"""Return the registred sidekiq workers.

Args:

**kwargs: Extra options to send to the server (e.g. sudo)

Raises:

GitlabAuthenticationError: If authentication is not correct

GitlabGetError: If the information couldn't be retrieved

Returns:

dict: Information about the register Sidekiq worker

"""

return self.gitlab.http_get("/sidekiq/process_metrics", **kwargs)

@cli.register_custom_action("SidekiqManager")

@exc.on_http_error(exc.GitlabGetError)

def job_stats(self, **kwargs):

"""Return statistics about the jobs performed.

Args:

**kwargs: Extra options to send to the server (e.g. sudo)

Raises:

GitlabAuthenticationError: If authentication is not correct

GitlabGetError: If the information couldn't be retrieved

Returns:

dict: Statistics about the Sidekiq jobs performed

"""

return self.gitlab.http_get("/sidekiq/job_stats", **kwargs)

@cli.register_custom_action("SidekiqManager")

@exc.on_http_error(exc.GitlabGetError)

def compound_metrics(self, **kwargs):

"""Return all available metrics and statistics.

Args:

**kwargs: Extra options to send to the server (e.g. sudo)

Raises:

GitlabAuthenticationError: If authentication is not correct

GitlabGetError: If the information couldn't be retrieved

Returns:

dict: All available Sidekiq metrics and statistics

"""

return self.gitlab.http_get("/sidekiq/compound_metrics", **kwargs)

class Event(RESTObject):

_id_attr = None

_short_print_attr = "target_title"

class AuditEvent(RESTObject):

_id_attr = "id"

class AuditEventManager(ListMixin, RESTManager):

_path = "/audit_events"

_obj_cls = AuditEvent

_list_filters = ("created_after", "created_before", "entity_type", "entity_id")

class EventManager(ListMixin, RESTManager):

_path = "/events"

_obj_cls = Event

_list_filters = ("action", "target_type", "before", "after", "sort")

class UserActivities(RESTObject):

_id_attr = "username"

class UserStatus(RESTObject):

_id_attr = None

_short_print_attr = "message"

class UserStatusManager(GetWithoutIdMixin, RESTManager):

_path = "/users/%(user_id)s/status"

_obj_cls = UserStatus

_from_parent_attrs = {"user_id": "id"}

class UserActivitiesManager(ListMixin, RESTManager):

_path = "/user/activities"

_obj_cls = UserActivities

class UserCustomAttribute(ObjectDeleteMixin, RESTObject):

_id_attr = "key"

class UserCustomAttributeManager(RetrieveMixin, SetMixin, DeleteMixin, RESTManager):

_path = "/users/%(user_id)s/custom_attributes"

_obj_cls = UserCustomAttribute

_from_parent_attrs = {"user_id": "id"}

class UserEmail(ObjectDeleteMixin, RESTObject):

_short_print_attr = "email"

class UserEmailManager(RetrieveMixin, CreateMixin, DeleteMixin, RESTManager):

_path = "/users/%(user_id)s/emails"

_obj_cls = UserEmail

_from_parent_attrs = {"user_id": "id"}

_create_attrs = (("email",), tuple())

class UserEvent(Event):

pass

class UserEventManager(EventManager):

_path = "/users/%(user_id)s/events"

_obj_cls = UserEvent

_from_parent_attrs = {"user_id": "id"}

class UserGPGKey(ObjectDeleteMixin, RESTObject):

pass

class UserGPGKeyManager(RetrieveMixin, CreateMixin, DeleteMixin, RESTManager):

_path = "/users/%(user_id)s/gpg_keys"

_obj_cls = UserGPGKey

_from_parent_attrs = {"user_id": "id"}

_create_attrs = (("key",), tuple())

class UserKey(ObjectDeleteMixin, RESTObject):

pass

class UserKeyManager(ListMixin, CreateMixin, DeleteMixin, RESTManager):

_path = "/users/%(user_id)s/keys"

_obj_cls = UserKey

_from_parent_attrs = {"user_id": "id"}

_create_attrs = (("title", "key"), tuple())

class UserStatus(RESTObject):

pass

class UserStatusManager(GetWithoutIdMixin, RESTManager):

_path = "/users/%(user_id)s/status"

_obj_cls = UserStatus

_from_parent_attrs = {"user_id": "id"}

class UserImpersonationToken(ObjectDeleteMixin, RESTObject):

pass

class UserImpersonationTokenManager(NoUpdateMixin, RESTManager):

_path = "/users/%(user_id)s/impersonation_tokens"

_obj_cls = UserImpersonationToken

_from_parent_attrs = {"user_id": "id"}

_create_attrs = (("name", "scopes"), ("expires_at",))

_list_filters = ("state",)

class UserProject(RESTObject):

pass

class UserProjectManager(ListMixin, CreateMixin, RESTManager):

_path = "/projects/user/%(user_id)s"

_obj_cls = UserProject

_from_parent_attrs = {"user_id": "id"}

_create_attrs = (

("name",),

(

"default_branch",

"issues_enabled",

"wall_enabled",

"merge_requests_enabled",

"wiki_enabled",

"snippets_enabled",

"public",

"visibility",

"description",

"builds_enabled",

"public_builds",

"import_url",

"only_allow_merge_if_build_succeeds",

),

)

_list_filters = (

"archived",

"visibility",

"order_by",

"sort",

"search",

"simple",

"owned",

"membership",

"starred",

"statistics",

"with_issues_enabled",

"with_merge_requests_enabled",

)

def list(self, **kwargs):

"""Retrieve a list of objects.

Args:

all (bool): If True, return all the items, without pagination

per_page (int): Number of items to retrieve per request

page (int): ID of the page to return (starts with page 1)

as_list (bool): 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:

list: 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

"""

if self._parent:

path = "/users/%s/projects" % self._parent.id

else:

path = "/users/%s/projects" % kwargs["user_id"]

return ListMixin.list(self, path=path, **kwargs)

class User(SaveMixin, ObjectDeleteMixin, RESTObject):

_short_print_attr = "username"

_managers = (

("customattributes", "UserCustomAttributeManager"),

("emails", "UserEmailManager"),

("events", "UserEventManager"),

("gpgkeys", "UserGPGKeyManager"),

("impersonationtokens", "UserImpersonationTokenManager"),

("keys", "UserKeyManager"),

("projects", "UserProjectManager"),

("status", "UserStatusManager"),

)

@cli.register_custom_action("User")

@exc.on_http_error(exc.GitlabBlockError)

def block(self, **kwargs):

"""Block the user.

Args:

**kwargs: Extra options to send to the server (e.g. sudo)

Raises:

GitlabAuthenticationError: If authentication is not correct

GitlabBlockError: If the user could not be blocked

Returns:

bool: Whether the user status has been changed

"""

path = "/users/%s/block" % self.id

server_data = self.manager.gitlab.http_post(path, **kwargs)

if server_data is True:

self._attrs["state"] = "blocked"

return server_data

@cli.register_custom_action("User")

@exc.on_http_error(exc.GitlabUnblockError)

def unblock(self, **kwargs):

"""Unblock the user.

Args:

**kwargs: Extra options to send to the server (e.g. sudo)

Raises:

GitlabAuthenticationError: If authentication is not correct

GitlabUnblockError: If the user could not be unblocked

Returns:

bool: Whether the user status has been changed

"""

path = "/users/%s/unblock" % self.id

server_data = self.manager.gitlab.http_post(path, **kwargs)

if server_data is True:

self._attrs["state"] = "active"

return server_data

@cli.register_custom_action("User")

@exc.on_http_error(exc.GitlabDeactivateError)

def deactivate(self, **kwargs):

"""Deactivate the user.

Args:

**kwargs: Extra options to send to the server (e.g. sudo)

Raises:

GitlabAuthenticationError: If authentication is not correct

GitlabDeactivateError: If the user could not be deactivated

Returns:

bool: Whether the user status has been changed

"""

path = "/users/%s/deactivate" % self.id

server_data = self.manager.gitlab.http_post(path, **kwargs)

if server_data:

self._attrs["state"] = "deactivated"

return server_data

@cli.register_custom_action("User")

@exc.on_http_error(exc.GitlabActivateError)

def activate(self, **kwargs):

"""Activate the user.

Args:

**kwargs: Extra options to send to the server (e.g. sudo)

Raises:

GitlabAuthenticationError: If authentication is not correct

GitlabActivateError: If the user could not be activated

Returns:

bool: Whether the user status has been changed

"""

path = "/users/%s/activate" % self.id

server_data = self.manager.gitlab.http_post(path, **kwargs)

if server_data:

self._attrs["state"] = "active"

return server_data

class UserManager(CRUDMixin, RESTManager):

_path = "/users"

_obj_cls = User

_list_filters = (

"active",

"blocked",

"username",

"extern_uid",

"provider",

"external",

"search",

"custom_attributes",

"status",

)

_create_attrs = (

tuple(),

(

"email",

"username",

"name",

"password",

"reset_password",

"skype",

"linkedin",

"twitter",

"projects_limit",

"extern_uid",

"provider",

"bio",

"admin",

"can_create_group",

"website_url",

"skip_confirmation",

"external",

"organization",

"location",

"avatar",

),

)

_update_attrs = (

("email", "username", "name"),

(

"password",

"skype",

"linkedin",

"twitter",

"projects_limit",

"extern_uid",

"provider",

"bio",

"admin",

"can_create_group",

"website_url",

"skip_confirmation",

"external",

"organization",

"location",

"avatar",

),

)

_types = {"confirm": types.LowercaseStringAttribute, "avatar": types.ImageAttribute}

class CurrentUserEmail(ObjectDeleteMixin, RESTObject):

_short_print_attr = "email"

class CurrentUserEmailManager(RetrieveMixin, CreateMixin, DeleteMixin, RESTManager):

_path = "/user/emails"

_obj_cls = CurrentUserEmail

_create_attrs = (("email",), tuple())

class CurrentUserGPGKey(ObjectDeleteMixin, RESTObject):

pass

class CurrentUserGPGKeyManager(RetrieveMixin, CreateMixin, DeleteMixin, RESTManager):

_path = "/user/gpg_keys"

_obj_cls = CurrentUserGPGKey

_create_attrs = (("key",), tuple())

class CurrentUserKey(ObjectDeleteMixin, RESTObject):

_short_print_attr = "title"

class CurrentUserKeyManager(RetrieveMixin, CreateMixin, DeleteMixin, RESTManager):

_path = "/user/keys"

_obj_cls = CurrentUserKey

_create_attrs = (("title", "key"), tuple())

class CurrentUserStatus(SaveMixin, RESTObject):

_id_attr = None

_short_print_attr = "message"

class CurrentUserStatusManager(GetWithoutIdMixin, UpdateMixin, RESTManager):

_path = "/user/status"

_obj_cls = CurrentUserStatus

_update_attrs = (tuple(), ("emoji", "message"))

class CurrentUser(RESTObject):

_id_attr = None

_short_print_attr = "username"

_managers = (

("status", "CurrentUserStatusManager"),

("emails", "CurrentUserEmailManager"),

("gpgkeys", "CurrentUserGPGKeyManager"),

("keys", "CurrentUserKeyManager"),

)

class CurrentUserManager(GetWithoutIdMixin, RESTManager):

_path = "/user"

_obj_cls = CurrentUser

class ApplicationSettings(SaveMixin, RESTObject):

_id_attr = None

class ApplicationSettingsManager(GetWithoutIdMixin, UpdateMixin, RESTManager):

_path = "/application/settings"

_obj_cls = ApplicationSettings

_update_attrs = (

tuple(),

(

"id",

"default_projects_limit",

"signup_enabled",

"password_authentication_enabled_for_web",

"gravatar_enabled",

"sign_in_text",

"created_at",

"updated_at",

"home_page_url",

"default_branch_protection",

"restricted_visibility_levels",

"max_attachment_size",

"session_expire_delay",

"default_project_visibility",

"default_snippet_visibility",

"default_group_visibility",

"outbound_local_requests_whitelist",

"domain_whitelist",

"domain_blacklist_enabled",

"domain_blacklist",

"external_authorization_service_enabled",

"external_authorization_service_url",

"external_authorization_service_default_label",

"external_authorization_service_timeout",

"user_oauth_applications",

"after_sign_out_path",

"container_registry_token_expire_delay",

"repository_storages",

"plantuml_enabled",

"plantuml_url",

"terminal_max_session_time",

"polling_interval_multiplier",

"rsa_key_restriction",

"dsa_key_restriction",

"ecdsa_key_restriction",

"ed25519_key_restriction",

"first_day_of_week",

"enforce_terms",

"terms",

"performance_bar_allowed_group_id",

"instance_statistics_visibility_private",

"user_show_add_ssh_key_message",

"file_template_project_id",

"local_markdown_version",

"asset_proxy_enabled",

"asset_proxy_url",

"asset_proxy_whitelist",

"geo_node_allowed_ips",

"allow_local_requests_from_hooks_and_services",

"allow_local_requests_from_web_hooks_and_services",

"allow_local_requests_from_system_hooks",

),

)

@exc.on_http_error(exc.GitlabUpdateError)

def update(self, id=None, new_data=None, **kwargs):

"""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:

dict: 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 {}

data = new_data.copy()

if "domain_whitelist" in data and data["domain_whitelist"] is None:

data.pop("domain_whitelist")

super(ApplicationSettingsManager, self).update(id, data, **kwargs)

class BroadcastMessage(SaveMixin, ObjectDeleteMixin, RESTObject):

pass

class BroadcastMessageManager(CRUDMixin, RESTManager):

_path = "/broadcast_messages"

_obj_cls = BroadcastMessage

_create_attrs = (("message",), ("starts_at", "ends_at", "color", "font"))

_update_attrs = (tuple(), ("message", "starts_at", "ends_at", "color", "font"))

class DeployKey(RESTObject):

pass

class DeployKeyManager(ListMixin, RESTManager):

_path = "/deploy_keys"

_obj_cls = DeployKey

class NotificationSettings(SaveMixin, RESTObject):

_id_attr = None

class NotificationSettingsManager(GetWithoutIdMixin, UpdateMixin, RESTManager):

_path = "/notification_settings"

_obj_cls = NotificationSettings

_update_attrs = (

tuple(),

(

"level",

"notification_email",

"new_note",

"new_issue",

"reopen_issue",

"close_issue",

"reassign_issue",

"new_merge_request",

"reopen_merge_request",

"close_merge_request",

"reassign_merge_request",

"merge_merge_request",

),

)

class Dockerfile(RESTObject):

_id_attr = "name"

class DockerfileManager(RetrieveMixin, RESTManager):

_path = "/templates/dockerfiles"

_obj_cls = Dockerfile

class Feature(ObjectDeleteMixin, RESTObject):

_id_attr = "name"

class FeatureManager(ListMixin, DeleteMixin, RESTManager):

_path = "/features/"

_obj_cls = Feature

@exc.on_http_error(exc.GitlabSetError)

def set(self, name, value, feature_group=None, user=None, **kwargs):

"""Create or update the object.

Args:

name (str): The value to set for the object

value (bool/int): The value to set for the object

feature_group (str): A feature group name

user (str): A GitLab username

**kwargs: Extra options to send to the server (e.g. sudo)

Raises:

GitlabAuthenticationError: If authentication is not correct

GitlabSetError: If an error occured

Returns:

obj: The created/updated attribute

"""

path = "%s/%s" % (self.path, name.replace("/", "%2F"))

data = {"value": value, "feature_group": feature_group, "user": user}

server_data = self.gitlab.http_post(path, post_data=data, **kwargs)

return self._obj_cls(self, server_data)

class Gitignore(RESTObject):

_id_attr = "name"

class GitignoreManager(RetrieveMixin, RESTManager):

_path = "/templates/gitignores"

_obj_cls = Gitignore

class Gitlabciyml(RESTObject):

_id_attr = "name"

class GitlabciymlManager(RetrieveMixin, RESTManager):

_path = "/templates/gitlab_ci_ymls"

_obj_cls = Gitlabciyml

class GroupAccessRequest(AccessRequestMixin, ObjectDeleteMixin, RESTObject):

pass

class GroupAccessRequestManager(ListMixin, CreateMixin, DeleteMixin, RESTManager):

_path = "/groups/%(group_id)s/access_requests"

_obj_cls = GroupAccessRequest

_from_parent_attrs = {"group_id": "id"}

class GroupBadge(SaveMixin, ObjectDeleteMixin, RESTObject):

pass

class GroupBadgeManager(BadgeRenderMixin, CRUDMixin, RESTManager):

_path = "/groups/%(group_id)s/badges"

_obj_cls = GroupBadge

_from_parent_attrs = {"group_id": "id"}

_create_attrs = (("link_url", "image_url"), tuple())

_update_attrs = (tuple(), ("link_url", "image_url"))

class GroupBoardList(SaveMixin, ObjectDeleteMixin, RESTObject):

pass

class GroupBoardListManager(CRUDMixin, RESTManager):

_path = "/groups/%(group_id)s/boards/%(board_id)s/lists"

_obj_cls = GroupBoardList

_from_parent_attrs = {"group_id": "group_id", "board_id": "id"}

_create_attrs = (("label_id",), tuple())

_update_attrs = (("position",), tuple())

class GroupBoard(SaveMixin, ObjectDeleteMixin, RESTObject):

_managers = (("lists", "GroupBoardListManager"),)

class GroupBoardManager(CRUDMixin, RESTManager):

_path = "/groups/%(group_id)s/boards"

_obj_cls = GroupBoard

_from_parent_attrs = {"group_id": "id"}

_create_attrs = (("name",), tuple())

class GroupCluster(SaveMixin, ObjectDeleteMixin, RESTObject):

pass

class GroupClusterManager(CRUDMixin, RESTManager):

_path = "/groups/%(group_id)s/clusters"

_obj_cls = GroupCluster

_from_parent_attrs = {"group_id": "id"}

_create_attrs = (

("name", "platform_kubernetes_attributes"),

("domain", "enabled", "managed", "environment_scope"),

)

_update_attrs = (

tuple(),

(

"name",

"domain",

"management_project_id",

"platform_kubernetes_attributes",

"environment_scope",

),

)

@exc.on_http_error(exc.GitlabStopError)

def create(self, data, **kwargs):

"""Create a new object.

Args:

data (dict): Parameters to send to the server to create the

resource

**kwargs: Extra options to send to the server (e.g. sudo or

'ref_name', 'stage', 'name', 'all')

Raises:

GitlabAuthenticationError: If authentication is not correct

GitlabCreateError: If the server cannot perform the request

Returns:

RESTObject: A new instance of the manage object class build with

the data sent by the server

"""

path = "%s/user" % (self.path)

return CreateMixin.create(self, data, path=path, **kwargs)

class GroupCustomAttribute(ObjectDeleteMixin, RESTObject):

_id_attr = "key"

class GroupCustomAttributeManager(RetrieveMixin, SetMixin, DeleteMixin, RESTManager):

_path = "/groups/%(group_id)s/custom_attributes"

_obj_cls = GroupCustomAttribute

_from_parent_attrs = {"group_id": "id"}

class GroupEpicIssue(ObjectDeleteMixin, SaveMixin, RESTObject):

_id_attr = "epic_issue_id"

def save(self, **kwargs):

"""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()

self.manager.update(obj_id, updated_data, **kwargs)

class GroupEpicIssueManager(

ListMixin, CreateMixin, UpdateMixin, DeleteMixin, RESTManager

):

_path = "/groups/%(group_id)s/epics/%(epic_iid)s/issues"

_obj_cls = GroupEpicIssue

_from_parent_attrs = {"group_id": "group_id", "epic_iid": "iid"}

_create_attrs = (("issue_id",), tuple())

_update_attrs = (tuple(), ("move_before_id", "move_after_id"))

@exc.on_http_error(exc.GitlabCreateError)

def create(self, data, **kwargs):

"""Create a new object.

Args:

data (dict): Parameters to send to the server to create the

resource

**kwargs: Extra options to send to the server (e.g. sudo)

Raises:

GitlabAuthenticationError: If authentication is not correct

GitlabCreateError: If the server cannot perform the request

Returns:

RESTObject: A new instance of the manage object class build with

the data sent by the server

"""

CreateMixin._check_missing_create_attrs(self, data)

path = "%s/%s" % (self.path, data.pop("issue_id"))

server_data = self.gitlab.http_post(path, **kwargs)

# The epic_issue_id attribute doesn't exist when creating the resource,

# but is used everywhere elese. Let's create it to be consistent client

# side

server_data["epic_issue_id"] = server_data["id"]

return self._obj_cls(self, server_data)

class GroupEpicResourceLabelEvent(RESTObject):

pass

class GroupEpicResourceLabelEventManager(RetrieveMixin, RESTManager):

_path = "/groups/%(group_id)s/epics/%(epic_id)s/resource_label_events"

_obj_cls = GroupEpicResourceLabelEvent

_from_parent_attrs = {"group_id": "group_id", "epic_id": "id"}

class GroupEpic(ObjectDeleteMixin, SaveMixin, RESTObject):

_id_attr = "iid"

_managers = (

("issues", "GroupEpicIssueManager"),

("resourcelabelevents", "GroupEpicResourceLabelEventManager"),

)

class GroupEpicManager(CRUDMixin, RESTManager):

_path = "/groups/%(group_id)s/epics"

_obj_cls = GroupEpic

_from_parent_attrs = {"group_id": "id"}

_list_filters = ("author_id", "labels", "order_by", "sort", "search")

_create_attrs = (("title",), ("labels", "description", "start_date", "end_date"))

_update_attrs = (

tuple(),

("title", "labels", "description", "start_date", "end_date"),

)

_types = {"labels": types.ListAttribute}

class GroupIssue(RESTObject):

pass

class GroupIssueManager(ListMixin, RESTManager):

_path = "/groups/%(group_id)s/issues"

_obj_cls = GroupIssue

_from_parent_attrs = {"group_id": "id"}

_list_filters = (

"state",

"labels",

"milestone",

"order_by",

"sort",

"iids",

"author_id",

"assignee_id",

"my_reaction_emoji",

"search",

"created_after",

"created_before",

"updated_after",

"updated_before",

)

_types = {"labels": types.ListAttribute}

class GroupLabel(SubscribableMixin, SaveMixin, ObjectDeleteMixin, RESTObject):

_id_attr = "name"

# Update without ID, but we need an ID to get from list.

@exc.on_http_error(exc.GitlabUpdateError)

def save(self, **kwargs):

"""Saves 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)

Raises:

GitlabAuthenticationError: If authentication is not correct.

GitlabUpdateError: If the server cannot perform the request.

"""

updated_data = self._get_updated_data()

# call the manager

server_data = self.manager.update(None, updated_data, **kwargs)

self._update_attrs(server_data)

class GroupLabelManager(ListMixin, CreateMixin, UpdateMixin, DeleteMixin, RESTManager):

_path = "/groups/%(group_id)s/labels"

_obj_cls = GroupLabel

_from_parent_attrs = {"group_id": "id"}

_create_attrs = (("name", "color"), ("description", "priority"))

_update_attrs = (("name",), ("new_name", "color", "description", "priority"))

# Update without ID.

def update(self, name, new_data=None, **kwargs):

"""Update a Label on the server.

Args:

name: The name of the label

**kwargs: Extra options to send to the server (e.g. sudo)

"""

new_data = new_data or {}

if name:

new_data["name"] = name

return super().update(id=None, new_data=new_data, **kwargs)

# Delete without ID.

@exc.on_http_error(exc.GitlabDeleteError)

def delete(self, name, **kwargs):

"""Delete a Label on the server.

Args:

name: The name of the label

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

"""

self.gitlab.http_delete(self.path, query_data={"name": name}, **kwargs)

class GroupMember(SaveMixin, ObjectDeleteMixin, RESTObject):

_short_print_attr = "username"

class GroupMemberManager(CRUDMixin, RESTManager):

_path = "/groups/%(group_id)s/members"

_obj_cls = GroupMember

_from_parent_attrs = {"group_id": "id"}

_create_attrs = (("access_level", "user_id"), ("expires_at",))

_update_attrs = (("access_level",), ("expires_at",))