Skip to content

Commit acfafb5

Browse files
gabeigliogabeiglio
committed
Refactor REST exceptions, add tests, and fix edge cases in hive/sql catalog
1 parent abae20f commit acfafb5

File tree

8 files changed

+201
-60
lines changed

8 files changed

+201
-60
lines changed

‎pyiceberg/catalog/hive.py‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,7 @@
3232
CheckLockRequest,
3333
EnvironmentContext,
3434
FieldSchema,
35+
InvalidObjectException,
3536
InvalidOperationException,
3637
LockComponent,
3738
LockLevel,
@@ -83,6 +84,7 @@
8384
)
8485
from pyiceberg.table.sorting import UNSORTED_SORT_ORDER, SortOrder
8586
from pyiceberg.table.update import (
87+
AssertCreate,
8688
TableRequirement,
8789
TableUpdate,
8890
)
@@ -381,6 +383,9 @@ def _create_hive_table(self, open_client: Client, hive_table: HiveTable) -> None
381383
open_client.create_table(hive_table)
382384
except AlreadyExistsException as e:
383385
raise TableAlreadyExistsError(f"Table {hive_table.dbName}.{hive_table.tableName} already exists") from e
386+
except InvalidObjectException as e:
387+
if "database" in e.message:
388+
raise NoSuchNamespaceError(f"Namespace does not exists: {hive_table.dbName}]") from e
384389

385390
def _get_hive_table(self, open_client: Client, database_name: str, table_name: str) -> HiveTable:
386391
try:
@@ -535,6 +540,10 @@ def commit_table(
535540
try:
536541
hive_table = self._get_hive_table(open_client, database_name, table_name)
537542
current_table = self._convert_hive_into_iceberg(hive_table)
543+
544+
if AssertCreate() in requirements:
545+
raise TableAlreadyExistsError(f"Table already exists: {table_name}")
546+
538547
except NoSuchTableError:
539548
hive_table = None
540549
current_table = None

‎pyiceberg/catalog/rest/__init__.py‎

Lines changed: 23 additions & 34 deletions
Original file line numberDiff line numberDiff line change
@@ -35,18 +35,11 @@
3535
PropertiesUpdateSummary,
3636
)
3737
from pyiceberg.catalog.rest.auth import AuthManager, AuthManagerAdapter, AuthManagerFactory, LegacyOAuth2AuthManager
38-
from pyiceberg.catalog.rest.response import _handle_non_200_response
38+
from pyiceberg.catalog.rest.response import ErrorHandlers
3939
from pyiceberg.exceptions import (
4040
AuthorizationExpiredError,
41-
CommitFailedException,
42-
CommitStateUnknownException,
43-
NamespaceAlreadyExistsError,
44-
NamespaceNotEmptyError,
4541
NoSuchIdentifierError,
4642
NoSuchNamespaceError,
47-
NoSuchTableError,
48-
NoSuchViewError,
49-
TableAlreadyExistsError,
5043
UnauthorizedError,
5144
)
5245
from pyiceberg.io import AWS_ACCESS_KEY_ID, AWS_REGION, AWS_SECRET_ACCESS_KEY, AWS_SESSION_TOKEN
@@ -64,6 +57,7 @@
6457
from pyiceberg.table.metadata import TableMetadata
6558
from pyiceberg.table.sorting import UNSORTED_SORT_ORDER, SortOrder, assign_fresh_sort_order_ids
6659
from pyiceberg.table.update import (
60+
AssertCreate,
6761
TableRequirement,
6862
TableUpdate,
6963
)
@@ -361,7 +355,7 @@ def _fetch_config(self) -> None:
361355
try:
362356
response.raise_for_status()
363357
except HTTPError as exc:
364-
_handle_non_200_response(exc, {})
358+
ErrorHandlers.default_error_handler(exc)
365359
config_response = ConfigResponse.model_validate_json(response.text)
366360

