Skip to content

Commit c868ea4

Browse files
dolfinusdolfinus
committed
[DOP-30631] Add support for Iceberg REST Catalog + S3 delegated access
1 parent 6be7f8d commit c868ea4

File tree

28 files changed

+812
-129
lines changed

28 files changed

+812
-129
lines changed

‎docs/changelog/next_release/297.feature.rst‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
1+
Added support for fetching S3 params & credentials from Iceberg REST Catalog (delegation).

‎pyproject.toml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -263,7 +263,7 @@ max-annotation-complexity = 4
263263
max-returns = 5
264264
max-awaits = 5
265265
max-local-variables = 20
266-
max-name-length = 60
266+
max-name-length = 65
267267
# Max of expressions in a function
268268
max-expressions = 15
269269
# Max args in a function

‎syncmaster/db/models/connection.py‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,7 @@
1818
class ConnectionType(StrEnum):
1919
POSTGRES = "postgres"
2020
HIVE = "hive"
21-
ICEBERG_REST_S3 = "iceberg_rest_s3"
21+
ICEBERG = "iceberg"
2222
ORACLE = "oracle"
2323
CLICKHOUSE = "clickhouse"
2424
MSSQL = "mssql"
@@ -62,7 +62,7 @@ class Connection(Base, ResourceMixin, TimestampMixin):
6262
'simple',
6363
translate(coalesce(data->>'host', ''), './-_:\\', ' ')
6464
)
65-
""",
65+
""", # noqa: WPS342
6666
persisted=True,
6767
),
6868
nullable=False,

‎syncmaster/dto/connections.py‎

Lines changed: 55 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -74,8 +74,14 @@ class HiveConnectionDTO(ConnectionDTO):
7474

7575

7676
@dataclass
77-
class IcebergRESTCatalogS3ConnectionBaseDTO(ConnectionDTO):
77+
class IcebergConnectionBaseDTO(ConnectionDTO):
7878
rest_catalog_url: str
79+
flavor: ClassVar[str]
80+
type: ClassVar[str] = "iceberg"
81+
82+
83+
@dataclass
84+
class IcebergRESTCatalogS3DirectConnectionBaseDTO(IcebergConnectionBaseDTO):
7985
s3_warehouse_path: str
8086
s3_host: str
8187
s3_bucket: str
@@ -86,18 +92,18 @@ class IcebergRESTCatalogS3ConnectionBaseDTO(ConnectionDTO):
8692
s3_port: int | None
8793
s3_protocol: str
8894
s3_additional_params: dict
89-
type: ClassVar[str] = "iceberg_rest_s3"
95+
implementation: ClassVar[str] = "iceberg_rest_s3_direct"
9096

9197

9298
@dataclass(kw_only=True)
93-
class IcebergRESTCatalogBasicAuthS3DTO(IcebergRESTCatalogS3ConnectionBaseDTO):
99+
class IcebergRESTCatalogBasicAuthS3BasicDTO(IcebergRESTCatalogS3DirectConnectionBaseDTO):
94100
rest_catalog_username: str
95101
rest_catalog_password: str
96102
rest_catalog_auth_type: Literal["basic"] = "basic"
97103

98104

