From b5805403e9895a4382dcaa94669778988ba40ac0 Mon Sep 17 00:00:00 2001 From: Shahadat Islam Koyel <179203101+ShahadatTest@users.noreply.github.com> Date: Tue, 27 Oct 2026 16:01:09 +0600 Subject: [PATCH] Release XLS and XLSX workbook resources when decoding fails --- benedict/serializers/xls.py | 204 +++++++++++++------------- tests/serializers/test_xls_cleanup.py | 53 +++++++ 2 files changed, 159 insertions(+), 98 deletions(-) create mode 100644 tests/serializers/test_xls_cleanup.py diff --git a/benedict/serializers/xls.py b/benedict/serializers/xls.py index d691d455..24d53884 100644 --- a/benedict/serializers/xls.py +++ b/benedict/serializers/xls.py @@ -82,52 +82,58 @@ def _decode_legacy(self, s: str, **kwargs: Any) -> list[dict[str, Any]]: # load the worksheet workbook = open_workbook(**options) - # get sheet by index or by name - sheet_index, sheet_name = self._get_sheet_index_and_name_from_options(**kwargs) - if sheet_name: - sheet_names = workbook.sheet_names() - sheet_index = self._get_sheet_index_by_name(sheet_name, sheet_names) - sheet = workbook.sheet_by_index(sheet_index) - sheet_columns_range = range(sheet.ncols) - - # get columns - columns = kwargs.pop("columns", None) - columns_row = kwargs.pop("columns_row", True) - columns_standardized = kwargs.pop( - "columns_standardized", columns is None and columns_row - ) - if not columns: - if columns_row: - # if first row is for column names read the names - # for row in sheet.iter_rows(min_row=1, max_row=1): + try: + # get sheet by index or by name + sheet_index, sheet_name = self._get_sheet_index_and_name_from_options( + **kwargs + ) + if sheet_name: + sheet_names = workbook.sheet_names() + sheet_index = self._get_sheet_index_by_name(sheet_name, sheet_names) + sheet = workbook.sheet_by_index(sheet_index) + sheet_columns_range = range(sheet.ncols) + + # get columns + columns = kwargs.pop("columns", None) + columns_row = kwargs.pop("columns_row", True) + columns_standardized = kwargs.pop( + "columns_standardized", columns is None and columns_row + ) + if not columns: + if columns_row: + # if first row is for column names read the names + # for row in sheet.iter_rows(min_row=1, max_row=1): + columns = [ + sheet.cell_value(0, col_index) + for col_index in sheet_columns_range + ] + else: + # otherwise use columns indexes as column names + # for row in sheet.iter_rows(min_row=1, max_row=1): + columns = list(sheet_columns_range) + + # standardize column names, eg. "Date Created" -> "date_created" + if columns_standardized: columns = [ - sheet.cell_value(0, col_index) for col_index in sheet_columns_range + slugify(self._get_column_name(column), separator="_") + for column in columns ] - else: - # otherwise use columns indexes as column names - # for row in sheet.iter_rows(min_row=1, max_row=1): - columns = list(sheet_columns_range) - - # standardize column names, eg. "Date Created" -> "date_created" - if columns_standardized: - columns = [ - slugify(self._get_column_name(column), separator="_") - for column in columns - ] - - # build list of dicts, one for each row - items = [] - items_row_start = 1 if columns_row else 0 - for row_index in range(items_row_start, sheet.nrows): - row = {} - for col_index in sheet_columns_range: - col_key = columns[col_index] - value = sheet.cell_value(row_index, col_index) - row[col_key] = value - items.append(row) - - # print(items) - return items + + # build list of dicts, one for each row + items = [] + items_row_start = 1 if columns_row else 0 + for row_index in range(items_row_start, sheet.nrows): + row = {} + for col_index in sheet_columns_range: + col_key = columns[col_index] + value = sheet.cell_value(row_index, col_index) + row[col_key] = value + items.append(row) + + # print(items) + return items + finally: + workbook.release_resources() def _decode(self, s: str, **kwargs: Any) -> list[dict[str, Any]]: options: dict[str, Any] = {} @@ -140,61 +146,63 @@ def _decode(self, s: str, **kwargs: Any) -> list[dict[str, Any]]: # load the worksheet workbook = load_workbook(**options) - # get sheet by index or by name - sheet_index, sheet_name = self._get_sheet_index_and_name_from_options(**kwargs) - sheets = list(workbook) - if sheet_name: - sheet_names = [sheet.title for sheet in sheets] - sheet_index = self._get_sheet_index_by_name(sheet_name, sheet_names) - sheet = sheets[sheet_index] - sheet_columns_cells = list(sheet.iter_rows(min_row=1, max_row=1))[0] - - # get columns - columns = kwargs.pop("columns", None) - columns_row = kwargs.pop("columns_row", True) - columns_standardized = kwargs.pop( - "columns_standardized", columns is None and columns_row - ) - if not columns: - if columns_row: - # if first row is for column names read the names - # for row in sheet.iter_rows(min_row=1, max_row=1): - columns = [cell.value for cell in sheet_columns_cells] - else: - # otherwise use columns indexes as column names - # for row in sheet.iter_rows(min_row=1, max_row=1): - columns = self._get_sheet_columns_indexes(len(sheet_columns_cells)) - - # standardize column names, eg. "Date Created" -> "date_created" - if columns_standardized: - columns = [ - slugify(self._get_column_name(column), separator="_") - for column in columns - ] - - # build list of dicts, one for each row - items = [] - items_row_start = 2 if columns_row else 1 - for row in sheet.iter_rows(min_row=items_row_start): - values = [cell.value for cell in row] - items.append(dict(zip(columns, values))) - try: - # explicitly close vba_archive if present: Workbook.close() never - # closes it, so relying on GC (__del__ -> close()) is unsafe because - # the underlying BytesIO may already be finalized first, causing a - # ValueError when ZipFile tries to flush the end-of-central-directory - vba_archive = getattr(workbook, "vba_archive", None) - if vba_archive is not None: - vba_archive.close() - except Exception: - pass - - # close the worksheet - workbook.close() - - # print(items) - return items + # get sheet by index or by name + sheet_index, sheet_name = self._get_sheet_index_and_name_from_options( + **kwargs + ) + sheets = list(workbook) + if sheet_name: + sheet_names = [sheet.title for sheet in sheets] + sheet_index = self._get_sheet_index_by_name(sheet_name, sheet_names) + sheet = sheets[sheet_index] + sheet_columns_cells = list(sheet.iter_rows(min_row=1, max_row=1))[0] + + # get columns + columns = kwargs.pop("columns", None) + columns_row = kwargs.pop("columns_row", True) + columns_standardized = kwargs.pop( + "columns_standardized", columns is None and columns_row + ) + if not columns: + if columns_row: + # if first row is for column names read the names + # for row in sheet.iter_rows(min_row=1, max_row=1): + columns = [cell.value for cell in sheet_columns_cells] + else: + # otherwise use columns indexes as column names + # for row in sheet.iter_rows(min_row=1, max_row=1): + columns = self._get_sheet_columns_indexes(len(sheet_columns_cells)) + + # standardize column names, eg. "Date Created" -> "date_created" + if columns_standardized: + columns = [ + slugify(self._get_column_name(column), separator="_") + for column in columns + ] + + # build list of dicts, one for each row + items = [] + items_row_start = 2 if columns_row else 1 + for row in sheet.iter_rows(min_row=items_row_start): + values = [cell.value for cell in row] + items.append(dict(zip(columns, values))) + + return items + finally: + try: + # explicitly close vba_archive if present: Workbook.close() never + # closes it, so relying on GC (__del__ -> close()) is unsafe because + # the underlying BytesIO may already be finalized first, causing a + # ValueError when ZipFile tries to flush the end-of-central-directory + vba_archive = getattr(workbook, "vba_archive", None) + if vba_archive is not None: + vba_archive.close() + except Exception: + pass + + # close the worksheet + workbook.close() def decode(self, s: str, **kwargs: Any) -> list[dict[str, Any]]: require_xls(installed=xls_installed) diff --git a/tests/serializers/test_xls_cleanup.py b/tests/serializers/test_xls_cleanup.py new file mode 100644 index 00000000..b4493899 --- /dev/null +++ b/tests/serializers/test_xls_cleanup.py @@ -0,0 +1,53 @@ +import unittest +from unittest.mock import MagicMock, patch + +from benedict.serializers import XLSSerializer + + +class xls_cleanup_test_case(unittest.TestCase): + def test_modern_workbooks_close_on_invalid_sheet(self) -> None: + for options, message in [ + ({"sheet": "missing"}, "Invalid sheet name"), + ({"sheet": 2}, "list index out of range"), + ]: + with self.subTest(options=options): + book = MagicMock() + sheet = MagicMock() + sheet.title = "Data" + book.__iter__.return_value = iter([sheet]) + with patch("benedict.serializers.xls.load_workbook", return_value=book): + with self.assertRaisesRegex(Exception, message): + XLSSerializer()._decode("example.xlsx", **options) + book.close.assert_called_once_with() + book.vba_archive.close.assert_called_once_with() + + def test_modern_workbooks_close_on_row_error(self) -> None: + book = MagicMock() + sheet = MagicMock() + sheet.iter_rows.side_effect = RuntimeError("unreadable worksheet") + book.__iter__.return_value = iter([sheet]) + with patch("benedict.serializers.xls.load_workbook", return_value=book): + with self.assertRaisesRegex(RuntimeError, "unreadable worksheet"): + XLSSerializer()._decode("example.xlsx") + book.close.assert_called_once_with() + book.vba_archive.close.assert_called_once_with() + + def test_legacy_workbooks_release_resources_on_error(self) -> None: + book = MagicMock() + book.sheet_by_index.side_effect = IndexError("sheet does not exist") + with patch("benedict.serializers.xls.open_workbook", return_value=book): + with self.assertRaisesRegex(IndexError, "sheet does not exist"): + XLSSerializer()._decode_legacy("example.xls", sheet=2) + book.release_resources.assert_called_once_with() + + def test_legacy_workbooks_release_resources_on_success(self) -> None: + book = MagicMock() + sheet = book.sheet_by_index.return_value + sheet.ncols = 1 + sheet.nrows = 2 + sheet.cell_value.side_effect = ["Name", "Ada"] + with patch("benedict.serializers.xls.open_workbook", return_value=book): + self.assertEqual( + XLSSerializer()._decode_legacy("example.xls"), [{"name": "Ada"}] + ) + book.release_resources.assert_called_once_with()