367361
config = config_response.defaults
@@ -519,7 +513,7 @@ def _create_table(
519513
try:
520514
response.raise_for_status()
521515
except HTTPError as exc:
522-
_handle_non_200_response(exc, {409: TableAlreadyExistsError, 404: NoSuchNamespaceError})
516+
ErrorHandlers.table_error_handler(exc)
523517
return TableResponse.model_validate_json(response.text)
524518

525519
@retry(**_RETRY_ARGS)
@@ -592,7 +586,7 @@ def register_table(self, identifier: str | Identifier, metadata_location: str) -
592586
try:
593587
response.raise_for_status()
594588
except HTTPError as exc:
595-
_handle_non_200_response(exc, {409: TableAlreadyExistsError})
589+
ErrorHandlers.table_error_handler(exc)
596590

597591
table_response = TableResponse.model_validate_json(response.text)
598592
return self._response_to_table(self.identifier_to_tuple(identifier), table_response)
@@ -605,7 +599,7 @@ def list_tables(self, namespace: str | Identifier) -> list[Identifier]:
605599
try:
606600
response.raise_for_status()
607601
except HTTPError as exc:
608-
_handle_non_200_response(exc, {404: NoSuchNamespaceError})
602+
ErrorHandlers.namespace_error_handler(exc)
609603
return [(*table.namespace, table.name) for table in ListTablesResponse.model_validate_json(response.text).identifiers]
610604

611605
@retry(**_RETRY_ARGS)
@@ -623,7 +617,7 @@ def load_table(self, identifier: str | Identifier) -> Table:
623617
try:
624618
response.raise_for_status()
625619
except HTTPError as exc:
626-
_handle_non_200_response(exc, {404: NoSuchTableError})
620+
ErrorHandlers.table_error_handler(exc)
627621

628622
table_response = TableResponse.model_validate_json(response.text)
629623
return self._response_to_table(self.identifier_to_tuple(identifier), table_response)
@@ -637,7 +631,7 @@ def drop_table(self, identifier: str | Identifier, purge_requested: bool = False
637631
try:
638632
response.raise_for_status()
639633
except HTTPError as exc:
640-
_handle_non_200_response(exc, {404: NoSuchTableError})
634+
ErrorHandlers.table_error_handler(exc)
641635

642636
@retry(**_RETRY_ARGS)
643637
def purge_table(self, identifier: str | Identifier) -> None:
@@ -663,7 +657,7 @@ def rename_table(self, from_identifier: str | Identifier, to_identifier: str | I
663657
try:
664658
response.raise_for_status()
665659
except HTTPError as exc:
666-
_handle_non_200_response(exc, {404: NoSuchTableError, 409: TableAlreadyExistsError})
660+
ErrorHandlers.table_error_handler(exc)
667661

668662
return self.load_table(to_identifier)
669663

@@ -686,7 +680,7 @@ def list_views(self, namespace: str | Identifier) -> list[Identifier]:
686680
try:
687681
response.raise_for_status()
688682
except HTTPError as exc:
689-
_handle_non_200_response(exc, {404: NoSuchNamespaceError})
683+
ErrorHandlers.view_error_handler(exc)
690684
return [(*view.namespace, view.name) for view in ListViewsResponse.model_validate_json(response.text).identifiers]
691685

692686
@retry(**_RETRY_ARGS)
@@ -724,15 +718,10 @@ def commit_table(
724718
try:
725719
response.raise_for_status()
726720
except HTTPError as exc:
727-
_handle_non_200_response(
728-
exc,
729-
{
730-
409: CommitFailedException,
731-
500: CommitStateUnknownException,
732-
502: CommitStateUnknownException,
733-
504: CommitStateUnknownException,
734-
},
735-
)
721+
if AssertCreate() in requirements:
722+
ErrorHandlers.table_error_handler(exc)
723+
else:
724+
ErrorHandlers.commit_error_handler(exc)
736725
return CommitTableResponse.model_validate_json(response.text)
737726

738727
@retry(**_RETRY_ARGS)
@@ -743,7 +732,7 @@ def create_namespace(self, namespace: str | Identifier, properties: Properties =
743732
try:
744733
response.raise_for_status()
745734
except HTTPError as exc:
746-
_handle_non_200_response(exc, {409: NamespaceAlreadyExistsError})
735+
ErrorHandlers.namespace_error_handler(exc)
747736

748737
@retry(**_RETRY_ARGS)
749738
def drop_namespace(self, namespace: str | Identifier) -> None:
@@ -753,7 +742,7 @@ def drop_namespace(self, namespace: str | Identifier) -> None:
753742
try:
754743
response.raise_for_status()
755744
except HTTPError as exc:
756-
_handle_non_200_response(exc, {404: NoSuchNamespaceError, 409: NamespaceNotEmptyError})
745+
ErrorHandlers.drop_namespace_error_handler(exc)
757746

758747
@retry(**_RETRY_ARGS)
759748
def list_namespaces(self, namespace: str | Identifier = ()) -> list[Identifier]:
@@ -768,7 +757,7 @@ def list_namespaces(self, namespace: str | Identifier = ()) -> list[Identifier]:
768757
try:
769758
response.raise_for_status()
770759
except HTTPError as exc:
771-
_handle_non_200_response(exc, {404: NoSuchNamespaceError})
760+
ErrorHandlers.namespace_error_handler(exc)
772761

773762
return ListNamespaceResponse.model_validate_json(response.text).namespaces
774763

@@ -780,7 +769,7 @@ def load_namespace_properties(self, namespace: str | Identifier) -> Properties:
780769
try:
781770
response.raise_for_status()
782771
except HTTPError as exc:
783-
_handle_non_200_response(exc, {404: NoSuchNamespaceError})
772+
ErrorHandlers.namespace_error_handler(exc)
784773

785774
return NamespaceResponse.model_validate_json(response.text).properties
786775

@@ -795,7 +784,7 @@ def update_namespace_properties(
795784
try:
796785
response.raise_for_status()
797786
except HTTPError as exc:
798-
_handle_non_200_response(exc, {404: NoSuchNamespaceError})
787+
ErrorHandlers.namespace_error_handler(exc)
799788
parsed_response = UpdateNamespacePropertiesResponse.model_validate_json(response.text)
800789
return PropertiesUpdateSummary(
801790
removed=parsed_response.removed,
@@ -817,7 +806,7 @@ def namespace_exists(self, namespace: str | Identifier) -> bool:
817806
try:
818807
response.raise_for_status()
819808
except HTTPError as exc:
820-
_handle_non_200_response(exc, {})
809+
ErrorHandlers.namespace_error_handler(exc)
821810

822811
return False
823812

@@ -843,7 +832,7 @@ def table_exists(self, identifier: str | Identifier) -> bool:
843832
try:
844833
response.raise_for_status()
845834
except HTTPError as exc:
846-
_handle_non_200_response(exc, {})
835+
ErrorHandlers.table_error_handler(exc)
847836

848837
return False
849838

@@ -868,7 +857,7 @@ def view_exists(self, identifier: str | Identifier) -> bool:
868857
try:
869858
response.raise_for_status()
870859
except HTTPError as exc:
871-
_handle_non_200_response(exc, {})
860+
ErrorHandlers.view_error_handler(exc)
872861

873862
return False
874863

@@ -880,7 +869,7 @@ def drop_view(self, identifier: str) -> None:
880869
try:
881870
response.raise_for_status()
882871
except HTTPError as exc:
883-
_handle_non_200_response(exc, {404: NoSuchViewError})
872+
ErrorHandlers.view_error_handler(exc)
884873

885874
def close(self) -> None:
886875
"""Close the catalog and release Session connection adapters.

‎pyiceberg/catalog/rest/response.py‎

Lines changed: 90 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -15,20 +15,29 @@
1515
# specific language governing permissions and limitations
1616
# under the License.
1717
from json import JSONDecodeError
18-
from typing import Literal
18+
from typing import Literal, TypeAlias
1919

2020
from pydantic import Field, ValidationError
2121
from requests import HTTPError
2222

2323
from pyiceberg.exceptions import (
2424
AuthorizationExpiredError,
2525
BadRequestError,
26+
CommitFailedException,
27+
CommitStateUnknownException,
2628
ForbiddenError,
29+
NamespaceAlreadyExistsError,
30+
NamespaceNotEmptyError,
31+
NoSuchNamespaceError,
32+
NoSuchTableError,
33+
NoSuchViewError,
2734
OAuthError,
2835
RESTError,
2936
ServerError,
3037
ServiceUnavailableError,
38+
TableAlreadyExistsError,
3139
UnauthorizedError,
40+
ViewAlreadyExistsError,
3241
)
3342
from pyiceberg.typedef import IcebergBaseModel
3443

@@ -60,33 +69,92 @@ class OAuthErrorResponse(IcebergBaseModel):
6069
error_uri: str | None = None
6170

6271

63-
def _handle_non_200_response(exc: HTTPError, error_handler: dict[int, type[Exception]]) -> None:
72+
_ErrorHandler: TypeAlias = dict[int, type[Exception]]
73+
74+
75+
class ErrorHandlers:
76+
"""
77+
Utility class providing static methods to handle HTTP errors for table, namespace, and view operations.
78+
79+
Maps HTTP error responses to appropriate custom exceptions, ensuring consistent error handling.
80+
"""
81+
82+
@staticmethod
83+
def default_error_handler(exc: HTTPError) -> None:
84+
_handle_non_200_response(exc, {})
85+
86+
@staticmethod
87+
def namespace_error_handler(exc: HTTPError) -> None:
88+
handler: _ErrorHandler = {
89+
400: BadRequestError,
90+
404: NoSuchNamespaceError,
91+
409: NamespaceAlreadyExistsError,
92+
422: RESTError,
93+
}
94+
95+
if "NamespaceNotEmpty" in exc.response.text:
96+
handler[400] = NamespaceNotEmptyError
97+
98+
_handle_non_200_response(exc, handler)
99+
100+
@staticmethod
101+
def drop_namespace_error_handler(exc: HTTPError) -> None:
102+
handler: _ErrorHandler = {404: NoSuchNamespaceError, 409: NamespaceNotEmptyError}
103+
104+
_handle_non_200_response(exc, handler)
105+
106+
@staticmethod
107+
def table_error_handler(exc: HTTPError) -> None:
108+
handler: _ErrorHandler = {404: NoSuchTableError, 409: TableAlreadyExistsError}
109+
110+
if "NoSuchNamespace" in exc.response.text:
111+
handler[404] = NoSuchNamespaceError
112+
113+
_handle_non_200_response(exc, handler)
114+
115+
@staticmethod
116+
def commit_error_handler(exc: HTTPError) -> None:
117+
handler: _ErrorHandler = {
118+
404: NoSuchTableError,
119+
409: CommitFailedException,
120+
500: CommitStateUnknownException,
121+
502: CommitStateUnknownException,
122+
503: CommitStateUnknownException,
123+
504: CommitStateUnknownException,
124+
}
125+
126+
_handle_non_200_response(exc, handler)
127+
128+
@staticmethod
129+
def view_error_handler(exc: HTTPError) -> None:
130+
handler: _ErrorHandler = {404: NoSuchViewError, 409: ViewAlreadyExistsError}
131+
132+
if "NoSuchNamespace" in exc.response.text:
133+
handler[404] = NoSuchNamespaceError
134+
135+
_handle_non_200_response(exc, handler)
136+
137+
138+
def _handle_non_200_response(exc: HTTPError, handler: _ErrorHandler) -> None:
64139
exception: type[Exception]
65140

66141
if exc.response is None:
67142
raise ValueError("Did not receive a response")
68143

69144
code = exc.response.status_code
70-
if code in error_handler:
71-
exception = error_handler[code]
72-
elif code == 400:
73-
exception = BadRequestError
74-
elif code == 401:
75-
exception = UnauthorizedError
76-
elif code == 403:
77-
exception = ForbiddenError
78-
elif code == 422:
79-
exception = RESTError
80-
elif code == 419:
81-
exception = AuthorizationExpiredError
82-
elif code == 501:
83-
exception = NotImplementedError
84-
elif code == 503:
85-
exception = ServiceUnavailableError
86-
elif 500 <= code < 600:
87-
exception = ServerError
88-
else:
89-
exception = RESTError
145+
146+
default_handler: _ErrorHandler = {
147+
400: BadRequestError,
148+
401: UnauthorizedError,
149+
403: ForbiddenError,
150+
419: AuthorizationExpiredError,
151+
422: RESTError,
152+
501: NotImplementedError,
153+
503: ServiceUnavailableError,
154+
}
155+
156+
# Merge handler passed with default handler map, if no match exception will be ServerError or RESTError
157+
exception = handler.get(code, default_handler.get(code, ServerError if 500 <= code < 600 else RESTError))
90158

91159
try:
92160
if exception == OAuthError:

0 commit comments

Comments
 (0)