groups.py

from typing import Any, BinaryIO, cast, Dict, List, Optional, Type, TYPE_CHECKING, Union

import requests

import gitlab
from gitlab import cli
from gitlab import exceptions as exc
from gitlab import types
from gitlab.base import RequiredOptional, RESTManager, RESTObject
from gitlab.mixins import CRUDMixin, ListMixin, ObjectDeleteMixin, SaveMixin

from .access_requests import GroupAccessRequestManager  # noqa: F401
from .audit_events import GroupAuditEventManager  # noqa: F401
from .badges import GroupBadgeManager  # noqa: F401
from .boards import GroupBoardManager  # noqa: F401
from .clusters import GroupClusterManager  # noqa: F401
from .custom_attributes import GroupCustomAttributeManager  # noqa: F401
from .deploy_tokens import GroupDeployTokenManager  # noqa: F401
from .epics import GroupEpicManager  # noqa: F401
from .export_import import GroupExportManager, GroupImportManager  # noqa: F401
from .group_access_tokens import GroupAccessTokenManager  # noqa: F401
from .hooks import GroupHookManager  # noqa: F401
from .issues import GroupIssueManager  # noqa: F401
from .labels import GroupLabelManager  # noqa: F401
from .members import (  # noqa: F401
    GroupBillableMemberManager,
    GroupMemberAllManager,
    GroupMemberManager,
)
from .merge_requests import GroupMergeRequestManager  # noqa: F401
from .milestones import GroupMilestoneManager  # noqa: F401
from .notification_settings import GroupNotificationSettingsManager  # noqa: F401
from .packages import GroupPackageManager  # noqa: F401
from .projects import GroupProjectManager  # noqa: F401
from .runners import GroupRunnerManager  # noqa: F401
from .statistics import GroupIssuesStatisticsManager  # noqa: F401
from .variables import GroupVariableManager  # noqa: F401
from .wikis import GroupWikiManager  # noqa: F401

__all__ = [
    "Group",
    "GroupManager",
    "GroupDescendantGroup",
    "GroupDescendantGroupManager",
    "GroupSubgroup",
    "GroupSubgroupManager",
]


