Skip to content

Commit 0a56faf

Browse files
authored
fix(warehouses): Add write_dispositions to each resource (#180)
### Summary Fix problems arising from removal of cli parameter in #177. Set write_dispositions explicitly on each resource. <!-- This is an auto-generated comment: release notes by coderabbit.ai --> ## Summary by CodeRabbit * **Documentation** * Removed parameter documentation from CLI docstring. * **Refactor** * Migrated write disposition configuration from global defaults to resource-level settings across data source connectors, enabling granular control over how data is written to targets (replace vs. merge operations). <sub>✏️ Tip: You can customize this high-level summary in your review settings.</sub> <!-- end of auto-generated comment: release notes by coderabbit.ai -->
1 parent cc89dcc commit 0a56faf

4 files changed

Lines changed: 17 additions & 8 deletions

File tree

elt-common/src/elt_common/cli.py

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -78,7 +78,6 @@ def cli_main(
7878
:param data_generator: Callable returning a dlt.DltSource or dlt.DltResource
7979
:param dataset_name_suffix: Suffix part of full dataset name in the destination. The given string is prefixed with
8080
a standard string defined in constants.DATASET_NAME_PREFIX_SRCS
81-
:param default_write_disposition: Default mode for dlt write_disposition defaults to "append"
8281
:param default_destination: Default destination, defaults to "filesystem"
8382
:param default_loader_file_format: Default dlt loader file format, defaults to "parquet"
8483
:param default_progress: Default progress reporter, defaults to NULL_COLLECTOR

warehouses/accelerator/extract_load/accelerator_sharepoint/extract_and_load.py

Lines changed: 11 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -55,7 +55,8 @@ def equipment_downtime_records_archive() -> DltResource:
5555
columns={
5656
"FaultDate": {"data_type": "text"},
5757
"FaultTime": {"data_type": "text"},
58-
}
58+
},
59+
write_disposition="replace",
5960
)
6061
)
6162
return reader
@@ -69,8 +70,15 @@ def edr_equipment_mapping() -> DltResource:
6970
extract_content=True,
7071
)
7172
return (
72-
files | read_excel(header=None, names=["equipment_name", "equipment_category"])
73-
).with_name("edr_equipment_mapping")
73+
(
74+
files
75+
| read_excel(header=None, names=["equipment_name", "equipment_category"])
76+
)
77+
.with_name("edr_equipment_mapping")
78+
.apply_hints(
79+
write_disposition="replace",
80+
)
81+
)
7482

7583

7684
# We need to set the section in order for dlt to find our named secret.
@@ -87,5 +95,4 @@ def resources():
8795
default_destination="elt_common.dlt_destinations.pyiceberg",
8896
data_generator=resources(),
8997
dataset_name_suffix="accelerator_sharepoint",
90-
default_write_disposition="replace",
9198
)

warehouses/accelerator/extract_load/electricity_sharepoint/extract_and_load.py

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -155,7 +155,9 @@ def rdm_data(
155155
- pendulum.Duration(seconds=ONE_DAY_SECS),
156156
)
157157
reader = files | extract_content_and_read()
158-
yield from reader
158+
yield from reader.apply_hints(
159+
write_disposition="merge",
160+
)
159161

160162

161163
cli_main(
@@ -165,5 +167,4 @@ def rdm_data(
165167
rdm_data, partition=PartitionTrBuilder.year("date_time")
166168
),
167169
dataset_name_suffix=PIPELINE_NAME,
168-
default_write_disposition="merge",
169170
)

warehouses/accelerator/extract_load/statusdisplay/extract_and_load.py

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,9 @@ def statusdisplay() -> DltSource:
2525
"base_url": dlt.config["sources.base_url"],
2626
},
2727
"resources": dlt.config["sources.resources"],
28+
"resource_defaults": {
29+
"write_disposition": "replace",
30+
},
2831
},
2932
)
3033

@@ -36,5 +39,4 @@ def statusdisplay() -> DltSource:
3639
default_destination="elt_common.dlt_destinations.pyiceberg",
3740
data_generator=statusdisplay(),
3841
dataset_name_suffix="statusdisplay",
39-
default_write_disposition="replace",
4042
)

0 commit comments

Comments
 (0)