# -*- 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 EventManager(ListMixin, RESTManager):

_path = "/events"

_obj_cls = Event

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

class UserActivities(RESTObject):

_id_attr = "username"

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

"""

path = "/users/%s/projects" % self._parent.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"),

)

@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

class UserManager(CRUDMixin, RESTManager):

_path = "/users"

_obj_cls = User

_list_filters = (

"active",

"blocked",

"username",

"extern_uid",

"provider",

"external",

"search",

"custom_attributes",

)

_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 CurrentUser(RESTObject):

_id_attr = None

_short_print_attr = "username"

_managers = (

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

(

"admin_notification_email",

"after_sign_out_path",

"after_sign_up_text",

"akismet_api_key",

"akismet_enabled",

"circuitbreaker_access_retries",

"circuitbreaker_check_interval",

"circuitbreaker_failure_count_threshold",

"circuitbreaker_failure_reset_time",

"circuitbreaker_storage_timeout",

"clientside_sentry_dsn",

"clientside_sentry_enabled",

"container_registry_token_expire_delay",

"default_artifacts_expire_in",

"default_branch_protection",

"default_group_visibility",

"default_project_visibility",

"default_projects_limit",

"default_snippet_visibility",

"disabled_oauth_sign_in_sources",

"domain_blacklist_enabled",

"domain_blacklist",

"domain_whitelist",

"dsa_key_restriction",

"ecdsa_key_restriction",

"ed25519_key_restriction",

"email_author_in_body",

"enabled_git_access_protocol",

"gravatar_enabled",

"help_page_hide_commercial_content",

"help_page_support_url",

"home_page_url",

"housekeeping_bitmaps_enabled",

"housekeeping_enabled",

"housekeeping_full_repack_period",

"housekeeping_gc_period",

"housekeeping_incremental_repack_period",

"html_emails_enabled",

"import_sources",

"koding_enabled",

"koding_url",

"max_artifacts_size",

"max_attachment_size",

"max_pages_size",

"metrics_enabled",

"metrics_host",

"metrics_method_call_threshold",

"metrics_packet_size",

"metrics_pool_size",

"metrics_port",

"metrics_sample_interval",

"metrics_timeout",

"password_authentication_enabled_for_web",

"password_authentication_enabled_for_git",

"performance_bar_allowed_group_id",

"performance_bar_enabled",

"plantuml_enabled",

"plantuml_url",

"polling_interval_multiplier",

"project_export_enabled",

"prometheus_metrics_enabled",

"recaptcha_enabled",

"recaptcha_private_key",

"recaptcha_site_key",

"repository_checks_enabled",

"repository_storages",

"require_two_factor_authentication",

"restricted_visibility_levels",

"rsa_key_restriction",

"send_user_confirmation_email",

"sentry_dsn",

"sentry_enabled",

"session_expire_delay",

"shared_runners_enabled",

"shared_runners_text",

"sidekiq_throttling_enabled",

"sidekiq_throttling_factor",

"sidekiq_throttling_queues",

"sign_in_text",

"signup_enabled",

"terminal_max_session_time",

"two_factor_grace_period",

"unique_ips_limit_enabled",

"unique_ips_limit_per_user",

"unique_ips_limit_time_window",

"usage_ping_enabled",

"user_default_external",

"user_oauth_applications",

"version_check_enabled",

"enforce_terms",

"terms",

),

)

@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 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 {}

new_data["name"] = name

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

@cli.register_custom_action("GroupMemberManager")

@exc.on_http_error(exc.GitlabListError)

def all(self, **kwargs):

"""List all the members, included inherited ones.

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)

Raises:

GitlabAuthenticationError: If authentication is not correct

GitlabListError: If the list could not be retrieved

Returns:

RESTObjectList: The list of members

"""

path = "%s/all" % self.path

obj = self.gitlab.http_list(path, **kwargs)

return [self._obj_cls(self, item) for item in obj]

class GroupMergeRequest(RESTObject):

pass

class GroupMergeRequestManager(ListMixin, RESTManager):

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

_obj_cls = GroupMergeRequest

_from_parent_attrs = {"group_id": "id"}

_list_filters = (

"state",

"order_by",

"sort",

"milestone",

"view",

"labels",

"created_after",

"created_before",

"updated_after",

"updated_before",

"scope",

"author_id",

"assignee_id",

"my_reaction_emoji",

"source_branch",

"target_branch",

"search",

)

_types = {"labels": types.ListAttribute}

class GroupMilestone(SaveMixin, ObjectDeleteMixin, RESTObject):

_short_print_attr = "title"

@cli.register_custom_action("GroupMilestone")

@exc.on_http_error(exc.GitlabListError)

def issues(self, **kwargs):

"""List issues related to this milestone.

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)

Raises:

GitlabAuthenticationError: If authentication is not correct

GitlabListError: If the list could not be retrieved

Returns:

RESTObjectList: The list of issues

"""

path = "%s/%s/issues" % (self.manager.path, self.get_id())

data_list = self.manager.gitlab.http_list(path, as_list=False, **kwargs)

manager = GroupIssueManager(self.manager.gitlab, parent=self.manager._parent)

# FIXME(gpocentek): the computed manager path is not correct

return RESTObjectList(manager, GroupIssue, data_list)

@cli.register_custom_action("GroupMilestone")

@exc.on_http_error(exc.GitlabListError)

def merge_requests(self, **kwargs):

"""List the merge requests related to this milestone.

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)