add custom auth implementation parsing

Author: sungwy

pyiceberg/catalog/rest/__init__.py

@@ -134,6 +134,7 @@ class IdentifierKind(Enum):
 SIGV4_SERVICE = "rest.signing-name"
 OAUTH2_SERVER_URI = "oauth2-server-uri"
 SNAPSHOT_LOADING_MODE = "snapshot-loading-mode"
+AUTH = "auth"
 
 NAMESPACE_SEPARATOR = b"\x1f".decode(UTF8)
 
@@ -247,7 +248,19 @@ def _create_session(self) -> Session:
                 elif ssl_client_cert := ssl_client.get(CERT):
                     session.cert = ssl_client_cert
 
-        session.auth = AuthManagerAdapter(self._create_legacy_oauth2_auth_manager(session))
+        if auth_config := self.properties.get(AUTH):
+            # set up auth_manager based on the properties
+            auth_type = auth_config.get("type")
+            if auth_type is None:
+                raise ValueError("auth.type must be defined")
+            auth_type_config = auth_config.get(auth_type, {})
+            if auth_impl := auth_config.get("impl"):
+                session.auth = AuthManagerAdapter(AuthManagerFactory.create(auth_impl, auth_type_config))
+            else:
+                session.auth = AuthManagerAdapter(AuthManagerFactory.create(auth_type, auth_type_config))
+        else:
+            session.auth = AuthManagerAdapter(self._create_legacy_oauth2_auth_manager(session))
+
         # Set HTTP headers
         self._config_headers(session)
 

tests/catalog/test_rest.py

@@ -1519,6 +1519,42 @@ def test_request_session_with_ssl_client_cert() -> None:
     assert "Could not find the TLS certificate file, invalid path: path_to_client_cert" in str(e.value)
 
 
+def test_rest_catalog_with_basic_auth_type() -> None:
+    # Given
+    catalog_properties = {
+        "uri": TEST_URI,
+        "auth": {
+            "type": "basic",
+            "basic": {
+                "username": "one",
+            },
+        },
+    }
+    with pytest.raises(TypeError) as e:
+        # Missing namespace
+        RestCatalog("rest", **catalog_properties)  # type: ignore
+    assert "BasicAuthManager.__init__() missing 1 required positional argument: 'password'" in str(e.value)
+
+
+def test_rest_catalog_with_auth_impl() -> None:
+    # Given
+    catalog_properties = {
+        "uri": TEST_URI,
+        "auth": {
+            "type": "custom",
+            "impl": "dummy.nonexistent.package",
+            "custom": {
+                "property1": "one",
+                "property2": "two",
+            },
+        },
+    }
+    with pytest.raises(ValueError) as e:
+        # Missing namespace
+        RestCatalog("rest", **catalog_properties)  # type: ignore
+    assert "Could not load AuthManager class for 'dummy.nonexistent.package'" in str(e.value)
+
+
 EXAMPLE_ENV = {"PYICEBERG_CATALOG__PRODUCTION__URI": TEST_URI}