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

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

This PR is split from #10783 to make review easier. It focuses on the
**User** part of the authorization model; Group, Role, Privilege,
SecurableObject and grant/revoke operations will follow in subsequent
PRs.

Changes in this PR:

1. **API layer** (`gravitino/api/authorization/`)
   - `User` — user interface with `name()` / `roles()` accessors

2. **DTO layer** (`gravitino/dto/authorization/`)
   - `UserDTO` — immutable DTO with builder pattern, serialization
     and audit info

3. **Request / Response DTOs** (`gravitino/dto/`)
   - `UserAddRequest`
   - `UserResponse` / `UserListResponse` / `UserNamesListResponse`
   - `RemoveResponse` (shared boolean-remove response used across
     authorization operations)

4. **Client layer** (`gravitino/client/gravitino_metalake.py`)
   - Five new public methods on `GravitinoMetalake`:
     `add_user` / `get_user` / `remove_user` / `list_users` /
     `list_user_names`

5. **Exception handling** (`gravitino/exceptions/`)
   - `NoSuchUserException`, `UserAlreadyExistsException`
   - `UserErrorHandler` mapping REST error codes to the above

### Why are the changes needed?

The Python SDK currently has no authorization management capabilities,
while the Java SDK and REST API have full support. This PR starts
bringing the Python SDK to feature parity with the Java SDK, beginning
with User management as the most foundational entity.

Fix: #11098

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

Yes. New public API:
- `GravitinoMetalake.add_user(name)` / `get_user(name)` /
  `remove_user(name)` / `list_users()` / `list_user_names()`
- New exceptions: `NoSuchUserException`, `UserAlreadyExistsException`

No existing APIs or behaviors are changed.

### How was this patch tested?

28 new unit tests (all passing, `ruff` + `pylint`):

- **DTO tests** 
(`tests/unittests/dto/test_user_dto.py`,
`tests/unittests/dto/responses/test_user_response.py`) — cover builder,
serialization / deserialization roundtrip, equality
- **Client-level mock tests**
(`tests/unittests/client/test_metalake_user_operations.py`) — cover URL,
HTTP method, request body, response parsing, and error propagation for
all five `GravitinoMetalake` methods
- **Error-handler tests** 
(added to `tests/unittests/test_error_handler.py`) — verify the full
error code mapping table for `UserErrorHandler`

Integration tests will be added in the final PR of the split together
with grant/revoke operations, where end-to-end multi-entity behavior
needs to be validated against a real server.

cc @jerryshao — could you help review this when you have time? Thanks!

---------

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

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_client.py

@@ -20,6 +20,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
@@ -369,3 +370,71 @@ def set_owner(
             UnsupportedOperationException: If the operation is not supported.
         """
         self.get_metalake().set_owner(metadata_object, owner_name, owner_type)
+
+    # User operations
+
+    def add_user(self, user: str) -> User:
+        """Add a user to the metalake.
+
+        Args:
+            user: The name of the user.
+
+        Returns:
+            The added User object.
+
+        Raises:
+            UserAlreadyExistsException: If a user with the same name already exists.
+            NoSuchMetalakeException: If the metalake does not exist.
+        """
+        return self.get_metalake().add_user(user)
+
+    def remove_user(self, user: str) -> bool:
+        """Remove a user from the metalake.
+
+        Args:
+            user: The name of the user.
+
+        Returns:
+            True if the user was removed, False if the user did not exist.
+
+        Raises:
+            NoSuchMetalakeException: If the metalake does not exist.
+        """
+        return self.get_metalake().remove_user(user)
+
+    def get_user(self, user: str) -> User:
+        """Get a user by name from the metalake.
+
+        Args:
+            user: The name of the user.
+
+        Returns:
+            The User object.
+
+        Raises:
+            NoSuchUserException: If the user does not exist.
+            NoSuchMetalakeException: If the metalake does not exist.
+        """
+        return self.get_metalake().get_user(user)
+
+    def list_users(self) -> list[User]:
+        """List all users with details under the metalake.
+
+        Returns:
+            A list of User objects.
+
+        Raises:
+            NoSuchMetalakeException: If the metalake does not exist.
+        """
+        return self.get_metalake().list_users()
+
+    def list_user_names(self) -> list[str]:
+        """List all user names under the metalake.
+
+        Returns:
+            A list of user name strings.
+
+        Raises:
+            NoSuchMetalakeException: If the metalake does not exist.
+        """
+        return self.get_metalake().list_user_names()

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,106 @@ 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.
+
+        Args:
+            user: The name of the user.
+
+        Returns:
+            The added User object.
+
+        Raises:
+            UserAlreadyExistsException: If a user with the same name already exists.
+            NoSuchMetalakeException: If the metalake does not exist.
+        """
+        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.
+
+        Args:
+            user: The name of the user.
+
+        Returns:
+            True if the user was removed, False if the user did not exist.
+
+        Raises:
+            NoSuchMetalakeException: If the metalake does not exist.
+        """
+        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.
+
+        Args:
+            user: The name of the user.
+
+        Returns:
+            The User object.
+
+        Raises:
+            NoSuchUserException: If the user does not exist.
+            NoSuchMetalakeException: If the metalake does not exist.
+        """
+        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.
+
+        Returns:
+            A list of User objects.
+
+        Raises:
+            NoSuchMetalakeException: If the metalake does not exist.
+        """
+        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.
+
+        Returns:
+            A list of user name strings.
+
+        Raises:
+            NoSuchMetalakeException: If the metalake does not exist.
+        """
+        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,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.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: 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, 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 list(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: tuple[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 = tuple(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)