[#11094] feat(client-python): add Group authorization management (#11109)

## Summary
This PR adds Group authorization management support to the Gravitino
Python client SDK, following the pattern established in #11058 (User
management). It is split from the original monolithic PR #10783.

### Changes
- **API**: `Group` interface with `name()` and `roles()` methods
- **DTO**: `GroupDTO` with immutable roles (`tuple[str, ...]`
internally, defensive `list` copy via `roles()`); audit null validation
in Builder
- **Request/Response**: `GroupAddRequest`, `GroupResponse`,
`GroupListResponse`, `GroupNamesListResponse`
- **Error Handler**: `GroupErrorHandler` mapping REST error codes to
`NoSuchGroupException`, `GroupAlreadyExistsException`, etc.
- **Client**: 5 CRUD methods on both `GravitinoMetalake` and
`GravitinoClient` (`add_group`, `remove_group`, `get_group`,
`list_groups`, `list_group_names`)
- **Tests**: Unit tests (33 cases) and integration tests (6 cases)

### Sub-task of
- #10782 (Authorization management for Python SDK)

## Related issues
- Closes #11094

## Reviewers
@jerryshao

---------

Co-authored-by: Sun Yuhan <sunyuhan1998@users.noreply.github.com>

Author: sunyuhan1998

clients/client-python/gravitino/api/authorization/__init__.py

@@ -17,12 +17,14 @@
 
 from __future__ import annotations
 
