From b1f44628fdfcb06ed87a3ae32a99766c2e874eba Mon Sep 17 00:00:00 2001 From: Mariia Borodii Date: Tue, 28 Jul 2026 16:44:35 +0300 Subject: [PATCH 1/5] Prevent recipe wrangle from writing partial batches (#1096) --- tests/connectors/test_recipe.py | 36 ++++++++++++++++++++++++++++++++ wrangles/recipe_wrangles/main.py | 15 ++++++++++--- 2 files changed, 48 insertions(+), 3 deletions(-) diff --git a/tests/connectors/test_recipe.py b/tests/connectors/test_recipe.py index f7fc3352e..10799612b 100644 --- a/tests/connectors/test_recipe.py +++ b/tests/connectors/test_recipe.py @@ -137,6 +137,42 @@ def test_wrangles(): ) assert df["header1"][0] == "VALUE1" + +def test_recipe_wrangle_in_batch_does_not_write_partial_batches(): + """ + Test that a recipe used as a wrangle does not run its own write step + once per batch. The outer recipe should write the final combined result. + """ + memory.clear() + + wrangles.recipe.run( + """ + read: + - test: + rows: 1000 + values: + header1: value1 + wrangles: + - batch: + batch_size: 100 + wrangles: + - recipe: + wrangles: + - convert.case: + input: header1 + case: upper + write: + - memory: + id: partial_recipe_write + write: + - memory: + id: final_recipe_write + """ + ) + + assert "partial_recipe_write" not in memory.dataframes + assert len(memory.dataframes["final_recipe_write"]["data"]) == 1000 + def test_write(): """ Test write defined within the recipe diff --git a/wrangles/recipe_wrangles/main.py b/wrangles/recipe_wrangles/main.py index 83426351b..4ff896805 100644 --- a/wrangles/recipe_wrangles/main.py +++ b/wrangles/recipe_wrangles/main.py @@ -1417,7 +1417,7 @@ def recipe( input: _Union[str, int, list] = None, output: _Union[str, list] = None, name: str = None, - variables = {}, + variables = None, functions: _Union[_types.FunctionType, list] = [], **kwargs ) -> _pd.DataFrame: @@ -1437,6 +1437,8 @@ def recipe( type: object description: A dictionary of variables to pass to the recipe """ + if variables is None: + variables = {} if not name: name = kwargs df_temp = df.copy() # copy of the original df @@ -1449,18 +1451,25 @@ def recipe( if output is None and input is not None: output = input + recipe_object, functions = _recipe._load_recipe( + name, + variables=variables, + functions=functions + ) + recipe_object.pop('write', None) + # If output columns are specified, only apply to those if output: if not isinstance(output, list): output = [output] df[output] = _recipe.run( - name, + recipe_object, variables=variables, functions=functions, dataframe=df_temp )[output] else: df = _recipe.run( - name, + recipe_object, variables=variables, functions=functions, dataframe=df_temp From c527c7583d2d3c1036ddb2a1a3bb3ac3cadc9640 Mon Sep 17 00:00:00 2001 From: Mariia Borodii Date: Tue, 28 Jul 2026 17:47:03 +0300 Subject: [PATCH 2/5] Accumulate batched Excel sheet writes (#1097) --- tests/connectors/test_excel.py | 72 +++++++++++++++++++++++++++++++++- wrangles/connectors/excel.py | 19 +++++++++ 2 files changed, 90 insertions(+), 1 deletion(-) diff --git a/tests/connectors/test_excel.py b/tests/connectors/test_excel.py index 41fbc4581..2b011b37a 100644 --- a/tests/connectors/test_excel.py +++ b/tests/connectors/test_excel.py @@ -1,4 +1,5 @@ import wrangles +import pandas as pd from wrangles.connectors import memory @@ -30,4 +31,73 @@ def test_default_write(): assert ( data["columns"] == ["header1", "header2"] and len(data["data"]) == 5 - ) \ No newline at end of file + ) + + +def test_recipe_wrangle_in_batch_writes_all_rows_to_excel_sheet(): + """ + Test the WranglesXL output connector path when a recipe wrangle is used + inside a batch. The Excel sheet output should receive the full combined + dataframe, not only one batch. + """ + memory.clear() + wrangles.recipe.run( + """ + read: + - test: + rows: 1000 + values: + header1: value1 + wrangles: + - batch: + batch_size: 100 + wrangles: + - recipe: + wrangles: + - convert.case: + input: header1 + case: upper + write: + - excel.sheet: + name: Partial + write: + - excel.sheet: + name: Final + """ + ) + + excel_outputs = [ + v + for v in memory.dataframes.values() + if v.get("connector") == "excel.sheet.write" + ] + memory.clear() + + assert len(excel_outputs) == 1 + assert excel_outputs[0]["name"] == "Final" + assert len(excel_outputs[0]["data"]) == 1000 + + +def test_excel_sheet_append_accumulates_repeated_writes(): + """ + WranglesXL may receive repeated writes to the same sheet when work is + batched. Default append behavior should accumulate rows in one payload. + """ + memory.clear() + df = pd.DataFrame({"header1": ["value1"] * 1000}) + for start in range(0, 1000, 100): + wrangles.connectors.excel.sheet.write( + df.iloc[start:start + 100], + name="Results" + ) + + excel_outputs = [ + v + for v in memory.dataframes.values() + if v.get("connector") == "excel.sheet.write" + ] + memory.clear() + + assert len(excel_outputs) == 1 + assert excel_outputs[0]["name"] == "Results" + assert len(excel_outputs[0]["data"]) == 1000 diff --git a/wrangles/connectors/excel.py b/wrangles/connectors/excel.py index 9e7102b44..34579c977 100644 --- a/wrangles/connectors/excel.py +++ b/wrangles/connectors/excel.py @@ -11,6 +11,25 @@ class sheet(): def write(df: _pd.DataFrame, **kwargs): _logging.info(f": Saving data for Excel Sheet") + action = kwargs.get("action", "append") + name = kwargs.get("name") + cell = kwargs.get("cell") + + if action == "append": + for saved in reversed(list(_memory.dataframes.values())): + if ( + isinstance(saved, dict) + and saved.get("connector") == "excel.sheet.write" + and saved.get("name") == name + and saved.get("cell") == cell + and saved.get("action", "append") == "append" + and saved.get("columns") == df.columns.tolist() + ): + new_data = df.to_dict(orient="split") + saved["data"].extend(new_data["data"]) + saved["index"].extend(new_data["index"]) + return + _memory.write( df, connector = "excel.sheet.write", From 0a50bdbfae6e6db93a84d5d3fe9bc6a885b08f5c Mon Sep 17 00:00:00 2001 From: Mariia Borodii Date: Tue, 28 Jul 2026 18:53:25 +0300 Subject: [PATCH 3/5] Accumulate overwrite Excel sheet batch writes (#1098) --- tests/connectors/test_excel.py | 28 ++++++++++++++++++++++++++++ wrangles/connectors/excel.py | 6 ++++-- 2 files changed, 32 insertions(+), 2 deletions(-) diff --git a/tests/connectors/test_excel.py b/tests/connectors/test_excel.py index 2b011b37a..1b9255924 100644 --- a/tests/connectors/test_excel.py +++ b/tests/connectors/test_excel.py @@ -101,3 +101,31 @@ def test_excel_sheet_append_accumulates_repeated_writes(): assert len(excel_outputs) == 1 assert excel_outputs[0]["name"] == "Results" assert len(excel_outputs[0]["data"]) == 1000 + + +def test_excel_sheet_overwrite_accumulates_repeated_writes(): + """ + Batched WranglesXL runs may emit repeated overwrite writes to the same + sheet. The connector should still return one full payload so Excel replaces + the sheet with all rows, not just the final batch. + """ + memory.clear() + df = pd.DataFrame({"header1": ["value1"] * 1000}) + for start in range(0, 1000, 100): + wrangles.connectors.excel.sheet.write( + df.iloc[start:start + 100], + name="Results", + action="overwrite" + ) + + excel_outputs = [ + v + for v in memory.dataframes.values() + if v.get("connector") == "excel.sheet.write" + ] + memory.clear() + + assert len(excel_outputs) == 1 + assert excel_outputs[0]["name"] == "Results" + assert excel_outputs[0]["action"] == "overwrite" + assert len(excel_outputs[0]["data"]) == 1000 diff --git a/wrangles/connectors/excel.py b/wrangles/connectors/excel.py index 34579c977..36c906620 100644 --- a/wrangles/connectors/excel.py +++ b/wrangles/connectors/excel.py @@ -15,19 +15,21 @@ def write(df: _pd.DataFrame, **kwargs): name = kwargs.get("name") cell = kwargs.get("cell") - if action == "append": + if action in ("append", "overwrite"): for saved in reversed(list(_memory.dataframes.values())): if ( isinstance(saved, dict) and saved.get("connector") == "excel.sheet.write" and saved.get("name") == name and saved.get("cell") == cell - and saved.get("action", "append") == "append" + and saved.get("action", "append") in ("append", "overwrite") and saved.get("columns") == df.columns.tolist() ): new_data = df.to_dict(orient="split") saved["data"].extend(new_data["data"]) saved["index"].extend(new_data["index"]) + if action == "overwrite": + saved["action"] = "overwrite" return _memory.write( From 4a65794095ac238e80a9c42484d13d04b4d6c652 Mon Sep 17 00:00:00 2001 From: Mariia Borodii Date: Tue, 28 Jul 2026 19:28:47 +0300 Subject: [PATCH 4/5] Append Excel sheet output after first XL batch (#1099) --- tests/connectors/test_excel.py | 39 ++++++++++++++++++++++++++++++++++ wrangles/connectors/excel.py | 16 +++++++++++++- 2 files changed, 54 insertions(+), 1 deletion(-) diff --git a/tests/connectors/test_excel.py b/tests/connectors/test_excel.py index 1b9255924..cf7a2428e 100644 --- a/tests/connectors/test_excel.py +++ b/tests/connectors/test_excel.py @@ -129,3 +129,42 @@ def test_excel_sheet_overwrite_accumulates_repeated_writes(): assert excel_outputs[0]["name"] == "Results" assert excel_outputs[0]["action"] == "overwrite" assert len(excel_outputs[0]["data"]) == 1000 + + +def test_excel_sheet_overwrite_uses_append_after_first_external_batch(): + """ + WranglesXL can execute each batch as a separate Python run. In that case, + in-memory accumulation is not available, so later overwrite batches must be + returned as append actions. + """ + df = pd.DataFrame({"header1": ["value1"] * 100}) + + memory.clear() + wrangles.connectors.excel.sheet.write( + df, + name="Results", + action="overwrite", + variables={"batch_number": 1, "batch_total": 10} + ) + first_batch = [ + v + for v in memory.dataframes.values() + if v.get("connector") == "excel.sheet.write" + ][0] + + memory.clear() + wrangles.connectors.excel.sheet.write( + df, + name="Results", + action="overwrite", + variables={"batch_number": 2, "batch_total": 10} + ) + second_batch = [ + v + for v in memory.dataframes.values() + if v.get("connector") == "excel.sheet.write" + ][0] + memory.clear() + + assert first_batch["action"] == "overwrite" + assert second_batch["action"] == "append" diff --git a/wrangles/connectors/excel.py b/wrangles/connectors/excel.py index 36c906620..64a0ab9a6 100644 --- a/wrangles/connectors/excel.py +++ b/wrangles/connectors/excel.py @@ -8,10 +8,24 @@ class sheet(): _schema = {} - def write(df: _pd.DataFrame, **kwargs): + def write(df: _pd.DataFrame, variables: dict = None, **kwargs): _logging.info(f": Saving data for Excel Sheet") + if variables is None: + variables = {} + action = kwargs.get("action", "append") + try: + batch_number = int(variables.get("batch_number", 1)) + batch_total = int(variables.get("batch_total", 1)) + except (TypeError, ValueError): + batch_number = 1 + batch_total = 1 + + if action == "overwrite" and batch_total > 1 and batch_number > 1: + kwargs["action"] = "append" + action = "append" + name = kwargs.get("name") cell = kwargs.get("cell") From ea824f4d772bd87c913625c152e0cdb50ee593d5 Mon Sep 17 00:00:00 2001 From: Eric Hills <53243273+ebhills@users.noreply.github.com> Date: Tue, 28 Jul 2026 16:12:36 -0500 Subject: [PATCH 5/5] Align batched Excel writes by column name Handle repeated `excel.sheet.write` batches with changing or reordered columns by aligning rows on column names instead of position. This preserves accumulated Excel output for dynamic dictionary splits and keeps overwrite batches in the original column order, with tests covering both connector-level and recipe batch behavior. --- tests/connectors/test_excel.py | 69 +++++++++++++++++++++++++++++ tests/recipes/wrangles/test_main.py | 35 +++++++++++++++ wrangles/connectors/excel.py | 65 ++++++++++++++++++++++++--- 3 files changed, 162 insertions(+), 7 deletions(-) diff --git a/tests/connectors/test_excel.py b/tests/connectors/test_excel.py index cf7a2428e..d2513e7be 100644 --- a/tests/connectors/test_excel.py +++ b/tests/connectors/test_excel.py @@ -103,6 +103,42 @@ def test_excel_sheet_append_accumulates_repeated_writes(): assert len(excel_outputs[0]["data"]) == 1000 +def test_excel_sheet_append_aligns_dynamic_batch_columns_by_name(): + """ + Dynamic dictionary keys can create different columns in each batch. + Accumulated Excel output must union the columns and align values by name + instead of stacking each batch positionally. + """ + memory.clear() + batches = [ + pd.DataFrame({"ID": [1], "A": [1], "X": [97]}), + pd.DataFrame({"ID": [2], "B": [2], "Y": [98]}), + pd.DataFrame({"ID": [3], "C": [3], "Z": [99]}), + pd.DataFrame({"ID": [4], "D": [4], "Zz": [100]}), + ] + + for df in batches: + wrangles.connectors.excel.sheet.write(df, name="Results") + + excel_outputs = [ + v + for v in memory.dataframes.values() + if v.get("connector") == "excel.sheet.write" + ] + memory.clear() + + assert len(excel_outputs) == 1 + assert excel_outputs[0]["columns"] == [ + "ID", "A", "X", "B", "Y", "C", "Z", "D", "Zz" + ] + assert excel_outputs[0]["data"] == [ + [1, 1, 97, "", "", "", "", "", ""], + [2, "", "", 2, 98, "", "", "", ""], + [3, "", "", "", "", 3, 99, "", ""], + [4, "", "", "", "", "", "", 4, 100], + ] + + def test_excel_sheet_overwrite_accumulates_repeated_writes(): """ Batched WranglesXL runs may emit repeated overwrite writes to the same @@ -131,6 +167,39 @@ def test_excel_sheet_overwrite_accumulates_repeated_writes(): assert len(excel_outputs[0]["data"]) == 1000 +def test_excel_sheet_overwrite_aligns_reordered_columns_by_name(): + """ + Overwrite batches with the same columns in a different order must retain + the first payload's column order without shifting values. + """ + memory.clear() + wrangles.connectors.excel.sheet.write( + pd.DataFrame({"ID": [1], "A": [1], "X": [97]}), + name="Results", + action="overwrite" + ) + wrangles.connectors.excel.sheet.write( + pd.DataFrame({"X": [98], "ID": [2], "A": [2]}), + name="Results", + action="overwrite" + ) + + excel_outputs = [ + v + for v in memory.dataframes.values() + if v.get("connector") == "excel.sheet.write" + ] + memory.clear() + + assert len(excel_outputs) == 1 + assert excel_outputs[0]["columns"] == ["ID", "A", "X"] + assert excel_outputs[0]["data"] == [ + [1, 1, 97], + [2, 2, 98], + ] + assert excel_outputs[0]["action"] == "overwrite" + + def test_excel_sheet_overwrite_uses_append_after_first_external_batch(): """ WranglesXL can execute each batch as a separate Python run. In that case, diff --git a/tests/recipes/wrangles/test_main.py b/tests/recipes/wrangles/test_main.py index 82ed87dff..c3c8b08a5 100644 --- a/tests/recipes/wrangles/test_main.py +++ b/tests/recipes/wrangles/test_main.py @@ -5715,6 +5715,41 @@ def record_batch(df): number_of_batches[0] == 1000 ) + def test_batch_dynamic_dictionary_columns_align_by_name(self): + """ + Dynamic dictionary keys can produce different columns in every batch. + The combined result must union those columns and align values by name. + """ + df = wrangles.recipe.run( + """ + wrangles: + - batch: + batch_size: 1 + wrangles: + - split.dictionary: + input: Spec Dicts + """, + dataframe=pd.DataFrame({ + "ID": [1, 2, 3, 4], + "Spec Dicts": [ + {"A": "1", "X": "97"}, + {"B": "2", "Y": "98"}, + {"C": "3", "Z": "99"}, + {"D": "4", "Zz": "100"}, + ] + }) + ) + + assert df.columns.tolist() == [ + "ID", "Spec Dicts", "A", "X", "B", "Y", "C", "Z", "D", "Zz" + ] + assert df[["A", "X", "B", "Y", "C", "Z", "D", "Zz"]].values.tolist() == [ + ["1", "97", "", "", "", "", "", ""], + ["", "", "2", "98", "", "", "", ""], + ["", "", "", "", "3", "99", "", ""], + ["", "", "", "", "", "", "4", "100"], + ] + def test_batch_preserves_order(self): """ Test batch preserves diff --git a/wrangles/connectors/excel.py b/wrangles/connectors/excel.py index 64a0ab9a6..5bf7b0041 100644 --- a/wrangles/connectors/excel.py +++ b/wrangles/connectors/excel.py @@ -5,6 +5,60 @@ from . import memory as _memory import logging as _logging + +def _append_rows_by_column(saved: dict, df: _pd.DataFrame) -> bool: + """ + Append a dataframe to an orient="split" payload, aligning values by column + name and adding newly encountered columns in first-seen order. + """ + new_data = df.to_dict(orient="split") + saved_columns = saved["columns"] + new_columns = new_data["columns"] + + # Duplicate labels cannot be aligned by name unambiguously. Preserve the + # existing behavior for identical layouts, but leave different layouts as + # separate writes rather than risking a positional shift. + if ( + saved_columns != new_columns + and ( + not _pd.Index(saved_columns).is_unique + or not _pd.Index(new_columns).is_unique + ) + ): + return False + + combined_columns = saved_columns + [ + column + for column in new_columns + if column not in saved_columns + ] + + if combined_columns == saved_columns == new_columns: + saved["data"].extend(new_data["data"]) + else: + added_columns = len(combined_columns) - len(saved_columns) + if added_columns: + saved["data"] = [ + list(row) + [""] * added_columns + for row in saved["data"] + ] + + column_positions = { + column: position + for position, column in enumerate(combined_columns) + } + for row in new_data["data"]: + aligned_row = [""] * len(combined_columns) + for column, value in zip(new_columns, row): + aligned_row[column_positions[column]] = value + saved["data"].append(aligned_row) + + saved["columns"] = combined_columns + + saved["index"].extend(new_data["index"]) + return True + + class sheet(): _schema = {} @@ -37,14 +91,11 @@ def write(df: _pd.DataFrame, variables: dict = None, **kwargs): and saved.get("name") == name and saved.get("cell") == cell and saved.get("action", "append") in ("append", "overwrite") - and saved.get("columns") == df.columns.tolist() ): - new_data = df.to_dict(orient="split") - saved["data"].extend(new_data["data"]) - saved["index"].extend(new_data["index"]) - if action == "overwrite": - saved["action"] = "overwrite" - return + if _append_rows_by_column(saved, df): + if action == "overwrite": + saved["action"] = "overwrite" + return _memory.write( df,