class Group(SaveMixin, ObjectDeleteMixin, RESTObject):
    _short_print_attr = "name"

    access_tokens: GroupAccessTokenManager
    accessrequests: GroupAccessRequestManager
    audit_events: GroupAuditEventManager
    badges: GroupBadgeManager
    billable_members: GroupBillableMemberManager
    boards: GroupBoardManager
    clusters: GroupClusterManager
    customattributes: GroupCustomAttributeManager
    deploytokens: GroupDeployTokenManager
    descendant_groups: "GroupDescendantGroupManager"
    epics: GroupEpicManager
    exports: GroupExportManager
    hooks: GroupHookManager
    imports: GroupImportManager
    issues: GroupIssueManager
    issues_statistics: GroupIssuesStatisticsManager
    labels: GroupLabelManager
    members: GroupMemberManager
    members_all: GroupMemberAllManager
    mergerequests: GroupMergeRequestManager
    milestones: GroupMilestoneManager
    notificationsettings: GroupNotificationSettingsManager
    packages: GroupPackageManager
    projects: GroupProjectManager
    runners: GroupRunnerManager
    subgroups: "GroupSubgroupManager"
    variables: GroupVariableManager
    wikis: GroupWikiManager

    @cli.register_custom_action("Group", ("project_id",))
    @exc.on_http_error(exc.GitlabTransferProjectError)
    def transfer_project(self, project_id: int, **kwargs: Any) -> None:
        """Transfer a project to this group.

        Args:
            to_project_id: ID of the project to transfer
            **kwargs: Extra options to send to the server (e.g. sudo)

        Raises:
            GitlabAuthenticationError: If authentication is not correct
            GitlabTransferProjectError: If the project could not be transferred
        """
        path = f"/groups/{self.encoded_id}/projects/{project_id}"
        self.manager.gitlab.http_post(path, **kwargs)

    @cli.register_custom_action("Group", (), ("group_id",))
    @exc.on_http_error(exc.GitlabGroupTransferError)
    def transfer(self, group_id: Optional[int] = None, **kwargs: Any) -> None:
        """Transfer the group to a new parent group or make it a top-level group.

        Requires GitLab ≥14.6.

        Args:
            group_id: ID of the new parent group. When not specified,
                the group to transfer is instead turned into a top-level group.
            **kwargs: Extra options to send to the server (e.g. sudo)

        Raises:
            GitlabAuthenticationError: If authentication is not correct
            GitlabGroupTransferError: If the group could not be transferred
        """
        path = f"/groups/{self.encoded_id}/transfer"
        post_data = {}
        if group_id is not None:
            post_data["group_id"] = group_id
        self.manager.gitlab.http_post(path, post_data=post_data, **kwargs)

    @cli.register_custom_action("Group", ("scope", "search"))
    @exc.on_http_error(exc.GitlabSearchError)
    def search(
        self, scope: str, search: str, **kwargs: Any
    ) -> Union[gitlab.GitlabList, List[Dict[str, Any]]]:
        """Search the group resources matching the provided string.

        Args:
            scope: Scope of the search
            search: Search string
            **kwargs: Extra options to send to the server (e.g. sudo)

        Raises:
            GitlabAuthenticationError: If authentication is not correct
            GitlabSearchError: If the server failed to perform the request

        Returns:
            A list of dicts describing the resources found.
        """
        data = {"scope": scope, "search": search}
        path = f"/groups/{self.encoded_id}/search"
        return self.manager.gitlab.http_list(path, query_data=data, **kwargs)

    @cli.register_custom_action("Group", ("cn", "group_access", "provider"))
    @exc.on_http_error(exc.GitlabCreateError)
    def add_ldap_group_link(
        self, cn: str, group_access: int, provider: str, **kwargs: Any
    ) -> None:
        """Add an LDAP group link.

        Args:
            cn: CN of the LDAP group
            group_access: Minimum access level for members of the LDAP
                group
            provider: LDAP provider for the LDAP group
            **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
        """
        path = f"/groups/{self.encoded_id}/ldap_group_links"
        data = {"cn": cn, "group_access": group_access, "provider": provider}
        self.manager.gitlab.http_post(path, post_data=data, **kwargs)

    @cli.register_custom_action("Group", ("cn",), ("provider",))
    @exc.on_http_error(exc.GitlabDeleteError)
    def delete_ldap_group_link(
        self, cn: str, provider: Optional[str] = None, **kwargs: Any
    ) -> None:
        """Delete an LDAP group link.

        Args:
            cn: CN of the LDAP group
            provider: LDAP provider for the LDAP group
            **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
        """
        path = f"/groups/{self.encoded_id}/ldap_group_links"
        if provider is not None:
            path += f"/{provider}"
        path += f"/{cn}"
        self.manager.gitlab.http_delete(path)

    @cli.register_custom_action("Group")
    @exc.on_http_error(exc.GitlabCreateError)
    def ldap_sync(self, **kwargs: Any) -> None:
        """Sync LDAP groups.

        Args:
            **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
        """
        path = f"/groups/{self.encoded_id}/ldap_sync"
        self.manager.gitlab.http_post(path, **kwargs)

    @cli.register_custom_action("Group", ("group_id", "group_access"), ("expires_at",))
    @exc.on_http_error(exc.GitlabCreateError)
    def share(
        self,
        group_id: int,
        group_access: int,
        expires_at: Optional[str] = None,
        **kwargs: Any,
    ) -> None:
        """Share the group with a group.

        Args:
            group_id: ID of the group.
            group_access: Access level for the group.
            **kwargs: Extra options to send to the server (e.g. sudo)

        Raises:
            GitlabAuthenticationError: If authentication is not correct
            GitlabCreateError: If the server failed to perform the request

        Returns:
            Group
        """
        path = f"/groups/{self.encoded_id}/share"
        data = {
            "group_id": group_id,
            "group_access": group_access,
            "expires_at": expires_at,
        }
        server_data = self.manager.gitlab.http_post(path, post_data=data, **kwargs)
        if TYPE_CHECKING:
            assert isinstance(server_data, dict)
        self._update_attrs(server_data)

    @cli.register_custom_action("Group", ("group_id",))
    @exc.on_http_error(exc.GitlabDeleteError)
    def unshare(self, group_id: int, **kwargs: Any) -> None:
        """Delete a shared group link within a group.

        Args:
            group_id: ID of the group.
            **kwargs: Extra options to send to the server (e.g. sudo)

        Raises:
            GitlabAuthenticationError: If authentication is not correct
            GitlabDeleteError: If the server failed to perform the request
        """
        path = f"/groups/{self.encoded_id}/share/{group_id}"
        self.manager.gitlab.http_delete(path, **kwargs)


