[#11095] feat(client-python): add Role CRUD and authorization data structures (#11210)

### What changes were proposed in this pull request?

Add Role CRUD operations and authorization data structures to the Python
client SDK:

- **DTOs**: `PrivilegeDTO`, `SecurableObjectDTO`, `RoleDTO` with JSON
serialization support
- **Role CRUD**: `create_role`, `get_role`, `delete_role`,
`list_role_names` on `GravitinoMetalake` and `GravitinoClient`
- **DTOConverters**: `to/from_privilege_dto` and
`to/from_securable_object_dto` conversion methods
- **Fix**: `GenericPrivilege.__eq__` now checks `isinstance(value,
Privilege)` for `PrivilegeDTO` compatibility
- **Tests**: Unit tests (DTO, response, client delegate, error handler)
and integration tests against a live server

### Why are the changes needed?

This is the Role authorization piece of issue #10782. User (#11058) and
Group (#11094) management are already merged. Role CRUD is a
prerequisite for Grant/Revoke operations, which will follow in a
separate PR.

Fix: #11095

### Does this PR introduce _any_ user-facing change?

Yes — new public APIs on `GravitinoClient` and `GravitinoMetalake`:
- `create_role(role_name, properties=None, securable_objects=None)`
- `get_role(role_name)`
- `delete_role(role_name) -> bool`
- `list_role_names() -> list[str]`

### How was this patch tested?

- Unit tests: all passed
- Integration tests: 4 passed against a live Gravitino server with
authorization enabled
- Linting: `ruff check` clean, `pylint` 10/10

---------

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

Author: sunyuhan1998

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

@@ -265,9 +265,9 @@ def __hash__(self) -> int:
         return hash((self._condition, self._name))
 
     def __eq__(self, value: object) -> bool:
-        if not isinstance(value, GenericPrivilege):
+        if not isinstance(value, Privilege):
             return False
-        return self._condition == value._condition and self._name == value._name
+        return self._condition == value.condition() and self._name == value.name()
 
 
 class CreateCatalog(GenericPrivilege):

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

@@ -17,6 +17,8 @@
 
 from __future__ import annotations
 
+from gravitino.api.authorization.privileges import Privilege
+from gravitino.api.authorization.securable_objects import SecurableObject
 from gravitino.api.catalog import Catalog
 from gravitino.api.catalog_change import CatalogChange
 from gravitino.api.job.job_template import JobTemplate, JobType
@@ -36,6 +38,8 @@
 from gravitino.client.fileset_catalog import FilesetCatalog
 from gravitino.client.generic_model_catalog import GenericModelCatalog
 from gravitino.client.relational_catalog import RelationalCatalog
+from gravitino.dto.authorization.privilege_dto import PrivilegeDTO
+from gravitino.dto.authorization.securable_object_dto import SecurableObjectDTO
 from gravitino.dto.catalog_dto import CatalogDTO
 from gravitino.dto.job.job_template_dto import JobTemplateDTO
 from gravitino.dto.job.shell_job_template_dto import ShellJobTemplateDTO
@@ -325,3 +329,12 @@ def to_tag_update_request(
             return TagUpdateRequest.RemoveTagPropertyRequest(change.removed_property)
 
         raise IllegalArgumentException(f"Unknown change type: {type(change)}")
+
+    @staticmethod
+    def to_privilege_dto(privilege: Privilege) -> PrivilegeDTO:
+        return PrivilegeDTO(privilege.name(), privilege.condition())
+
+    @staticmethod
+    def to_securable_object_dto(obj: SecurableObject) -> SecurableObjectDTO:
+        privilege_dtos = [DTOConverters.to_privilege_dto(p) for p in obj.privileges()]
+        return SecurableObjectDTO(obj.full_name(), obj.type(), privilege_dtos)

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

@@ -21,6 +21,8 @@
 
 from gravitino.api.authorization.group import Group
 from gravitino.api.authorization.owner import Owner
+from gravitino.api.authorization.role import Role
+from gravitino.api.authorization.securable_objects import SecurableObject
 from gravitino.api.authorization.user import User
 from gravitino.api.catalog import Catalog
 from gravitino.api.catalog_change import CatalogChange
@@ -507,3 +509,67 @@ def list_group_names(self) -> list[str]:
             NoSuchMetalakeException: If the metalake does not exist.
         """
         return self.get_metalake().list_group_names()
+
+    # Role operations
+
+    def create_role(
+        self,
+        role_name: str,
+        properties: Optional[Dict[str, str]] = None,
+        securable_objects: Optional[List[SecurableObject]] = None,
+    ) -> Role:
+        """Create a new role under the metalake.
+
+        Args:
+            role_name: The name of the role.
+            properties: The properties of the role.
+            securable_objects: The securable objects of the role.
+
+        Returns:
+            The created Role object.
+
+        Raises:
+            RoleAlreadyExistsException: If a role with the same name already exists.
+            NoSuchMetalakeException: If the metalake does not exist.
+        """
+        return self.get_metalake().create_role(role_name, properties, securable_objects)
+
+    def get_role(self, role_name: str) -> Role:
+        """Get a role by name from the metalake.
+
+        Args:
+            role_name: The name of the role.
+
+        Returns:
+            The Role object.
+
+        Raises:
+            NoSuchRoleException: If the role does not exist.
+            NoSuchMetalakeException: If the metalake does not exist.
+        """
+        return self.get_metalake().get_role(role_name)
+
+    def delete_role(self, role_name: str) -> bool:
+        """Delete a role from the metalake.
+
+        Args:
+            role_name: The name of the role.
+
+        Returns:
+            True if the role was deleted, False otherwise.
+
+        Raises:
+            NoSuchMetalakeException: If the metalake does not exist.
+        """
+        return self.get_metalake().delete_role(role_name)
+
+    def list_role_names(self) -> list[str]:
+        """List all role names under the metalake.
+
+        Returns:
+            A list of role name strings.
+
+        Raises:
+            NoSuchMetalakeException: If the metalake does not exist.
+        """
+        return self.get_metalake().list_role_names()

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

@@ -15,11 +15,14 @@
 # specific language governing permissions and limitations
 # under the License.
 
+# pylint: disable=too-many-lines
 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.role import Role
+from gravitino.api.authorization.securable_objects import SecurableObject
 from gravitino.api.authorization.user import User
 from gravitino.api.catalog import Catalog
 from gravitino.api.catalog_change import CatalogChange
@@ -49,6 +52,7 @@
 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.requests.role_create_request import RoleCreateRequest
 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
@@ -75,8 +79,13 @@
     GroupNamesListResponse,
     GroupResponse,
 )
+from gravitino.dto.responses.role_response import (
+    RoleNamesListResponse,
+    RoleResponse,
+)
 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.role_error_handler import ROLE_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
@@ -114,6 +123,8 @@ class GravitinoMetalake(
     API_METALAKES_USER_PATH = "api/metalakes/{}/users/{}"
     API_METALAKES_GROUPS_PATH = "api/metalakes/{}/groups"
     API_METALAKES_GROUP_PATH = "api/metalakes/{}/groups/{}"
+    API_METALAKES_ROLES_PATH = "api/metalakes/{}/roles"
+    API_METALAKES_ROLE_PATH = "api/metalakes/{}/roles/{}"
 
     def __init__(self, metalake: MetalakeDTO = None, client: HTTPClient = None):
         super().__init__(
@@ -1002,3 +1013,106 @@ def list_group_names(self) -> list[str]:
         resp = GroupNamesListResponse.from_json(response.body, infer_missing=True)
         resp.validate()
         return resp.names()
+
+    #####################
+    # Role operations
+    #####################
+
+    def create_role(
+        self,
+        role_name: str,
+        properties: Optional[Dict[str, str]] = None,
+        securable_objects: Optional[List[SecurableObject]] = None,
+    ) -> Role:
+        """Create a new role under this metalake.
+
+        Args:
+            role_name: The name of the role.
+            properties: The properties of the role.
+            securable_objects: The securable objects of the role.
+
+        Returns:
+            The created Role object.
+
+        Raises:
+            RoleAlreadyExistsException: If a role with the same name already exists.
+            NoSuchMetalakeException: If the metalake does not exist.
+        """
+        Precondition.check_string_not_empty(
+            role_name, "role name must not be null or empty"
+        )
+        securable_object_dtos = [
+            DTOConverters.to_securable_object_dto(obj)
+            for obj in (securable_objects or [])
+        ]
+        req = RoleCreateRequest(role_name, properties, securable_object_dtos)
+        req.validate()
+        url = self.API_METALAKES_ROLES_PATH.format(encode_string(self.name()))
+        response = self.rest_client.post(
+            url, json=req, error_handler=ROLE_ERROR_HANDLER
+        )
+        resp = RoleResponse.from_json(response.body, infer_missing=True)
+        resp.validate()
+        return resp.role()
+
+    def get_role(self, role_name: str) -> Role:
+        """Get a role by name from this metalake.
+
+        Args:
+            role_name: The name of the role.
+
+        Returns:
+            The Role object.
+
+        Raises:
+            NoSuchRoleException: If the role does not exist.
+            NoSuchMetalakeException: If the metalake does not exist.
+        """
+        Precondition.check_string_not_empty(
+            role_name, "role name must not be null or empty"
+        )
+        url = self.API_METALAKES_ROLE_PATH.format(
+            encode_string(self.name()), encode_string(role_name)
+        )
+        response = self.rest_client.get(url, error_handler=ROLE_ERROR_HANDLER)
+        resp = RoleResponse.from_json(response.body, infer_missing=True)
+        resp.validate()
+        return resp.role()
+
+    def delete_role(self, role_name: str) -> bool:
+        """Delete a role from this metalake.
+
+        Args:
+            role_name: The name of the role.
+
+        Returns:
+            True if the role was deleted, False otherwise.
+
+        Raises:
+            NoSuchMetalakeException: If the metalake does not exist.
+        """
+        Precondition.check_string_not_empty(
+            role_name, "role name must not be null or empty"
+        )
+        url = self.API_METALAKES_ROLE_PATH.format(
+            encode_string(self.name()), encode_string(role_name)
+        )
+        response = self.rest_client.delete(url, error_handler=ROLE_ERROR_HANDLER)
+        drop_response = DropResponse.from_json(response.body, infer_missing=True)
+        drop_response.validate()
+        return drop_response.dropped()
+
+    def list_role_names(self) -> list[str]:
+        """List all role names under this metalake.
+
+        Returns:
+            A list of role name strings.
+
+        Raises:
+            NoSuchMetalakeException: If the metalake does not exist.
+        """
+        url = self.API_METALAKES_ROLES_PATH.format(encode_string(self.name()))
+        response = self.rest_client.get(url, error_handler=ROLE_ERROR_HANDLER)
+        resp = RoleNamesListResponse.from_json(response.body, infer_missing=True)
+        resp.validate()
+        return resp.names()

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

@@ -16,3 +16,8 @@
 # under the License.
 
 from gravitino.dto.authorization.owner_dto import OwnerDTO
+from gravitino.dto.authorization.privilege_dto import PrivilegeDTO
+from gravitino.dto.authorization.role_dto import RoleDTO
+from gravitino.dto.authorization.securable_object_dto import SecurableObjectDTO
+
+__all__ = ["OwnerDTO", "PrivilegeDTO", "RoleDTO", "SecurableObjectDTO"]

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

@@ -0,0 +1,112 @@
+# 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 dataclasses_json import config, dataclass_json
+
+from gravitino.api.authorization.privileges import Privilege, Privileges
+from gravitino.api.metadata_object import MetadataObject
+
+
+def _encode_privilege_name(name: Privilege.Name) -> str:
+    return name.name.lower()
+
+
+def _decode_privilege_name(val: str) -> Privilege.Name:
+    return Privilege.Name[val.upper()]
+
+
+def _encode_condition(condition: Privilege.Condition) -> str:
+    return condition.value.lower()
+
+
+def _decode_condition(val: str) -> Privilege.Condition:
+    return Privilege.Condition(val.upper())
+
+
+@dataclass_json
+@dataclass
+class PrivilegeDTO(Privilege):
+    """Data transfer object representing a privilege."""
+
+    _name: Privilege.Name = field(
+        metadata=config(
+            field_name="name",
+            encoder=_encode_privilege_name,
+            decoder=_decode_privilege_name,
+        )
+    )
+    _condition: Privilege.Condition = field(
+        metadata=config(
+            field_name="condition",
+            encoder=_encode_condition,
+            decoder=_decode_condition,
+        )
+    )
+
+    def name(self) -> Privilege.Name:
+        return self._name
+
+    def simple_string(self) -> str:
+        return f"{self._condition.value} {self._name.name.lower().replace('_', ' ')}"
+
+    def condition(self) -> Privilege.Condition:
+        return self._condition
+
+    def can_bind_to(self, obj_type: MetadataObject.Type) -> bool:
+        if self._condition == Privilege.Condition.ALLOW:
+            return Privileges.allow(self._name.name).can_bind_to(obj_type)
+        return Privileges.deny(self._name.name).can_bind_to(obj_type)
+
+    def __eq__(self, other: object) -> bool:
+        if not isinstance(other, Privilege):
+            return False
+        return self._name == other.name() and self._condition == other.condition()
+
+    def __hash__(self) -> int:
+        return hash((self._name, self._condition))
+
+    @staticmethod
+    def builder() -> PrivilegeDTO.Builder:
+        return PrivilegeDTO.Builder()
+
+    class Builder:
+        """Helper class to build a PrivilegeDTO object."""
+
+        def __init__(self) -> None:
+            self._name: Privilege.Name = None
+            self._condition: Privilege.Condition = None
+
+        def with_name(self, name: Privilege.Name) -> PrivilegeDTO.Builder:
+            self._name = name
+            return self
+
+        def with_condition(
+            self, condition: Privilege.Condition
+        ) -> PrivilegeDTO.Builder:
+            self._condition = condition
+            return self
+
+        def build(self) -> PrivilegeDTO:
+            if self._name is None:
+                raise ValueError("name cannot be None")
+            if self._condition is None:
+                raise ValueError("condition cannot be None")
+            return PrivilegeDTO(self._name, self._condition)