[#10782] feat(client-python): add User authorization management

Add User management capabilities to the Python client, supporting the
same set of User operations already exposed by the Java client and REST
API layer.

This is part of a larger split of #10783, which introduced the full
authorization model in a single PR. To make review easier, the change
is being broken down by entity; subsequent PRs will add Group, Role,
Privilege, SecurableObject and grant/revoke operations.

Changes
- Add User interface and UserDTO (Builder-based, immutable)
- Add UserAddRequest / UserResponse / UserListResponse /
  UserNamesListResponse DTOs, plus a shared RemoveResponse
- Add add_user / get_user / remove_user / list_users /
  list_user_names methods to GravitinoMetalake
- Add UserErrorHandler to map User-related REST errors to
  NoSuchUserException, UserAlreadyExistsException, etc.
- Add 28 unit tests covering DTO serialization, response parsing,
  GravitinoMetalake HTTP interactions (URL, method, body, error
  propagation) and the error-handler mapping table

Author: sunyuhan1998

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

@@ -20,9 +20,11 @@
 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__ = [
     "Role",
     "SecurableObjects",
     "Privileges",
+    "User",
 ]

clients/client-python/gravitino/api/authorization/user.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 User(Auditable):
+    """The interface of a user. The user is a basic entity in the authorization system."""
+
+    @abstractmethod
+    def name(self) -> str:
+        """
+        The name of the user.
+
+        Returns:
+            str: The name of the user.
+        """
+        raise NotImplementedError()
+
+    @abstractmethod
+    def roles(self) -> list[str]:
+        """
+        The roles of the user. A user can have multiple roles.
+        Every role binds several privileges.
+
+        Returns:
+            list[str]: The role names of the user.
+        """
+        raise NotImplementedError()

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

@@ -19,6 +19,7 @@
 from typing import Dict, List, Optional
 
 from gravitino.api.authorization.owner import Owner
+from gravitino.api.authorization.user import User
 from gravitino.api.catalog import Catalog
 from gravitino.api.catalog_change import CatalogChange
 from gravitino.api.job.job_handle import JobHandle
@@ -45,6 +46,7 @@
 from gravitino.dto.requests.owner_set_request import OwnerSetRequest
 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.responses.catalog_list_response import CatalogListResponse
 from gravitino.dto.responses.catalog_response import CatalogResponse
 from gravitino.dto.responses.drop_response import DropResponse
@@ -54,16 +56,23 @@
 from gravitino.dto.responses.job_template_list_response import JobTemplateListResponse
 from gravitino.dto.responses.job_template_response import JobTemplateResponse
 from gravitino.dto.responses.owner_response import OwnerResponse
+from gravitino.dto.responses.remove_response import RemoveResponse
 from gravitino.dto.responses.set_response import SetResponse
 from gravitino.dto.responses.tag_response import (
     TagListResponse,
     TagNamesListResponse,
     TagResponse,
 )
+from gravitino.dto.responses.user_response import (
+    UserListResponse,
+    UserNamesListResponse,
+    UserResponse,
+)
 from gravitino.exceptions.handlers.catalog_error_handler import CATALOG_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
+from gravitino.exceptions.handlers.user_error_handler import USER_ERROR_HANDLER
 from gravitino.rest.rest_utils import encode_string
 from gravitino.utils.http_client import HTTPClient
 from gravitino.utils.precondition import Precondition
@@ -92,6 +101,10 @@ class GravitinoMetalake(
     API_METALAKES_TAG_PATH = "api/metalakes/{}/tags/{}"
     API_METALAKES_TAGS_PATH = "api/metalakes/{}/tags"
 
+    # Authorization paths
+    API_METALAKES_USERS_PATH = "api/metalakes/{}/users"
+    API_METALAKES_USER_PATH = "api/metalakes/{}/users/{}"
+
     def __init__(self, metalake: MetalakeDTO = None, client: HTTPClient = None):
         super().__init__(
             _name=metalake.name(),
@@ -767,3 +780,60 @@ def set_owner(
         )
         set_resp = SetResponse.from_json(response.body, infer_missing=True)
         set_resp.validate()
+
+    ####################
+    # User operations
+    ####################
+
+    def add_user(self, user: str) -> User:
+        """Add a user to this metalake."""
+        Precondition.check_string_not_empty(user, "user name must not be null or empty")
+        req = UserAddRequest(user)
+        req.validate()
+        url = self.API_METALAKES_USERS_PATH.format(encode_string(self.name()))
+        response = self.rest_client.post(
+            url, json=req, error_handler=USER_ERROR_HANDLER
+        )
+        resp = UserResponse.from_json(response.body, infer_missing=True)
+        resp.validate()
+        return resp.user()
+
+    def remove_user(self, user: str) -> bool:
+        """Remove a user from this metalake."""
+        Precondition.check_string_not_empty(user, "user name must not be null or empty")
+        url = self.API_METALAKES_USER_PATH.format(
+            encode_string(self.name()), encode_string(user)
+        )
+        response = self.rest_client.delete(url, error_handler=USER_ERROR_HANDLER)
+        remove_response = RemoveResponse.from_json(response.body, infer_missing=True)
+        remove_response.validate()
+        return remove_response.removed()
+
+    def get_user(self, user: str) -> User:
+        """Get a user by name from this metalake."""
+        Precondition.check_string_not_empty(user, "user name must not be null or empty")
+        url = self.API_METALAKES_USER_PATH.format(
+            encode_string(self.name()), encode_string(user)
+        )
+        response = self.rest_client.get(url, error_handler=USER_ERROR_HANDLER)
+        resp = UserResponse.from_json(response.body, infer_missing=True)
+        resp.validate()
+        return resp.user()
+
+    def list_users(self) -> list[User]:
+        """List all users with details under this metalake."""
+        url = self.API_METALAKES_USERS_PATH.format(encode_string(self.name()))
+        response = self.rest_client.get(
+            url, params={"details": "true"}, error_handler=USER_ERROR_HANDLER
+        )
+        resp = UserListResponse.from_json(response.body, infer_missing=True)
+        resp.validate()
+        return resp.users()
+
+    def list_user_names(self) -> list[str]:
+        """List all user names under this metalake."""
+        url = self.API_METALAKES_USERS_PATH.format(encode_string(self.name()))
+        response = self.rest_client.get(url, error_handler=USER_ERROR_HANDLER)
+        resp = UserNamesListResponse.from_json(response.body, infer_missing=True)
+        resp.validate()
+        return resp.names()

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

@@ -0,0 +1,89 @@
+# 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.user import User
+from gravitino.dto.audit_dto import AuditDTO
+
+
+@dataclass_json
+@dataclass
+class UserDTO(User):
+    """Represents a User Data Transfer Object (DTO)."""
+
+    _name: str = field(metadata=config(field_name="name"))
+    _roles: list[str] = field(default_factory=list, 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, UserDTO):
+            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() -> UserDTO.Builder:
+        return UserDTO.Builder()
+
+    def name(self) -> str:
+        return self._name
+
+    def roles(self) -> list[str]:
+        return self._roles if self._roles else []
+
+    def audit_info(self) -> Optional[AuditDTO]:
+        return self._audit
+
+    class Builder:
+        """Helper class to build a UserDTO object."""
+
+        def __init__(self) -> None:
+            self._name: str = ""
+            self._roles: list[str] = []
+            self._audit: Optional[AuditDTO] = None
+
+        def with_name(self, name: str) -> UserDTO.Builder:
+            self._name = name
+            return self
+
+        def with_roles(self, roles: list[str]) -> UserDTO.Builder:
+            if roles is not None:
+                self._roles = roles
+            return self
+
+        def with_audit(self, audit: AuditDTO) -> UserDTO.Builder:
+            self._audit = audit
+            return self
+
+        def build(self) -> UserDTO:
+            if not self._name:
+                raise ValueError("name cannot be null or empty")
+            return UserDTO(self._name, self._roles, self._audit)

clients/client-python/gravitino/dto/requests/user_add_request.py

@@ -0,0 +1,38 @@
+# 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.rest.rest_message import RESTRequest
+from gravitino.utils.precondition import Precondition
+
+
+@dataclass_json
+@dataclass
+class UserAddRequest(RESTRequest):
+    """Represents a request to add a user."""
+
+    _name: str = field(metadata=config(field_name="name"))
+
+    def validate(self) -> None:
+        Precondition.check_string_not_empty(
+            self._name, "name is required and cannot be empty"
+        )

clients/client-python/gravitino/dto/responses/remove_response.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 dataclasses import dataclass, field
+
+from dataclasses_json import config, dataclass_json
+
+from gravitino.dto.responses.base_response import BaseResponse
+from gravitino.utils.precondition import Precondition
+
+
+@dataclass_json
+@dataclass
+class RemoveResponse(BaseResponse):
+    """Represents a response for a remove operation."""
+
+    _removed: bool = field(metadata=config(field_name="removed"))
+
+    def removed(self) -> bool:
+        return self._removed
+
+    def validate(self) -> None:
+        """Validates the response.
+
+        Raises:
+            IllegalArgumentException: If the removed field is not set.
+        """
+        Precondition.check_argument(
+            self._removed is not None,
+            "Remove response must contain 'removed' field",
+        )