[DOP-30631] Add support for Iceberg REST Catalog + S3 delegated access

Author: dolfinus

docs/changelog/next_release/297.feature.rst

@@ -0,0 +1 @@
+Added support for fetching S3 params & credentials from Iceberg REST Catalog (delegation).

poetry.lock

@@ -263,7 +263,7 @@ max-annotation-complexity = 4
 max-returns = 5
 max-awaits = 5
 max-local-variables = 20
-max-name-length = 60
+max-name-length = 65
 # Max of expressions in a function
 max-expressions = 15
 # Max args in a function
@@ -281,7 +281,7 @@ max-cognitive-score = 20
 # Max amount of cognitive complexity per module
 max-cognitive-average = 25
 max-imports = 25
-max-imported-names = 50
+max-imported-names = 55
 # Max of expression usages in a module
 max-module-expressions = 15
 # Max of expression usages in a function

pyproject.toml

@@ -18,7 +18,8 @@
 class ConnectionType(StrEnum):
     POSTGRES = "postgres"
     HIVE = "hive"
-    ICEBERG_REST_S3 = "iceberg_rest_s3"
+    ICEBERG_REST_S3_DIRECT = "iceberg_rest_s3_direct"
+    ICEBERG_REST_S3_DELEGATED = "iceberg_rest_s3_delegated"
     ORACLE = "oracle"
     CLICKHOUSE = "clickhouse"
     MSSQL = "mssql"
@@ -62,7 +63,7 @@ class Connection(Base, ResourceMixin, TimestampMixin):
                 'simple',
                 translate(coalesce(data->>'host', ''), './-_:\\', '      ')
             )
-            """,
+            """,  # noqa: WPS342
             persisted=True,
         ),
         nullable=False,

syncmaster/db/models/connection.py

@@ -74,7 +74,7 @@ class HiveConnectionDTO(ConnectionDTO):
 
 
 @dataclass
-class IcebergRESTCatalogS3ConnectionBaseDTO(ConnectionDTO):
+class IcebergRESTCatalogS3DirectConnectionBaseDTO(ConnectionDTO):
     metastore_url: str
     s3_warehouse_path: str
     s3_host: str
@@ -86,18 +86,18 @@ class IcebergRESTCatalogS3ConnectionBaseDTO(ConnectionDTO):
     s3_port: int | None
     s3_protocol: str
     s3_additional_params: dict
-    type: ClassVar[str] = "iceberg_rest_s3"
+    type: ClassVar[str] = "iceberg_rest_s3_direct"
 
 
 @dataclass(kw_only=True)
-class IcebergRESTCatalogBasicAuthS3DTO(IcebergRESTCatalogS3ConnectionBaseDTO):
+class IcebergRESTCatalogBasicAuthS3BasicDTO(IcebergRESTCatalogS3DirectConnectionBaseDTO):
     metastore_username: str
     metastore_password: str
     metastore_auth_type: Literal["basic"] = "basic"
 
 
 @dataclass(kw_only=True)
-class IcebergRESTCatalogOAuth2ClientCredentialsS3DTO(IcebergRESTCatalogS3ConnectionBaseDTO):
+class IcebergRESTCatalogOAuth2ClientCredentialsS3BasicDTO(IcebergRESTCatalogS3DirectConnectionBaseDTO):
     metastore_oauth2_client_id: str
     metastore_oauth2_client_secret: str
     metastore_oauth2_scopes: list[str]
