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 .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" 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.id}/projects/{project_id}" self.manager.gitlab.http_post(path, **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.get_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.get_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.get_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.get_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.get_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.get_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.ListAttribute} 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.ListAttribute} 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