class GroupManager(CRUDMixin, RESTManager):
    _path = "/groups"
    _obj_cls = Group
    _list_filters = (
        "skip_groups",
        "all_available",
        "search",
        "order_by",
        "sort",
        "statistics",
        "owned",
        "with_custom_attributes",
        "min_access_level",
        "top_level_only",
    )
    _create_attrs = RequiredOptional(
        required=("name", "path"),
        optional=(
            "description",
            "membership_lock",
            "visibility",
            "share_with_group_lock",
            "require_two_factor_authentication",
            "two_factor_grace_period",
            "project_creation_level",
            "auto_devops_enabled",
            "subgroup_creation_level",
            "emails_disabled",
            "avatar",
            "mentions_disabled",
            "lfs_enabled",
            "request_access_enabled",
            "parent_id",
            "default_branch_protection",
            "shared_runners_minutes_limit",
            "extra_shared_runners_minutes_limit",
        ),
    )
    _update_attrs = RequiredOptional(
        optional=(
            "name",
            "path",
            "description",
            "membership_lock",
            "share_with_group_lock",
            "visibility",
            "require_two_factor_authentication",
            "two_factor_grace_period",
            "project_creation_level",
            "auto_devops_enabled",
            "subgroup_creation_level",
            "emails_disabled",
            "avatar",
            "mentions_disabled",
            "lfs_enabled",
            "request_access_enabled",
            "default_branch_protection",
            "file_template_project_id",
            "shared_runners_minutes_limit",
            "extra_shared_runners_minutes_limit",
            "prevent_forking_outside_group",
            "shared_runners_setting",
        ),
    )
    _types = {"avatar": types.ImageAttribute, "skip_groups": types.ArrayAttribute}

    def get(self, id: Union[str, int], lazy: bool = False, **kwargs: Any) -> Group:
        return cast(Group, super().get(id=id, lazy=lazy, **kwargs))

    @exc.on_http_error(exc.GitlabImportError)
    def import_group(
        self,
        file: BinaryIO,
        path: str,
        name: str,
        parent_id: Optional[str] = None,
        **kwargs: Any,
    ) -> Union[Dict[str, Any], requests.Response]:
        """Import a group from an archive file.

        Args:
            file: Data or file object containing the group
            path: The path for the new group to be imported.
            name: The name for the new group.
            parent_id: ID of a parent group that the group will
                be imported into.
            **kwargs: Extra options to send to the server (e.g. sudo)

        Raises:
            GitlabAuthenticationError: If authentication is not correct
            GitlabImportError: If the server failed to perform the request

        Returns:
            A representation of the import status.
        """
        files = {"file": ("file.tar.gz", file, "application/octet-stream")}
        data = {"path": path, "name": name}
        if parent_id is not None:
            data["parent_id"] = parent_id

        return self.gitlab.http_post(
            "/groups/import", post_data=data, files=files, **kwargs
        )


class GroupSubgroup(RESTObject):
    pass


class GroupSubgroupManager(ListMixin, RESTManager):
    _path = "/groups/{group_id}/subgroups"
    _obj_cls: Union[Type["GroupDescendantGroup"], Type[GroupSubgroup]] = GroupSubgroup
    _from_parent_attrs = {"group_id": "id"}
    _list_filters = (
        "skip_groups",
        "all_available",
        "search",
        "order_by",
        "sort",
        "statistics",
        "owned",
        "with_custom_attributes",
        "min_access_level",
    )
    _types = {"skip_groups": types.ArrayAttribute}


class GroupDescendantGroup(RESTObject):
    pass


class GroupDescendantGroupManager(GroupSubgroupManager):
    """
    This manager inherits from GroupSubgroupManager as descendant groups
    share all attributes with subgroups, except the path and object class.
    """

    _path = "/groups/{group_id}/descendant_groups"
    _obj_cls: Type[GroupDescendantGroup] = GroupDescendantGroup