|
1 | 1 | from copy import copy, deepcopy
|
2 |
| -from typing import List, TypedDict, cast, Any |
| 2 | +from typing import List, TypedDict, cast, Any, Optional, Dict |
3 | 3 |
|
4 | 4 | from dlt.common.schema.utils import DEFAULT_WRITE_DISPOSITION, merge_columns, new_column, new_table
|
5 | 5 | from dlt.common.schema.typing import (
|
@@ -125,6 +125,7 @@ def apply_hints(
|
125 | 125 | merge_key: TTableHintTemplate[TColumnNames] = None,
|
126 | 126 | incremental: Incremental[Any] = None,
|
127 | 127 | schema_contract: TTableHintTemplate[TSchemaContract] = None,
|
| 128 | + additional_table_hints: Optional[Dict[str, TTableHintTemplate[Any]]] = None, |
128 | 129 | ) -> None:
|
129 | 130 | """Creates or modifies existing table schema by setting provided hints. Accepts both static and dynamic hints based on data.
|
130 | 131 |
|
@@ -208,6 +209,14 @@ def apply_hints(
|
208 | 209 | t["incremental"] = None
|
209 | 210 | else:
|
210 | 211 | t["incremental"] = incremental
|
| 212 | + if additional_table_hints is not None: |
| 213 | + # loop through provided hints and add, overwrite, or remove them |
| 214 | + for k, v in additional_table_hints.items(): |
| 215 | + if v: |
| 216 | + t[k] = v # type: ignore[literal-required] |
| 217 | + else: |
| 218 | + t.pop(k, None) # type: ignore[misc] |
| 219 | + |
211 | 220 | self.set_hints(t)
|
212 | 221 |
|
213 | 222 | def set_hints(self, hints_template: TResourceHints) -> None:
|
|
0 commit comments