[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).

pyproject.toml

@@ -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

syncmaster/db/models/connection.py

@@ -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/dto/connections.py

@@ -74,7 +74,7 @@ class HiveConnectionDTO(ConnectionDTO):
 
 
 @dataclass
-class IcebergRESTCatalogS3ConnectionBaseDTO(ConnectionDTO):
+class IcebergRESTCatalogS3DirectConnectionBaseDTO(ConnectionDTO):
     rest_catalog_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):
     rest_catalog_username: str
     rest_catalog_password: str
     rest_catalog_auth_type: Literal["basic"] = "basic"
 
 
 @dataclass(kw_only=True)
-class IcebergRESTCatalogOAuth2ClientCredentialsS3DTO(IcebergRESTCatalogS3ConnectionBaseDTO):
+class IcebergRESTCatalogOAuth2ClientCredentialsS3BasicDTO(IcebergRESTCatalogS3DirectConnectionBaseDTO):
     rest_catalog_oauth2_client_id: str
     rest_catalog_oauth2_client_secret: str
     rest_catalog_oauth2_scopes: list[str]
@@ -107,13 +107,54 @@ class IcebergRESTCatalogOAuth2ClientCredentialsS3DTO(IcebergRESTCatalogS3Connect
     rest_catalog_auth_type: Literal["oauth2"] = "oauth2"
 
 
+IcebergRESTCatalogS3DirectConnectionDTO = (
+    IcebergRESTCatalogBasicAuthS3BasicDTO | IcebergRESTCatalogOAuth2ClientCredentialsS3BasicDTO
+)
+
+
 # TODO: should be refactored
-class IcebergRESTCatalogS3ConnectionDTO:
-    def __new__(cls, **data):
-        if "rest_catalog_oauth2_client_id" in data:
-            return IcebergRESTCatalogOAuth2ClientCredentialsS3DTO(**data)
+def get_iceberg_rest_catalog_s3_direct_connection_dto(**data) -> IcebergRESTCatalogS3DirectConnectionDTO:
+    if "rest_catalog_oauth2_client_id" in data:
+        return IcebergRESTCatalogOAuth2ClientCredentialsS3BasicDTO(**data)
+    return IcebergRESTCatalogBasicAuthS3BasicDTO(**data)
+
+
+@dataclass
+class IcebergRESTCatalogS3DelegatedConnectionBaseDTO(ConnectionDTO):
+    rest_catalog_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):
+    rest_catalog_username: str
+    rest_catalog_password: str
+    rest_catalog_auth_type: Literal["basic"] = "basic"
+
 
-        return IcebergRESTCatalogBasicAuthS3DTO(**data)
+@dataclass(kw_only=True)
+class IcebergRESTCatalogOAuth2ClientCredentialsS3DelegatedDTO(IcebergRESTCatalogS3DelegatedConnectionBaseDTO):
+    rest_catalog_oauth2_client_id: str
+    rest_catalog_oauth2_client_secret: str
+    rest_catalog_oauth2_scopes: list[str]
+    rest_catalog_oauth2_resource: str | None = None
+    rest_catalog_oauth2_audience: str | None = None
+    rest_catalog_oauth2_token_endpoint: str | None = None
+    rest_catalog_auth_type: Literal["oauth2"] = "oauth2"
+
+
+IcebergRESTCatalogS3DelegatedConnectionDTO = (
+    IcebergRESTCatalogBasicAuthS3DelegatedDTO | IcebergRESTCatalogOAuth2ClientCredentialsS3DelegatedDTO
+)
+
+
+# TODO: should be refactored
+def get_iceberg_rest_catalog_s3_delegated_connection_dto(**data) -> IcebergRESTCatalogS3DelegatedConnectionDTO:
+    if "rest_catalog_oauth2_client_id" in data:
+        return IcebergRESTCatalogOAuth2ClientCredentialsS3DelegatedDTO(**data)
+    return IcebergRESTCatalogBasicAuthS3DelegatedDTO(**data)
 
 
 @dataclass

syncmaster/dto/transfers.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__()

…ster/schemas/v1/auth/iceberg/__init__.py …th/iceberg_rest_s3_delegated/__init__.pysyncmaster/schemas/v1/auth/iceberg/__init__.py renamed to syncmaster/schemas/v1/auth/iceberg_rest_s3_delegated/__init__.py syncmaster/schemas/v1/auth/iceberg/__init__.py renamed to syncmaster/schemas/v1/auth/iceberg_rest_s3_delegated/__init__.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"),
 ]

…cmaster/schemas/v1/auth/iceberg/basic.py …/auth/iceberg_rest_s3_delegated/basic.pysyncmaster/schemas/v1/auth/iceberg/basic.py renamed to syncmaster/schemas/v1/auth/iceberg_rest_s3_delegated/basic.py syncmaster/schemas/v1/auth/iceberg/basic.py renamed to syncmaster/schemas/v1/auth/iceberg_rest_s3_delegated/basic.py

@@ -6,19 +6,16 @@
 
 
 class ReadIcebergRESTCatalogBasicAuthSchema(BaseModel):
-    type: Literal["iceberg_rest_basic_s3_basic"] = Field(description="Auth type")
+    type: Literal["iceberg_rest_basic"] = Field(description="Auth type")
     rest_catalog_username: str
-    s3_access_key: str
 
 
 class CreateIcebergRESTCatalogBasicAuthSchema(ReadIcebergRESTCatalogBasicAuthSchema):
     rest_catalog_password: SecretStr
-    s3_secret_key: SecretStr
 
 
 class UpdateIcebergRESTCatalogBasicAuthSchema(ReadIcebergRESTCatalogBasicAuthSchema):
     rest_catalog_password: SecretStr | None = None