+from gravitino.api.authorization.group import Group
 from gravitino.api.authorization.privileges import Privileges
 from gravitino.api.authorization.role import Role
 from gravitino.api.authorization.securable_objects import SecurableObjects
 from gravitino.api.authorization.user import User
 
 __all__ = [
+    "Group",
     "Role",
     "SecurableObjects",
     "Privileges",

clients/client-python/gravitino/api/authorization/group.py

@@ -0,0 +1,47 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from __future__ import annotations
+
+from abc import abstractmethod
+
+from gravitino.api.auditable import Auditable
+
+
+class Group(Auditable):
+    """The interface of a group. The group is a collection of users in the authorization system."""
+
+    @abstractmethod
+    def name(self) -> str:
+        """
+        The name of the group.
+
+        Returns:
+            str: The name of the group.
+        """
+        raise NotImplementedError()
+
+    @abstractmethod
+    def roles(self) -> list[str]:
+        """
+        The roles of the group. A group can have multiple roles.
+        Every role binds several privileges.
+
+        Returns:
+            list[str]: The role names of the group.
+        """
+        raise NotImplementedError()

clients/client-python/gravitino/client/gravitino_client.py

@@ -19,6 +19,7 @@
 
 from typing import Dict, List, Optional
 
+from gravitino.api.authorization.group import Group
 from gravitino.api.authorization.owner import Owner
 from gravitino.api.authorization.user import User
 from gravitino.api.catalog import Catalog
@@ -438,3 +439,71 @@ def list_user_names(self) -> list[str]:
             NoSuchMetalakeException: If the metalake does not exist.
         """
         return self.get_metalake().list_user_names()
+
+    # Group operations
+
+    def add_group(self, group: str) -> Group:
+        """Add a group to the metalake.
+
+        Args:
+            group: The name of the group.
+
+        Returns:
+            The added Group object.
+
+        Raises:
+            GroupAlreadyExistsException: If a group with the same name already exists.
+            NoSuchMetalakeException: If the metalake does not exist.
+        """
+        return self.get_metalake().add_group(group)
+
+    def remove_group(self, group: str) -> bool:
+        """Remove a group from the metalake.
+
+        Args:
+            group: The name of the group.
+
+        Returns:
+            True if the group was removed, False if the group did not exist.
+
+        Raises:
+            NoSuchMetalakeException: If the metalake does not exist.
+        """
+        return self.get_metalake().remove_group(group)
+
+    def get_group(self, group: str) -> Group:
+        """Get a group by name from the metalake.
+
+        Args:
+            group: The name of the group.
+
+        Returns:
+            The Group object.
+
+        Raises:
+            NoSuchGroupException: If the group does not exist.
+            NoSuchMetalakeException: If the metalake does not exist.
+        """
+        return self.get_metalake().get_group(group)
+
+    def list_groups(self) -> list[Group]:
+        """List all groups with details under the metalake.
+
+        Returns:
+            A list of Group objects.
+
+        Raises:
+            NoSuchMetalakeException: If the metalake does not exist.
+        """
+        return self.get_metalake().list_groups()
+
+    def list_group_names(self) -> list[str]:
+        """List all group names under the metalake.
+
+        Returns:
+            A list of group name strings.
+
+        Raises:
+            NoSuchMetalakeException: If the metalake does not exist.
+        """
+        return self.get_metalake().list_group_names()

clients/client-python/gravitino/client/gravitino_metalake.py

@@ -18,6 +18,7 @@
 import logging
 from typing import Dict, List, Optional
 
+from gravitino.api.authorization.group import Group
 from gravitino.api.authorization.owner import Owner
 from gravitino.api.authorization.user import User
 from gravitino.api.catalog import Catalog
@@ -47,6 +48,7 @@
 from gravitino.dto.requests.tag_create_request import TagCreateRequest
 from gravitino.dto.requests.tag_updates_request import TagUpdatesRequest
 from gravitino.dto.requests.user_add_request import UserAddRequest
+from gravitino.dto.requests.group_add_request import GroupAddRequest
 from gravitino.dto.responses.catalog_list_response import CatalogListResponse
 from gravitino.dto.responses.catalog_response import CatalogResponse
 from gravitino.dto.responses.drop_response import DropResponse
@@ -68,7 +70,13 @@
     UserNamesListResponse,
     UserResponse,
 )
+from gravitino.dto.responses.group_response import (
+    GroupListResponse,
+    GroupNamesListResponse,
+    GroupResponse,
+)
 from gravitino.exceptions.handlers.catalog_error_handler import CATALOG_ERROR_HANDLER
+from gravitino.exceptions.handlers.group_error_handler import GROUP_ERROR_HANDLER
 from gravitino.exceptions.handlers.job_error_handler import JOB_ERROR_HANDLER
 from gravitino.exceptions.handlers.owner_error_handler import OWNER_ERROR_HANDLER
 from gravitino.exceptions.handlers.tag_error_handler import TAG_ERROR_HANDLER
@@ -104,6 +112,8 @@ class GravitinoMetalake(
     # Authorization paths
     API_METALAKES_USERS_PATH = "api/metalakes/{}/users"
     API_METALAKES_USER_PATH = "api/metalakes/{}/users/{}"
+    API_METALAKES_GROUPS_PATH = "api/metalakes/{}/groups"
+    API_METALAKES_GROUP_PATH = "api/metalakes/{}/groups/{}"
 
     def __init__(self, metalake: MetalakeDTO = None, client: HTTPClient = None):
         super().__init__(
@@ -883,3 +893,112 @@ def list_user_names(self) -> list[str]:
         resp = UserNamesListResponse.from_json(response.body, infer_missing=True)
         resp.validate()
         return resp.names()
+
+    #####################
+    # Group operations
+    #####################
+
+    def add_group(self, group: str) -> Group:
+        """Add a group to this metalake.
+
+        Args:
+            group: The name of the group.
+
+        Returns:
+            The added Group object.
+
+        Raises:
+            GroupAlreadyExistsException: If a group with the same name already exists.
+            NoSuchMetalakeException: If the metalake does not exist.
+        """
+        Precondition.check_string_not_empty(
+            group, "group name must not be null or empty"
+        )
+        req = GroupAddRequest(group)
+        req.validate()
+        url = self.API_METALAKES_GROUPS_PATH.format(encode_string(self.name()))
+        response = self.rest_client.post(
+            url, json=req, error_handler=GROUP_ERROR_HANDLER
+        )
+        resp = GroupResponse.from_json(response.body, infer_missing=True)
+        resp.validate()
+        return resp.group()
+
+    def remove_group(self, group: str) -> bool:
+        """Remove a group from this metalake.
+
+        Args:
+            group: The name of the group.
+
+        Returns:
+            True if the group was removed, False if the group did not exist.
+
+        Raises:
+            NoSuchMetalakeException: If the metalake does not exist.
+        """
+        Precondition.check_string_not_empty(
+            group, "group name must not be null or empty"
+        )
+        url = self.API_METALAKES_GROUP_PATH.format(
+            encode_string(self.name()), encode_string(group)
+        )
+        response = self.rest_client.delete(url, error_handler=GROUP_ERROR_HANDLER)
+        remove_response = RemoveResponse.from_json(response.body, infer_missing=True)
+        remove_response.validate()
+        return remove_response.removed()
+
+    def get_group(self, group: str) -> Group:
+        """Get a group by name from this metalake.
+
+        Args:
+            group: The name of the group.
+
+        Returns:
+            The Group object.
+
+        Raises:
+            NoSuchGroupException: If the group does not exist.
+            NoSuchMetalakeException: If the metalake does not exist.
+        """
+        Precondition.check_string_not_empty(
+            group, "group name must not be null or empty"
+        )
+        url = self.API_METALAKES_GROUP_PATH.format(
+            encode_string(self.name()), encode_string(group)
+        )
+        response = self.rest_client.get(url, error_handler=GROUP_ERROR_HANDLER)
+        resp = GroupResponse.from_json(response.body, infer_missing=True)
+        resp.validate()
+        return resp.group()
+
+    def list_groups(self) -> list[Group]:
+        """List all groups with details under this metalake.
+
+        Returns:
+            A list of Group objects.
+
+        Raises:
+            NoSuchMetalakeException: If the metalake does not exist.
+        """
+        url = self.API_METALAKES_GROUPS_PATH.format(encode_string(self.name()))
+        response = self.rest_client.get(
+            url, params={"details": "true"}, error_handler=GROUP_ERROR_HANDLER
+        )
+        resp = GroupListResponse.from_json(response.body, infer_missing=True)
+        resp.validate()
+        return resp.groups()
+
+    def list_group_names(self) -> list[str]:
+        """List all group names under this metalake.
+
+        Returns:
+            A list of group name strings.
+
+        Raises:
+            NoSuchMetalakeException: If the metalake does not exist.
+        """
+        url = self.API_METALAKES_GROUPS_PATH.format(encode_string(self.name()))
+        response = self.rest_client.get(url, error_handler=GROUP_ERROR_HANDLER)
+        resp = GroupNamesListResponse.from_json(response.body, infer_missing=True)
+        resp.validate()
+        return resp.names()

clients/client-python/gravitino/dto/authorization/group_dto.py

@@ -0,0 +1,91 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Optional
+
+from dataclasses_json import config, dataclass_json
+
+from gravitino.api.authorization.group import Group
+from gravitino.dto.audit_dto import AuditDTO
+
+
+@dataclass_json
+@dataclass
+class GroupDTO(Group):
+    """Represents a Group Data Transfer Object (DTO)."""
+
+    _name: str = field(metadata=config(field_name="name"))
+    _roles: tuple[str, ...] = field(
+        default_factory=tuple, metadata=config(field_name="roles")
+    )
+    _audit: Optional[AuditDTO] = field(
+        default=None, metadata=config(field_name="audit")
+    )
+
+    def __eq__(self, other: object) -> bool:
+        if not isinstance(other, GroupDTO):
+            return False
+        return (
+            self._name == other._name
+            and self._roles == other._roles
+            and self._audit == other._audit
+        )
+
+    def __hash__(self) -> int:
+        return hash((self._name, tuple(self._roles), self._audit))
+
+    @staticmethod
+    def builder() -> GroupDTO.Builder:
+        return GroupDTO.Builder()
+
+    def name(self) -> str:
+        return self._name
+
+    def roles(self) -> list[str]:
+        return list(self._roles) if self._roles else []
+
+    def audit_info(self) -> Optional[AuditDTO]:
+        return self._audit
+
+    class Builder:
+        """Helper class to build a GroupDTO object."""
+
+        def __init__(self) -> None:
+            self._name: str = ""
+            self._roles: tuple[str, ...] = ()
+            self._audit: Optional[AuditDTO] = None
+
+        def with_name(self, name: str) -> GroupDTO.Builder:
+            self._name = name
+            return self
+
+        def with_roles(self, roles: list[str]) -> GroupDTO.Builder:
+            if roles is not None:
+                self._roles = tuple(roles)
+            return self
+
+        def with_audit(self, audit: AuditDTO) -> GroupDTO.Builder:
+            self._audit = audit
+            return self
+
+        def build(self) -> GroupDTO:
+            if not self._name:
+                raise ValueError("name cannot be null or empty")
+            return GroupDTO(self._name, self._roles, self._audit)