99105
@dataclass(kw_only=True)
100-
class IcebergRESTCatalogOAuth2ClientCredentialsS3DTO(IcebergRESTCatalogS3ConnectionBaseDTO):
106+
class IcebergRESTCatalogOAuth2ClientCredentialsS3BasicDTO(IcebergRESTCatalogS3DirectConnectionBaseDTO):
101107
rest_catalog_oauth2_client_id: str
102108
rest_catalog_oauth2_client_secret: str
103109
rest_catalog_oauth2_scopes: list[str]
@@ -107,13 +113,52 @@ class IcebergRESTCatalogOAuth2ClientCredentialsS3DTO(IcebergRESTCatalogS3Connect
107113
rest_catalog_auth_type: Literal["oauth2"] = "oauth2"
108114

109115

110-
# TODO: should be refactored
111-
class IcebergRESTCatalogS3ConnectionDTO:
112-
def __new__(cls, **data):
113-
if "rest_catalog_oauth2_client_id" in data:
114-
return IcebergRESTCatalogOAuth2ClientCredentialsS3DTO(**data)
116+
@dataclass
117+
class IcebergRESTCatalogS3DelegatedConnectionBaseDTO(IcebergConnectionBaseDTO):
118+
s3_warehouse_name: str | None = None
119+
s3_access_delegation: Literal["vended-credentials", "remote-signing"] = "vended-credentials"
120+
implementation: ClassVar[str] = "iceberg_rest_s3_delegated"
121+
122+
123+
@dataclass(kw_only=True)
124+
class IcebergRESTCatalogBasicAuthS3DelegatedDTO(IcebergRESTCatalogS3DelegatedConnectionBaseDTO):
125+
rest_catalog_username: str
126+
rest_catalog_password: str
127+
rest_catalog_auth_type: Literal["basic"] = "basic"
128+
115129

116-
return IcebergRESTCatalogBasicAuthS3DTO(**data)
130+
@dataclass(kw_only=True)
131+
class IcebergRESTCatalogOAuth2ClientCredentialsS3DelegatedDTO(IcebergRESTCatalogS3DelegatedConnectionBaseDTO):
132+
rest_catalog_oauth2_client_id: str
133+
rest_catalog_oauth2_client_secret: str
134+
rest_catalog_oauth2_scopes: list[str]
135+
rest_catalog_oauth2_resource: str | None = None
136+
rest_catalog_oauth2_audience: str | None = None
137+
rest_catalog_oauth2_token_endpoint: str | None = None
138+
rest_catalog_auth_type: Literal["oauth2"] = "oauth2"
139+
140+
141+
# TODO: should be refactored
142+
def get_iceberg_rest_catalog_s3_direct_connection_dto(
143+
**data,
144+
) -> IcebergRESTCatalogBasicAuthS3BasicDTO | IcebergRESTCatalogOAuth2ClientCredentialsS3BasicDTO:
145+
if "rest_catalog_oauth2_client_id" in data:
146+
return IcebergRESTCatalogOAuth2ClientCredentialsS3BasicDTO(**data)
147+
return IcebergRESTCatalogBasicAuthS3BasicDTO(**data)
148+
149+
150+
def get_iceberg_rest_catalog_s3_delegated_connection_dto(
151+
**data,
152+
) -> IcebergRESTCatalogBasicAuthS3DelegatedDTO | IcebergRESTCatalogOAuth2ClientCredentialsS3DelegatedDTO:
153+
if "rest_catalog_oauth2_client_id" in data:
154+
return IcebergRESTCatalogOAuth2ClientCredentialsS3DelegatedDTO(**data)
155+
return IcebergRESTCatalogBasicAuthS3DelegatedDTO(**data)
156+
157+
158+
def get_iceberg_connection_dto(**data) -> IcebergConnectionBaseDTO:
159+
if "s3_warehouse_path" in data:
160+
return get_iceberg_rest_catalog_s3_direct_connection_dto(**data)
161+
return get_iceberg_rest_catalog_s3_delegated_connection_dto(**data)
117162

118163

119164
@dataclass

‎syncmaster/dto/transfers.py‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -128,9 +128,9 @@ def __post_init__(self):
128128

129129

130130
@dataclass
131-
class IcebergRESTCatalogS3TransferDTO(DBTransferDTO):
132-
type: ClassVar[str] = "iceberg_rest_s3"
133-
catalog_name: str = field(default_factory=lambda: f"iceberg_rest_s3_{uuid4().hex[:8]}") # noqa: WPS237
131+
class IcebergTransferDTO(DBTransferDTO):
132+
type: ClassVar[str] = "iceberg"
133+
catalog_name: str = field(default_factory=lambda: f"iceberg_{uuid4().hex[:8]}") # noqa: WPS237
134134

135135
def __post_init__(self):
136136
super().__post_init__()

syncmaster/schemas/v1/auth/iceberg/__init__.py renamed to syncmaster/schemas/v1/auth/iceberg_rest_s3_delegated/__init__.py

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -4,28 +4,28 @@
44

55
from pydantic import Field
66

7-
from syncmaster.schemas.v1.auth.iceberg.basic import (
7+
from syncmaster.schemas.v1.auth.iceberg_rest_s3_delegated.basic import (
88
CreateIcebergRESTCatalogBasicAuthSchema,
99
ReadIcebergRESTCatalogBasicAuthSchema,
1010
UpdateIcebergRESTCatalogBasicAuthSchema,
1111
)
12-
from syncmaster.schemas.v1.auth.iceberg.oauth2_client_credentials import (
12+
from syncmaster.schemas.v1.auth.iceberg_rest_s3_delegated.oauth2_client_credentials import (
1313
CreateIcebergRESTCatalogOAuth2ClientCredentialsAuthSchema,
1414
ReadIcebergRESTCatalogOAuth2ClientCredentialsAuthSchema,
1515
UpdateIcebergRESTCatalogOAuth2ClientCredentialsAuthSchema,
1616
)
1717

18-
CreateIcebergRESTCatalogS3ConnectionAuthDataSchema = Annotated[
18+
CreateIcebergRESTCatalogS3DelegatedConnectionAuthDataSchema = Annotated[
1919
CreateIcebergRESTCatalogBasicAuthSchema | CreateIcebergRESTCatalogOAuth2ClientCredentialsAuthSchema,
2020
Field(discriminator="type"),
2121
]
2222

23-
ReadIcebergRESTCatalogS3ConnectionAuthDataSchema = Annotated[
23+
ReadIcebergRESTCatalogS3DelegatedConnectionAuthDataSchema = Annotated[
2424
ReadIcebergRESTCatalogBasicAuthSchema | ReadIcebergRESTCatalogOAuth2ClientCredentialsAuthSchema,
2525
Field(discriminator="type"),
2626
]
2727

28-
UpdateIcebergRESTCatalogS3ConnectionAuthDataSchema = Annotated[
28+
UpdateIcebergRESTCatalogS3DelegatedConnectionAuthDataSchema = Annotated[
2929
UpdateIcebergRESTCatalogBasicAuthSchema | UpdateIcebergRESTCatalogOAuth2ClientCredentialsAuthSchema,
3030
Field(discriminator="type"),
3131
]

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

Lines changed: 2 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -6,19 +6,16 @@
66

77

88
class ReadIcebergRESTCatalogBasicAuthSchema(BaseModel):
9-
type: Literal["iceberg_rest_basic_s3_basic"] = Field(description="Auth type")
9+
type: Literal["iceberg_rest_basic"] = Field(description="Auth type")
1010
rest_catalog_username: str
11-
s3_access_key: str
1211

1312

1413
class CreateIcebergRESTCatalogBasicAuthSchema(ReadIcebergRESTCatalogBasicAuthSchema):
1514
rest_catalog_password: SecretStr
16-
s3_secret_key: SecretStr
1715

1816

1917
class UpdateIcebergRESTCatalogBasicAuthSchema(ReadIcebergRESTCatalogBasicAuthSchema):
2018
rest_catalog_password: SecretStr | None = None
21-
s3_secret_key: SecretStr | None = None
2219

2320
def get_secret_fields(self) -> tuple[str, ...]:
24-
return ("rest_catalog_password", "s3_secret_key")
21+
return ("rest_catalog_password",)

syncmaster/schemas/v1/auth/iceberg/oauth2_client_credentials.py renamed to syncmaster/schemas/v1/auth/iceberg_rest_s3_delegated/oauth2_client_credentials.py

Lines changed: 2 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -8,27 +8,24 @@
88

99

1010
class ReadIcebergRESTCatalogOAuth2ClientCredentialsAuthSchema(BaseModel):
11-
type: Literal["iceberg_rest_oauth2_client_credentials_s3_basic"] = Field(description="Auth type")
11+
type: Literal["iceberg_rest_oauth2_client_credentials"] = Field(description="Auth type")
1212
rest_catalog_oauth2_client_id: str
1313
rest_catalog_oauth2_scopes: list[str] = Field(default_factory=list)
1414
rest_catalog_oauth2_resource: str | None = None
1515
rest_catalog_oauth2_audience: str | None = None
1616
rest_catalog_oauth2_token_endpoint: URL | None = None
17-
s3_access_key: str
1817

1918

2019
class CreateIcebergRESTCatalogOAuth2ClientCredentialsAuthSchema(
2120
ReadIcebergRESTCatalogOAuth2ClientCredentialsAuthSchema,
2221
):
2322
rest_catalog_oauth2_client_secret: SecretStr
24-
s3_secret_key: SecretStr
2523

2624

2725
class UpdateIcebergRESTCatalogOAuth2ClientCredentialsAuthSchema(
2826
ReadIcebergRESTCatalogOAuth2ClientCredentialsAuthSchema,
2927
):
3028
rest_catalog_oauth2_client_secret: SecretStr | None = None
31-
s3_secret_key: SecretStr | None = None
3229

3330
def get_secret_fields(self) -> tuple[str, ...]:
34-
return ("rest_catalog_oauth2_client_secret", "s3_secret_key")
31+
return ("rest_catalog_oauth2_client_secret",)

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

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,31 @@
1+
# SPDX-FileCopyrightText: 2023-2024 MTS PJSC
2+
# SPDX-License-Identifier: Apache-2.0
3+
from typing import Annotated
4+
5+
from pydantic import Field
6+
7+
from syncmaster.schemas.v1.auth.iceberg_rest_s3_direct.basic import (
8+
CreateIcebergRESTCatalogBasicS3BasicAuthSchema,
9+
ReadIcebergRESTCatalogBasicS3BasicAuthSchema,
10+
UpdateIcebergRESTCatalogBasicS3BasicAuthSchema,
11+
)
12+
from syncmaster.schemas.v1.auth.iceberg_rest_s3_direct.oauth2_client_credentials import (
13+
CreateIcebergRESTCatalogOAuth2ClientCredentialsS3BasicAuthSchema,
14+
ReadIcebergRESTCatalogOAuth2ClientCredentialsS3BasicAuthSchema,
15+
UpdateIcebergRESTCatalogOAuth2ClientCredentialsS3BasicAuthSchema,
16+
)
17+
18+
CreateIcebergRESTCatalogS3DirectConnectionAuthDataSchema = Annotated[
19+
CreateIcebergRESTCatalogBasicS3BasicAuthSchema | CreateIcebergRESTCatalogOAuth2ClientCredentialsS3BasicAuthSchema,
20+
Field(discriminator="type"),
21+
]
22+
23+
ReadIcebergRESTCatalogS3DirectConnectionAuthDataSchema = Annotated[
24+
ReadIcebergRESTCatalogBasicS3BasicAuthSchema | ReadIcebergRESTCatalogOAuth2ClientCredentialsS3BasicAuthSchema,
25+
Field(discriminator="type"),
26+
]
27+
28+
UpdateIcebergRESTCatalogS3DirectConnectionAuthDataSchema = Annotated[
29+
UpdateIcebergRESTCatalogBasicS3BasicAuthSchema | UpdateIcebergRESTCatalogOAuth2ClientCredentialsS3BasicAuthSchema,
30+
Field(discriminator="type"),
31+
]

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

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,24 @@
1+
# SPDX-FileCopyrightText: 2023-2024 MTS PJSC
2+
# SPDX-License-Identifier: Apache-2.0
3+
from typing import Literal
4+
5+
from pydantic import BaseModel, Field, SecretStr
6+
7+
8+
class ReadIcebergRESTCatalogBasicS3BasicAuthSchema(BaseModel):
9+
type: Literal["iceberg_rest_basic_s3_basic"] = Field(description="Auth type")
10+
rest_catalog_username: str
11+
s3_access_key: str
12+
13+
14+
class CreateIcebergRESTCatalogBasicS3BasicAuthSchema(ReadIcebergRESTCatalogBasicS3BasicAuthSchema):
15+
rest_catalog_password: SecretStr
16+
s3_secret_key: SecretStr
17+
18+
19+
class UpdateIcebergRESTCatalogBasicS3BasicAuthSchema(ReadIcebergRESTCatalogBasicS3BasicAuthSchema):
20+
rest_catalog_password: SecretStr | None = None
21+
s3_secret_key: SecretStr | None = None
22+
23+
def get_secret_fields(self) -> tuple[str, ...]:
24+
return ("rest_catalog_password", "s3_secret_key")

0 commit comments

Comments
 (0)