@@ -107,13 +107,54 @@ class IcebergRESTCatalogOAuth2ClientCredentialsS3DTO(IcebergRESTCatalogS3Connect
     metastore_auth_type: Literal["oauth2"] = "oauth2"
 
 
+IcebergRESTCatalogS3DirectConnectionDTO = (
+    IcebergRESTCatalogBasicAuthS3BasicDTO | IcebergRESTCatalogOAuth2ClientCredentialsS3BasicDTO
+)
+
+
 # TODO: should be refactored
-class IcebergRESTCatalogS3ConnectionDTO:
-    def __new__(cls, **data):
-        if "metastore_oauth2_client_id" in data:
-            return IcebergRESTCatalogOAuth2ClientCredentialsS3DTO(**data)
+def get_iceberg_rest_catalog_s3_direct_connection_dto(**data) -> IcebergRESTCatalogS3DirectConnectionDTO:
+    if "metastore_oauth2_client_id" in data:
+        return IcebergRESTCatalogOAuth2ClientCredentialsS3BasicDTO(**data)
+    return IcebergRESTCatalogBasicAuthS3BasicDTO(**data)
+
+
+@dataclass
+class IcebergRESTCatalogS3DelegatedConnectionBaseDTO(ConnectionDTO):
+    metastore_url: str
+    s3_warehouse_name: str | None = None
+    s3_access_delegation: Literal["vended-credentials", "remote-signing"] = "vended-credentials"
+    type: ClassVar[str] = "iceberg_rest_s3_delegated"
+
+
+@dataclass(kw_only=True)
+class IcebergRESTCatalogBasicAuthS3DelegatedDTO(IcebergRESTCatalogS3DelegatedConnectionBaseDTO):
+    metastore_username: str
+    metastore_password: str
+    metastore_auth_type: Literal["basic"] = "basic"
+
 
-        return IcebergRESTCatalogBasicAuthS3DTO(**data)
+@dataclass(kw_only=True)
+class IcebergRESTCatalogOAuth2ClientCredentialsS3DelegatedDTO(IcebergRESTCatalogS3DelegatedConnectionBaseDTO):
+    metastore_oauth2_client_id: str
+    metastore_oauth2_client_secret: str
+    metastore_oauth2_scopes: list[str]
+    metastore_oauth2_resource: str | None = None
+    metastore_oauth2_audience: str | None = None
+    metastore_oauth2_server_uri: str | None = None
+    metastore_auth_type: Literal["oauth2"] = "oauth2"
+
+
+IcebergRESTCatalogS3DelegatedConnectionDTO = (
+    IcebergRESTCatalogBasicAuthS3DelegatedDTO | IcebergRESTCatalogOAuth2ClientCredentialsS3DelegatedDTO
+)
+
+
+# TODO: should be refactored
+def get_iceberg_rest_catalog_s3_delegated_connection_dto(**data) -> IcebergRESTCatalogS3DelegatedConnectionDTO:
+    if "metastore_oauth2_client_id" in data:
+        return IcebergRESTCatalogOAuth2ClientCredentialsS3DelegatedDTO(**data)
+    return IcebergRESTCatalogBasicAuthS3DelegatedDTO(**data)
 
 
 @dataclass

syncmaster/dto/connections.py

@@ -128,9 +128,19 @@ def __post_init__(self):
 
 
 @dataclass
-class IcebergRESTCatalogS3TransferDTO(DBTransferDTO):
-    type: ClassVar[str] = "iceberg_rest_s3"
-    catalog_name: str = field(default_factory=lambda: f"iceberg_rest_s3_{uuid4().hex[:8]}")  # noqa: WPS237
+class IcebergRESTCatalogS3DelegatedTransferDTO(DBTransferDTO):
+    type: ClassVar[str] = "iceberg_rest_s3_delegated"
+    catalog_name: str = field(default_factory=lambda: f"iceberg_rest_s3_delegated_{uuid4().hex[:8]}")  # noqa: WPS237
+
+    def __post_init__(self):
+        super().__post_init__()
+        self.options.setdefault("if_exists", "replace_overlapping_partitions")
+
+
+@dataclass
+class IcebergRESTCatalogS3DirectTransferDTO(DBTransferDTO):
+    type: ClassVar[str] = "iceberg_rest_s3_direct"
+    catalog_name: str = field(default_factory=lambda: f"iceberg_rest_s3_direct_{uuid4().hex[:8]}")  # noqa: WPS237
 
     def __post_init__(self):
         super().__post_init__()

syncmaster/dto/transfers.py

@@ -4,28 +4,28 @@
 
 from pydantic import Field
 
-from syncmaster.schemas.v1.auth.iceberg.basic import (
+from syncmaster.schemas.v1.auth.iceberg_rest_s3_delegated.basic import (
     CreateIcebergRESTCatalogBasicAuthSchema,
     ReadIcebergRESTCatalogBasicAuthSchema,
     UpdateIcebergRESTCatalogBasicAuthSchema,
 )
-from syncmaster.schemas.v1.auth.iceberg.oauth2_client_credentials import (
+from syncmaster.schemas.v1.auth.iceberg_rest_s3_delegated.oauth2_client_credentials import (
     CreateIcebergRESTCatalogOAuth2ClientCredentialsAuthSchema,
     ReadIcebergRESTCatalogOAuth2ClientCredentialsAuthSchema,
     UpdateIcebergRESTCatalogOAuth2ClientCredentialsAuthSchema,
 )
 
-CreateIcebergRESTCatalogS3ConnectionAuthDataSchema = Annotated[
+CreateIcebergRESTCatalogS3DelegatedConnectionAuthDataSchema = Annotated[
     CreateIcebergRESTCatalogBasicAuthSchema | CreateIcebergRESTCatalogOAuth2ClientCredentialsAuthSchema,
     Field(discriminator="type"),
 ]
 
-ReadIcebergRESTCatalogS3ConnectionAuthDataSchema = Annotated[
+ReadIcebergRESTCatalogS3DelegatedConnectionAuthDataSchema = Annotated[
     ReadIcebergRESTCatalogBasicAuthSchema | ReadIcebergRESTCatalogOAuth2ClientCredentialsAuthSchema,
     Field(discriminator="type"),
 ]
 
-UpdateIcebergRESTCatalogS3ConnectionAuthDataSchema = Annotated[
+UpdateIcebergRESTCatalogS3DelegatedConnectionAuthDataSchema = Annotated[
     UpdateIcebergRESTCatalogBasicAuthSchema | UpdateIcebergRESTCatalogOAuth2ClientCredentialsAuthSchema,
     Field(discriminator="type"),
 ]

syncmaster/schemas/v1/auth/iceberg/__init__.py

@@ -0,0 +1,21 @@
+# SPDX-FileCopyrightText: 2023-2024 MTS PJSC
+# SPDX-License-Identifier: Apache-2.0
+from typing import Literal
+
+from pydantic import BaseModel, SecretStr
+
+
+class ReadIcebergRESTCatalogBasicAuthSchema(BaseModel):
+    type: Literal["iceberg_rest_basic"]
+    metastore_username: str
+
+
+class CreateIcebergRESTCatalogBasicAuthSchema(ReadIcebergRESTCatalogBasicAuthSchema):
+    metastore_password: SecretStr
+
+
+class UpdateIcebergRESTCatalogBasicAuthSchema(ReadIcebergRESTCatalogBasicAuthSchema):
+    metastore_password: SecretStr | None = None
+
+    def get_secret_fields(self) -> tuple[str, ...]:
+        return ("metastore_password",)

…ncmaster/schemas/v1/auth/iceberg/auth.py …th/iceberg_rest_s3_delegated/__init__.pysyncmaster/schemas/v1/auth/iceberg/auth.py renamed to syncmaster/schemas/v1/auth/iceberg_rest_s3_delegated/__init__.py syncmaster/schemas/v1/auth/iceberg/auth.py renamed to syncmaster/schemas/v1/auth/iceberg_rest_s3_delegated/__init__.py

@@ -0,0 +1,29 @@
+# SPDX-FileCopyrightText: 2023-2024 MTS PJSC
+# SPDX-License-Identifier: Apache-2.0
+from typing import Literal
+
+from pydantic import BaseModel, Field, SecretStr
+
+
+class ReadIcebergRESTCatalogOAuth2ClientCredentialsAuthSchema(BaseModel):
+    type: Literal["iceberg_rest_oauth2_client_credentials"]
+    metastore_oauth2_client_id: str
+    metastore_oauth2_scopes: list[str] = Field(default_factory=list)
+    metastore_oauth2_resource: str | None = None
+    metastore_oauth2_audience: str | None = None
+    metastore_oauth2_server_uri: str | None = None
+
+
+class CreateIcebergRESTCatalogOAuth2ClientCredentialsAuthSchema(
+    ReadIcebergRESTCatalogOAuth2ClientCredentialsAuthSchema,
+):
+    metastore_oauth2_client_secret: SecretStr
+
+
+class UpdateIcebergRESTCatalogOAuth2ClientCredentialsAuthSchema(
+    ReadIcebergRESTCatalogOAuth2ClientCredentialsAuthSchema,
+):
+    metastore_oauth2_client_secret: SecretStr | None = None
+
+    def get_secret_fields(self) -> tuple[str, ...]:
+        return ("metastore_oauth2_client_secret",)