This is an automated email from the ASF dual-hosted git repository.
jshao pushed a commit to branch branch-0.6
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/branch-0.6 by this push:
new 3591709ac [#4227] improvement(client-python): Add Catalog Error
Handler and related exceptions, test cases in client-python (#4456)
3591709ac is described below
commit 3591709ac6a6cde9f1e0815089a9090d581550b3
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Fri Aug 9 15:11:45 2024 +0800
[#4227] improvement(client-python): Add Catalog Error Handler and related
exceptions, test cases in client-python (#4456)
### What changes were proposed in this pull request?
- Add Catalog Error Handler and related exceptions, test cases in
`client-python` based on `client-java`
- Add lacking ITs in `test_catalog.py`
### Why are the changes needed?
Fix: #4227
### Does this PR introduce _any_ user-facing change?
No
### How was this patch tested?
UT Added and test with `./gradlew client:client-python:test`
Co-authored-by: Jing-Jia Hung <[email protected]>
---
.../gravitino/client/gravitino_metalake.py | 19 ++++--
clients/client-python/gravitino/constants/error.py | 5 ++
.../client-python/gravitino/dto/dto_converters.py | 4 +-
.../dto/requests/catalog_update_request.py | 23 +++----
.../gravitino/dto/responses/catalog_response.py | 10 +--
clients/client-python/gravitino/exceptions/base.py | 8 +++
.../exceptions/handlers/catalog_error_handler.py | 52 +++++++++++++++
.../tests/integration/test_catalog.py | 76 +++++++++++++++++++++-
.../tests/unittests/test_error_handler.py | 43 ++++++++++++
9 files changed, 213 insertions(+), 27 deletions(-)
diff --git a/clients/client-python/gravitino/client/gravitino_metalake.py
b/clients/client-python/gravitino/client/gravitino_metalake.py
index 9ef787a07..1cf5c4f72 100644
--- a/clients/client-python/gravitino/client/gravitino_metalake.py
+++ b/clients/client-python/gravitino/client/gravitino_metalake.py
@@ -30,6 +30,7 @@ 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
from gravitino.dto.responses.entity_list_response import EntityListResponse
+from gravitino.exceptions.handlers.catalog_error_handler import
CATALOG_ERROR_HANDLER
from gravitino.utils import HTTPClient
@@ -66,7 +67,7 @@ class GravitinoMetalake(MetalakeDTO):
A list of the catalog names under this metalake.
"""
url = f"api/metalakes/{self.name()}/catalogs"
- response = self.rest_client.get(url)
+ response = self.rest_client.get(url,
error_handler=CATALOG_ERROR_HANDLER)
entity_list = EntityListResponse.from_json(response.body,
infer_missing=True)
entity_list.validate()
return [identifier.name() for identifier in entity_list.identifiers()]
@@ -82,7 +83,9 @@ class GravitinoMetalake(MetalakeDTO):
"""
params = {"details": "true"}
url = f"api/metalakes/{self.name()}/catalogs"
- response = self.rest_client.get(url, params=params)
+ response = self.rest_client.get(
+ url, params=params, error_handler=CATALOG_ERROR_HANDLER
+ )
catalog_list = CatalogListResponse.from_json(response.body,
infer_missing=True)
return [
@@ -103,7 +106,7 @@ class GravitinoMetalake(MetalakeDTO):
The Catalog with specified name.
"""
url = self.API_METALAKES_CATALOGS_PATH.format(self.name(), name)
- response = self.rest_client.get(url)
+ response = self.rest_client.get(url,
error_handler=CATALOG_ERROR_HANDLER)
catalog_resp = CatalogResponse.from_json(response.body,
infer_missing=True)
return DTOConverters.to_catalog(
@@ -145,7 +148,9 @@ class GravitinoMetalake(MetalakeDTO):
catalog_create_request.validate()
url = f"api/metalakes/{self.name()}/catalogs"
- response = self.rest_client.post(url, json=catalog_create_request)
+ response = self.rest_client.post(
+ url, json=catalog_create_request,
error_handler=CATALOG_ERROR_HANDLER
+ )
catalog_resp = CatalogResponse.from_json(response.body,
infer_missing=True)
return DTOConverters.to_catalog(
@@ -172,7 +177,9 @@ class GravitinoMetalake(MetalakeDTO):
updates_request.validate()
url = self.API_METALAKES_CATALOGS_PATH.format(self.name(), name)
- response = self.rest_client.put(url, json=updates_request)
+ response = self.rest_client.put(
+ url, json=updates_request, error_handler=CATALOG_ERROR_HANDLER
+ )
catalog_response = CatalogResponse.from_json(response.body,
infer_missing=True)
catalog_response.validate()
@@ -191,7 +198,7 @@ class GravitinoMetalake(MetalakeDTO):
"""
try:
url = self.API_METALAKES_CATALOGS_PATH.format(self.name(), name)
- response = self.rest_client.delete(url)
+ response = self.rest_client.delete(url,
error_handler=CATALOG_ERROR_HANDLER)
drop_response = DropResponse.from_json(response.body,
infer_missing=True)
drop_response.validate()
diff --git a/clients/client-python/gravitino/constants/error.py
b/clients/client-python/gravitino/constants/error.py
index e4b0ab61e..877d523ad 100644
--- a/clients/client-python/gravitino/constants/error.py
+++ b/clients/client-python/gravitino/constants/error.py
@@ -20,6 +20,7 @@ under the License.
from enum import IntEnum
from gravitino.exceptions.base import (
+ ConnectionFailedException,
RESTException,
IllegalArgumentException,
NotFoundException,
@@ -54,6 +55,9 @@ class ErrorConstants(IntEnum):
# Error codes for unsupported operation.
UNSUPPORTED_OPERATION_CODE = 1006
+ # Error codes for connect to catalog failed.
+ CONNECTION_FAILED_CODE = 1007
+
# Error codes for invalid state.
UNKNOWN_ERROR_CODE = 1100
@@ -66,6 +70,7 @@ EXCEPTION_MAPPING = {
AlreadyExistsException: ErrorConstants.ALREADY_EXISTS_CODE,
NotEmptyException: ErrorConstants.NON_EMPTY_CODE,
UnsupportedOperationException: ErrorConstants.UNSUPPORTED_OPERATION_CODE,
+ ConnectionFailedException: ErrorConstants.CONNECTION_FAILED_CODE,
}
ERROR_CODE_MAPPING = {v: k for k, v in EXCEPTION_MAPPING.items()}
diff --git a/clients/client-python/gravitino/dto/dto_converters.py
b/clients/client-python/gravitino/dto/dto_converters.py
index 86e614557..a55f053b3 100644
--- a/clients/client-python/gravitino/dto/dto_converters.py
+++ b/clients/client-python/gravitino/dto/dto_converters.py
@@ -83,6 +83,8 @@ class DTOConverters:
change.property(), change.value()
)
if isinstance(change, CatalogChange.RemoveProperty):
- return
CatalogUpdateRequest.RemoveCatalogPropertyRequest(change.property())
+ return CatalogUpdateRequest.RemoveCatalogPropertyRequest(
+ change.get_property()
+ )
raise ValueError(f"Unknown change type: {type(change).__name__}")
diff --git
a/clients/client-python/gravitino/dto/requests/catalog_update_request.py
b/clients/client-python/gravitino/dto/requests/catalog_update_request.py
index d3ba6298a..6ae06b118 100644
--- a/clients/client-python/gravitino/dto/requests/catalog_update_request.py
+++ b/clients/client-python/gravitino/dto/requests/catalog_update_request.py
@@ -62,9 +62,8 @@ class CatalogUpdateRequest:
Raises:
IllegalArgumentException if the new name is not set.
"""
- assert (
- self._new_name is None
- ), '"newName" field is required and cannot be empty'
+ if not self._new_name:
+ raise ValueError('"newName" field is required and cannot be
empty')
@dataclass
class UpdateCatalogCommentRequest(CatalogUpdateRequestBase):
@@ -81,9 +80,8 @@ class CatalogUpdateRequest:
return CatalogChange.update_comment(self._new_comment)
def validate(self):
- assert (
- self._new_comment is None
- ), '"newComment" field is required and cannot be empty'
+ if not self._new_comment:
+ raise ValueError('"newComment" field is required and cannot be
empty')
@dataclass
class SetCatalogPropertyRequest(CatalogUpdateRequestBase):
@@ -104,10 +102,10 @@ class CatalogUpdateRequest:
return CatalogChange.set_property(self._property, self._value)
def validate(self):
- assert (
- self._property is None
- ), '"property" field is required and cannot be empty'
- assert self._value is None, '"value" field is required and cannot
be empty'
+ if not self._property:
+ raise ValueError('"property" field is required and cannot be
empty')
+ if not self._value:
+ raise ValueError('"value" field is required and cannot be
empty')
class RemoveCatalogPropertyRequest(CatalogUpdateRequestBase):
"""Request to remove a property from a catalog."""
@@ -123,6 +121,5 @@ class CatalogUpdateRequest:
return CatalogChange.remove_property(self._property)
def validate(self):
- assert (
- self._property is None
- ), '"property" field is required and cannot be empty'
+ if not self._property:
+ raise ValueError('"property" field is required and cannot be
empty')
diff --git a/clients/client-python/gravitino/dto/responses/catalog_response.py
b/clients/client-python/gravitino/dto/responses/catalog_response.py
index 711816563..d1cefa0ba 100644
--- a/clients/client-python/gravitino/dto/responses/catalog_response.py
+++ b/clients/client-python/gravitino/dto/responses/catalog_response.py
@@ -39,12 +39,14 @@ class CatalogResponse(BaseResponse):
"""
super().validate()
- assert self.catalog is not None, "catalog must not be null"
+ assert self._catalog is not None, "catalog must not be null"
assert (
- self.catalog.name() is not None
+ self._catalog.name() is not None
), "catalog 'name' must not be null and empty"
- assert self.catalog.type() is not None, "catalog 'type' must not be
null"
- assert self.catalog.audit_info() is not None, "catalog 'audit' must
not be null"
+ assert self._catalog.type() is not None, "catalog 'type' must not be
null"
+ assert (
+ self._catalog.audit_info() is not None
+ ), "catalog 'audit' must not be null"
def catalog(self) -> CatalogDTO:
return self._catalog
diff --git a/clients/client-python/gravitino/exceptions/base.py
b/clients/client-python/gravitino/exceptions/base.py
index a7d6eee18..674121a5d 100644
--- a/clients/client-python/gravitino/exceptions/base.py
+++ b/clients/client-python/gravitino/exceptions/base.py
@@ -83,6 +83,10 @@ class SchemaAlreadyExistsException(AlreadyExistsException):
"""An exception thrown when a schema already exists."""
+class CatalogAlreadyExistsException(AlreadyExistsException):
+ """An exception thrown when a resource already exists."""
+
+
class NotEmptyException(GravitinoRuntimeException):
"""Base class for all exceptions thrown when a resource is not empty."""
@@ -95,6 +99,10 @@ class UnknownError(RuntimeError):
"""An exception thrown when other unknown exception is thrown"""
+class ConnectionFailedException(GravitinoRuntimeException):
+ """An exception thrown when connect to catalog failed."""
+
+
class UnauthorizedException(GravitinoRuntimeException):
"""An exception thrown when a user is not authorized to perform an
action."""
diff --git
a/clients/client-python/gravitino/exceptions/handlers/catalog_error_handler.py
b/clients/client-python/gravitino/exceptions/handlers/catalog_error_handler.py
new file mode 100644
index 000000000..e714edad3
--- /dev/null
+++
b/clients/client-python/gravitino/exceptions/handlers/catalog_error_handler.py
@@ -0,0 +1,52 @@
+"""
+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 gravitino.constants.error import ErrorConstants
+from gravitino.dto.responses.error_response import ErrorResponse
+from gravitino.exceptions.handlers.rest_error_handler import RestErrorHandler
+from gravitino.exceptions.base import (
+ ConnectionFailedException,
+ NoSuchMetalakeException,
+ NoSuchCatalogException,
+ CatalogAlreadyExistsException,
+)
+
+
+class CatalogErrorHandler(RestErrorHandler):
+
+ def handle(self, error_response: ErrorResponse):
+
+ error_message = error_response.format_error_message()
+ code = error_response.code()
+ exception_type = error_response.type()
+
+ if code == ErrorConstants.CONNECTION_FAILED_CODE:
+ raise ConnectionFailedException(error_message)
+ if code == ErrorConstants.NOT_FOUND_CODE:
+ if exception_type == NoSuchMetalakeException.__name__:
+ raise NoSuchMetalakeException(error_message)
+ if exception_type == NoSuchCatalogException.__name__:
+ raise NoSuchCatalogException(error_message)
+ if code == ErrorConstants.ALREADY_EXISTS_CODE:
+ raise CatalogAlreadyExistsException(error_message)
+
+ super().handle(error_response)
+
+
+CATALOG_ERROR_HANDLER = CatalogErrorHandler()
diff --git a/clients/client-python/tests/integration/test_catalog.py
b/clients/client-python/tests/integration/test_catalog.py
index 3f8d2ae4b..2cf13f2cd 100644
--- a/clients/client-python/tests/integration/test_catalog.py
+++ b/clients/client-python/tests/integration/test_catalog.py
@@ -26,6 +26,11 @@ from gravitino import (
GravitinoClient,
Catalog,
)
+from gravitino.api.catalog_change import CatalogChange
+from gravitino.exceptions.base import (
+ CatalogAlreadyExistsException,
+ NoSuchCatalogException,
+)
from tests.integration.integration_test_env import IntegrationTestEnv
@@ -36,6 +41,7 @@ class TestCatalog(IntegrationTestEnv):
metalake_name: str = "TestSchema_metalake" + str(randint(1, 10000))
catalog_name: str = "testCatalog"
+ catalog_comment: str = "catalogComment"
catalog_location_prop: str = "location" # Fileset Catalog must set
`location`
catalog_provider: str = "hadoop"
@@ -59,11 +65,13 @@ class TestCatalog(IntegrationTestEnv):
self.gravitino_client = GravitinoClient(
uri="http://localhost:8090";, metalake_name=self.metalake_name
)
- self.gravitino_client.create_catalog(
- name=self.catalog_name,
+
+ def create_catalog(self, catalog_name) -> Catalog:
+ return self.gravitino_client.create_catalog(
+ name=catalog_name,
catalog_type=Catalog.Type.FILESET,
provider=self.catalog_provider,
- comment="",
+ comment=self.catalog_comment,
properties={self.catalog_location_prop: "/tmp/test_schema"},
)
@@ -86,5 +94,67 @@ class TestCatalog(IntegrationTestEnv):
logger.error("Clean test data failed: %s", e)
def test_list_catalogs(self):
+ self.create_catalog(self.catalog_name)
catalog_names = self.gravitino_client.list_catalogs()
self.assertTrue(self.catalog_name in catalog_names)
+
+ def test_create_catalog(self):
+ catalog = self.create_catalog(self.catalog_name)
+ self.assertEqual(catalog.name(), self.catalog_name)
+ self.assertEqual(
+ catalog.properties(), {self.catalog_location_prop:
"/tmp/test_schema"}
+ )
+
+ def test_failed_create_catalog(self):
+ self.create_catalog(self.catalog_name)
+ with self.assertRaises(CatalogAlreadyExistsException):
+ _ = self.create_catalog(self.catalog_name)
+
+ def test_alter_catalog(self):
+ catalog = self.create_catalog(self.catalog_name)
+
+ catalog_new_name = self.catalog_name + "_new"
+ catalog_new_comment = self.catalog_comment + "_new"
+ catalog_properties_new_value = self.catalog_location_prop + "_new"
+ catalog_properties_new_key: str = "catalog_properties_new_key"
+
+ changes = (
+ CatalogChange.rename(catalog_new_name),
+ CatalogChange.update_comment(catalog_new_comment),
+ CatalogChange.set_property(
+ catalog_properties_new_key, catalog_properties_new_value
+ ),
+ )
+
+ catalog = self.gravitino_client.alter_catalog(self.catalog_name,
*changes)
+ self.assertEqual(catalog.name(), catalog_new_name)
+ self.assertEqual(catalog.comment(), catalog_new_comment)
+ self.assertEqual(
+ catalog.properties().get(catalog_properties_new_key),
+ catalog_properties_new_value,
+ )
+ self.catalog_name = self.catalog_name + "_new"
+
+ def test_drop_catalog(self):
+ self.create_catalog(self.catalog_name)
+
self.assertTrue(self.gravitino_client.drop_catalog(name=self.catalog_name))
+
+ def test_list_catalogs_info(self):
+ self.create_catalog(self.catalog_name)
+ catalogs_info = self.gravitino_client.list_catalogs_info()
+ self.assertTrue(any(item.name() == self.catalog_name for item in
catalogs_info))
+
+ def test_load_catalog(self):
+ self.create_catalog(self.catalog_name)
+ catalog = self.gravitino_client.load_catalog(self.catalog_name)
+ self.assertIsNotNone(catalog)
+ self.assertEqual(catalog.name(), self.catalog_name)
+ self.assertEqual(catalog.comment(), self.catalog_comment)
+ self.assertEqual(
+ catalog.properties(), {self.catalog_location_prop:
"/tmp/test_schema"}
+ )
+ self.assertEqual(catalog.audit_info().creator(), "anonymous")
+
+ def test_failed_load_catalog(self):
+ with self.assertRaises(NoSuchCatalogException):
+ self.gravitino_client.load_catalog(self.catalog_name)
diff --git a/clients/client-python/tests/unittests/test_error_handler.py
b/clients/client-python/tests/unittests/test_error_handler.py
index fd92af495..b2695471f 100644
--- a/clients/client-python/tests/unittests/test_error_handler.py
+++ b/clients/client-python/tests/unittests/test_error_handler.py
@@ -34,11 +34,14 @@ from gravitino.exceptions.base import (
NotEmptyException,
SchemaAlreadyExistsException,
UnsupportedOperationException,
+ ConnectionFailedException,
+ CatalogAlreadyExistsException,
)
from gravitino.exceptions.handlers.rest_error_handler import REST_ERROR_HANDLER
from gravitino.exceptions.handlers.fileset_error_handler import
FILESET_ERROR_HANDLER
from gravitino.exceptions.handlers.metalake_error_handler import
METALAKE_ERROR_HANDLER
+from gravitino.exceptions.handlers.catalog_error_handler import
CATALOG_ERROR_HANDLER
from gravitino.exceptions.handlers.schema_error_handler import
SCHEMA_ERROR_HANDLER
@@ -151,6 +154,46 @@ class TestErrorHandler(unittest.TestCase):
ErrorResponse.generate_error_response(Exception, "mock error")
)
+ def test_catalog_error_handler(self):
+
+ with self.assertRaises(ConnectionFailedException):
+ CATALOG_ERROR_HANDLER.handle(
+ ErrorResponse.generate_error_response(
+ ConnectionFailedException, "mock error"
+ )
+ )
+
+ with self.assertRaises(NoSuchMetalakeException):
+ CATALOG_ERROR_HANDLER.handle(
+ ErrorResponse.generate_error_response(
+ NoSuchMetalakeException, "mock error"
+ )
+ )
+
+ with self.assertRaises(NoSuchCatalogException):
+ CATALOG_ERROR_HANDLER.handle(
+ ErrorResponse.generate_error_response(
+ NoSuchCatalogException, "mock error"
+ )
+ )
+
+ with self.assertRaises(CatalogAlreadyExistsException):
+ CATALOG_ERROR_HANDLER.handle(
+ ErrorResponse.generate_error_response(
+ CatalogAlreadyExistsException, "mock error"
+ )
+ )
+
+ with self.assertRaises(InternalError):
+ CATALOG_ERROR_HANDLER.handle(
+ ErrorResponse.generate_error_response(InternalError, "mock
error")
+ )
+
+ with self.assertRaises(RESTException):
+ CATALOG_ERROR_HANDLER.handle(
+ ErrorResponse.generate_error_response(Exception, "mock error")
+ )
+
def test_schema_error_handler(self):
with self.assertRaises(NoSuchCatalogException):