-    s3_secret_key: SecretStr | None = None
 
     def get_secret_fields(self) -> tuple[str, ...]:
-        return ("rest_catalog_password", "s3_secret_key")
+        return ("rest_catalog_password",)

…uth/iceberg/oauth2_client_credentials.py …3_delegated/oauth2_client_credentials.pysyncmaster/schemas/v1/auth/iceberg/oauth2_client_credentials.py renamed to syncmaster/schemas/v1/auth/iceberg_rest_s3_delegated/oauth2_client_credentials.py syncmaster/schemas/v1/auth/iceberg/oauth2_client_credentials.py renamed to syncmaster/schemas/v1/auth/iceberg_rest_s3_delegated/oauth2_client_credentials.py

@@ -8,27 +8,24 @@
 
 
 class ReadIcebergRESTCatalogOAuth2ClientCredentialsAuthSchema(BaseModel):
-    type: Literal["iceberg_rest_oauth2_client_credentials_s3_basic"] = Field(description="Auth type")
+    type: Literal["iceberg_rest_oauth2_client_credentials"] = Field(description="Auth type")
     rest_catalog_oauth2_client_id: str
     rest_catalog_oauth2_scopes: list[str] = Field(default_factory=list)
     rest_catalog_oauth2_resource: str | None = None
     rest_catalog_oauth2_audience: str | None = None
     rest_catalog_oauth2_token_endpoint: URL | None = None
-    s3_access_key: str
 
 
 class CreateIcebergRESTCatalogOAuth2ClientCredentialsAuthSchema(
     ReadIcebergRESTCatalogOAuth2ClientCredentialsAuthSchema,
 ):
     rest_catalog_oauth2_client_secret: SecretStr
-    s3_secret_key: SecretStr
 
 
 class UpdateIcebergRESTCatalogOAuth2ClientCredentialsAuthSchema(
     ReadIcebergRESTCatalogOAuth2ClientCredentialsAuthSchema,
 ):
     rest_catalog_oauth2_client_secret: SecretStr | None = None
-    s3_secret_key: SecretStr | None = None
 
     def get_secret_fields(self) -> tuple[str, ...]:
-        return ("rest_catalog_oauth2_client_secret", "s3_secret_key")
+        return ("rest_catalog_oauth2_client_secret",)

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

@@ -0,0 +1,31 @@
+# SPDX-FileCopyrightText: 2023-2024 MTS PJSC
+# SPDX-License-Identifier: Apache-2.0
+from typing import Annotated
+
+from pydantic import Field
+
+from syncmaster.schemas.v1.auth.iceberg_rest_s3_direct.basic import (
+    CreateIcebergRESTCatalogBasicS3BasicAuthSchema,
+    ReadIcebergRESTCatalogBasicS3BasicAuthSchema,
+    UpdateIcebergRESTCatalogBasicS3BasicAuthSchema,
+)
+from syncmaster.schemas.v1.auth.iceberg_rest_s3_direct.oauth2_client_credentials import (
+    CreateIcebergRESTCatalogOAuth2ClientCredentialsS3BasicAuthSchema,
+    ReadIcebergRESTCatalogOAuth2ClientCredentialsS3BasicAuthSchema,
+    UpdateIcebergRESTCatalogOAuth2ClientCredentialsS3BasicAuthSchema,
+)
+
+CreateIcebergRESTCatalogS3DirectConnectionAuthDataSchema = Annotated[
+    CreateIcebergRESTCatalogBasicS3BasicAuthSchema | CreateIcebergRESTCatalogOAuth2ClientCredentialsS3BasicAuthSchema,
+    Field(discriminator="type"),
+]
+
+ReadIcebergRESTCatalogS3DirectConnectionAuthDataSchema = Annotated[
+    ReadIcebergRESTCatalogBasicS3BasicAuthSchema | ReadIcebergRESTCatalogOAuth2ClientCredentialsS3BasicAuthSchema,
+    Field(discriminator="type"),
+]
+
+UpdateIcebergRESTCatalogS3DirectConnectionAuthDataSchema = Annotated[
+    UpdateIcebergRESTCatalogBasicS3BasicAuthSchema | UpdateIcebergRESTCatalogOAuth2ClientCredentialsS3BasicAuthSchema,
+    Field(discriminator="type"),
+]

syncmaster/schemas/v1/auth/iceberg_rest_s3_direct/basic.py

@@ -0,0 +1,24 @@
+# SPDX-FileCopyrightText: 2023-2024 MTS PJSC
+# SPDX-License-Identifier: Apache-2.0
+from typing import Literal
+
+from pydantic import BaseModel, Field, SecretStr
+
+
+class ReadIcebergRESTCatalogBasicS3BasicAuthSchema(BaseModel):
+    type: Literal["iceberg_rest_basic_s3_basic"] = Field(description="Auth type")
+    rest_catalog_username: str
+    s3_access_key: str
+
+
+class CreateIcebergRESTCatalogBasicS3BasicAuthSchema(ReadIcebergRESTCatalogBasicS3BasicAuthSchema):
+    rest_catalog_password: SecretStr
+    s3_secret_key: SecretStr
+
+
+class UpdateIcebergRESTCatalogBasicS3BasicAuthSchema(ReadIcebergRESTCatalogBasicS3BasicAuthSchema):
+    rest_catalog_password: SecretStr | None = None
+    s3_secret_key: SecretStr | None = None
+
+    def get_secret_fields(self) -> tuple[str, ...]:
+        return ("rest_catalog_password", "s3_secret_key")