Add write_dispositions to each resource

Fix problems arising from removal of cli parameter

Author: martyngigg

elt-common/src/elt_common/cli.py

@@ -78,7 +78,6 @@ def cli_main(
     :param data_generator: Callable returning a dlt.DltSource or dlt.DltResource
     :param dataset_name_suffix: Suffix part of full dataset name in the destination. The given string is prefixed with
                                 a standard string defined in constants.DATASET_NAME_PREFIX_SRCS
-    :param default_write_disposition: Default mode for dlt write_disposition defaults to "append"
     :param default_destination: Default destination, defaults to "filesystem"
     :param default_loader_file_format: Default dlt loader file format, defaults to "parquet"
     :param default_progress: Default progress reporter, defaults to NULL_COLLECTOR

warehouses/accelerator/extract_load/accelerator_sharepoint/extract_and_load.py

@@ -55,7 +55,8 @@ def equipment_downtime_records_archive() -> DltResource:
             columns={
                 "FaultDate": {"data_type": "text"},
                 "FaultTime": {"data_type": "text"},
-            }
+            },
+            write_disposition={"disposition": "replace"},
         )
     )
     return reader
@@ -69,8 +70,15 @@ def edr_equipment_mapping() -> DltResource:
         extract_content=True,
     )
     return (
-        files | read_excel(header=None, names=["equipment_name", "equipment_category"])
-    ).with_name("edr_equipment_mapping")
+        (
+            files
+            | read_excel(header=None, names=["equipment_name", "equipment_category"])
+        )
+        .with_name("edr_equipment_mapping")
+        .apply_hints(
+            write_disposition={"disposition": "replace"},
+        )
+    )
 
 
 # We need to set the section in order for dlt to find our named secret.
@@ -87,5 +95,4 @@ def resources():
         default_destination="elt_common.dlt_destinations.pyiceberg",
         data_generator=resources(),
         dataset_name_suffix="accelerator_sharepoint",
-        default_write_disposition="replace",
     )

warehouses/accelerator/extract_load/electricity_sharepoint/extract_and_load.py

@@ -155,7 +155,9 @@ def rdm_data(
         - pendulum.Duration(seconds=ONE_DAY_SECS),
     )
     reader = files | extract_content_and_read()
-    yield from reader
+    yield from reader.apply_hints(
+        write_disposition={"disposition": "merge"},
+    )
 
 
 cli_main(
@@ -165,5 +167,4 @@ def rdm_data(
         rdm_data, partition=PartitionTrBuilder.year("date_time")
     ),
     dataset_name_suffix=PIPELINE_NAME,
-    default_write_disposition="merge",
 )

warehouses/accelerator/extract_load/statusdisplay/extract_and_load.py

@@ -25,6 +25,9 @@ def statusdisplay() -> DltSource:
                 "base_url": dlt.config["sources.base_url"],
             },
             "resources": dlt.config["sources.resources"],
+            "resource_defaults": {
+                "write_disposition": "replace",
+            },
         },
     )
 
@@ -36,5 +39,4 @@ def statusdisplay() -> DltSource:
         default_destination="elt_common.dlt_destinations.pyiceberg",
         data_generator=statusdisplay(),
         dataset_name_suffix="statusdisplay",
-        default_write_disposition="replace",
     )