Bases: IcebergWarehouse, FrozenModel
Delegate configuring Iceberg warehouse to Iceberg catalog. 
Used by some Iceberg catalog implementations like
Parameters
name : str, optional
Warehouse name/alias, if supported by specific Iceberg catalog
"vended-credentials" | "remote-signing"
Value of X-Iceberg-Access-Delegation header.
Examples
Source code in onetl/connection/db_connection/iceberg/warehouse/delegated.py
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82 | @support_hooks
class IcebergDelegatedWarehouse(IcebergWarehouse, FrozenModel):
"""Delegate configuring Iceberg warehouse to Iceberg catalog. [](/hooks/)
Used by some Iceberg catalog implementations like:
* [Lakekeeper](https://docs.lakekeeper.io/docs/latest/storage/#s3)
* [Polaris](https://polaris.apache.org/in-dev/unreleased/polaris-spark-client/)
* [Apache Gravitino](https://gravitino.apache.org/docs/1.0.0/security/credential-vending/)
* [Databricks Unity Catalog](https://docs.databricks.com/aws/en/external-access/iceberg#use-iceberg-tables-with-apache-spark)
!!! success "Added in 0.15.0"
Parameters
----------
name : str, optional
Warehouse name/alias, if supported by specific Iceberg catalog
access_delegation : "vended-credentials" | "remote-signing"
Value of [X-Iceberg-Access-Delegation](https://github.com/apache/iceberg/blob/apache-iceberg-1.10.0/open-api/rest-catalog-open-api.yaml#L1854) header.
extra : Dict[str, str], default: {}
Additional configuration parameters
Examples
--------
=== "S3 client with vended credentials"
```python
from onetl.connection import Iceberg
warehouse = Iceberg.DeletatedWarehouse(
name="my-warehouse",
access_delegation="vended-credentials",
# other params passed to S3 client (optional)
extra={"client.region": "us-east-1"},
)
```
=== "S3 client with remote signing"
```python
from onetl.connection import Iceberg
warehouse = Iceberg.DeletatedWarehouse(
name="my-warehouse",
access_delegation="remote-signing",
# other params passed to S3 client (optional)
extra={"client.region": "us-east-1"},
)
```
""" # noqa: E501
name: Optional[str] = None
access_delegation: Literal["vended-credentials", "remote-signing"]
extra: Dict[str, Any] = Field(default_factory=dict)
@slot
def get_config(self) -> dict[str, str]:
config = {
"warehouse": self.name,
"header.X-Iceberg-Access-Delegation": self.access_delegation,
**stringify(self.extra),
}
return {k: v for k, v in config.items() if v is not None}
|