From 9ce1b02e6e086a084bebb4e261829df6f5acfd6c Mon Sep 17 00:00:00 2001 From: Mihir Kandoi Date: Fri, 19 Jun 2026 16:15:32 +0530 Subject: [PATCH] refactor(postgres): rebuild available_stock_for_packing_items report without raw SQL Co-Authored-By: Claude Opus 4.8 (1M context) --- .../available_stock_for_packing_items.py | 93 ++++++---- .../test_available_stock_for_packing_items.py | 160 ++++++++++++++++++ 2 files changed, 216 insertions(+), 37 deletions(-) create mode 100644 erpnext/selling/report/available_stock_for_packing_items/test_available_stock_for_packing_items.py diff --git a/erpnext/selling/report/available_stock_for_packing_items/available_stock_for_packing_items.py b/erpnext/selling/report/available_stock_for_packing_items/available_stock_for_packing_items.py index b312abd4607..0c93800bd8b 100644 --- a/erpnext/selling/report/available_stock_for_packing_items/available_stock_for_packing_items.py +++ b/erpnext/selling/report/available_stock_for_packing_items/available_stock_for_packing_items.py @@ -12,7 +12,7 @@ def execute(filters=None): columns = get_columns() iwq_map = get_item_warehouse_quantity_map() - item_map = get_item_details() + item_map = get_item_details(list(iwq_map.keys())) data = [] for sbom, warehouse in iwq_map.items(): total = 0 @@ -53,48 +53,67 @@ def get_columns(): return columns -def get_item_details(): +def get_item_details(item_codes): + # only the bundle items actually shown in the report need detail lookup, not the whole catalogue + if not item_codes: + return {} item_map = {} - for item in frappe.db.sql( - """SELECT name, item_name, description, stock_uom - from `tabItem`""", - as_dict=1, + for item in frappe.get_all( + "Item", + filters={"name": ["in", item_codes]}, + fields=["name", "item_name", "description", "stock_uom"], ): item_map.setdefault(item.name, item) return item_map def get_item_warehouse_quantity_map(): - query = """SELECT parent, warehouse, MIN(qty) AS qty - FROM (SELECT b.parent, bi.item_code, bi.warehouse, - sum(bi.projected_qty) / b.qty AS qty - FROM tabBin AS bi, (SELECT pb.new_item_code as parent, b.item_code, b.qty, w.name - FROM `tabProduct Bundle Item` b, `tabWarehouse` w, - `tabProduct Bundle` pb - where b.parent = pb.name - and pb.is_active = 1 and pb.docstatus = 1) AS b - WHERE bi.item_code = b.item_code - AND bi.warehouse = b.name - GROUP BY b.parent, b.item_code, bi.warehouse - UNION ALL - SELECT b.parent, b.item_code, b.name, 0 AS qty - FROM (SELECT pb.new_item_code as parent, b.item_code, b.qty, w.name - FROM `tabProduct Bundle Item` b, `tabWarehouse` w, - `tabProduct Bundle` pb - where b.parent = pb.name - and pb.is_active = 1 and pb.docstatus = 1) AS b - WHERE NOT EXISTS(SELECT * - FROM `tabBin` AS bi - WHERE bi.item_code = b.item_code - AND bi.warehouse = b.name)) AS r - GROUP BY parent, warehouse - HAVING MIN(qty) != 0""" - result = frappe.db.sql(query, as_dict=1) - last_sbom = "" + # Components of every active product bundle: (bundle item code, component item, qty per bundle) + pb = frappe.qb.DocType("Product Bundle") + pbi = frappe.qb.DocType("Product Bundle Item") + bundle_components = ( + frappe.qb.from_(pbi) + .inner_join(pb) + .on(pbi.parent == pb.name) + .select(pb.new_item_code.as_("parent"), pbi.item_code, pbi.qty) + .where((pb.is_active == 1) & (pb.docstatus == 1)) + .run(as_dict=True) + ) + + if not bundle_components: + return {} + + component_items = list({c.item_code for c in bundle_components}) + + bin_projected = { + (b.item_code, b.warehouse): flt(b.projected_qty) + for b in frappe.get_all( + "Bin", + filters={"item_code": ["in", component_items]}, + fields=["item_code", "warehouse", "projected_qty"], + ) + } + + # Only warehouses that hold at least one component can yield a non-zero packable qty; a warehouse + # missing any component yields MIN()=0 and is dropped below, so scanning every warehouse in the + # system is wasted work. Scope the loop to warehouses present in the Bin result. + bin_warehouses = {wh for (_, wh) in bin_projected} + + # For each (bundle, warehouse) the number of complete bundles that can be packed is the + # MIN over components of (component projected_qty in that warehouse / component qty per bundle). + # A component with no Bin in a warehouse contributes 0 (the original UNION ALL/NOT EXISTS branch). + packable_qty = {} + for component in bundle_components: + if not component.qty: + continue + for warehouse in bin_warehouses: + qty = bin_projected.get((component.item_code, warehouse), 0) / flt(component.qty) + key = (component.parent, warehouse) + packable_qty[key] = min(packable_qty[key], qty) if key in packable_qty else qty + sbom_map = {} - for line in result: - if line.get("parent") != last_sbom: - last_sbom = line.get("parent") - actual_dict = sbom_map.setdefault(last_sbom, {}) - actual_dict.setdefault(line.get("warehouse"), line.get("qty")) + for (parent, warehouse), qty in packable_qty.items(): + if qty != 0: # HAVING MIN(qty) != 0 + sbom_map.setdefault(parent, {})[warehouse] = qty + return sbom_map diff --git a/erpnext/selling/report/available_stock_for_packing_items/test_available_stock_for_packing_items.py b/erpnext/selling/report/available_stock_for_packing_items/test_available_stock_for_packing_items.py new file mode 100644 index 00000000000..75c116e00a0 --- /dev/null +++ b/erpnext/selling/report/available_stock_for_packing_items/test_available_stock_for_packing_items.py @@ -0,0 +1,160 @@ +# Copyright (c) 2024, Frappe Technologies Pvt. Ltd. and Contributors +# License: GNU General Public License v3. See license.txt + +import frappe +from frappe.utils import flt, random_string + +from erpnext.selling.report.available_stock_for_packing_items.available_stock_for_packing_items import ( + execute, +) +from erpnext.stock.doctype.item.test_item import make_item +from erpnext.tests.utils import ERPNextTestSuite + +WAREHOUSE = "_Test Warehouse - _TC" + + +class TestAvailableStockForPackingItems(ERPNextTestSuite): + """Cover the MIN-over-components / per-warehouse logic of the rewritten report. + + The report computes, for each (active Product Bundle, warehouse): + packable bundles = MIN over components of (Bin.projected_qty / qty per bundle) + and drops rows where that MIN is 0. We use freshly created component items so the + only Bin rows and the only bundle referencing them are the ones built here -- this + keeps the asserted number exact and makes the test fail if the conversion breaks. + """ + + def make_component(self): + return make_item( + f"_Test Packing Component {random_string(10)}", + {"is_stock_item": 1}, + ).name + + def make_bundle_parent(self): + return make_item( + f"_Test Packing Bundle {random_string(10)}", + {"is_stock_item": 0, "is_sales_item": 1}, + ).name + + def set_bin_projected_qty(self, item_code, warehouse, projected_qty): + """Create (if needed) the Bin for (item, warehouse) and pin projected_qty. + + Bin recomputes projected_qty from actual/ordered/... on save, so after the + Bin exists we force the exact value with db.set_value (no controller recompute). + This is precisely the column the report reads back. + """ + name = frappe.db.get_value("Bin", {"item_code": item_code, "warehouse": warehouse}) + if not name: + bin_doc = frappe.get_doc(doctype="Bin", item_code=item_code, warehouse=warehouse) + bin_doc.flags.ignore_permissions = True + bin_doc.insert() + name = bin_doc.name + frappe.db.set_value("Bin", name, "projected_qty", projected_qty) + return name + + def make_active_bundle(self, parent, components): + """components: list of (item_code, qty_per_bundle). Submitted => is_active, docstatus 1.""" + bundle = frappe.get_doc({"doctype": "Product Bundle", "new_item_code": parent}) + for item_code, qty in components: + bundle.append("items", {"item_code": item_code, "qty": qty}) + bundle.insert() + bundle.submit() + return bundle + + def report_rows_for(self, parent): + """Run the report and return the data rows whose Item Code == parent (drops Total rows).""" + _columns, data = execute(filters=None) + return [row for row in data if row and row[0] == parent] + + def test_min_over_components_binds(self): + comp_a = self.make_component() + comp_b = self.make_component() + parent = self.make_bundle_parent() + + # comp_a: 2 per bundle, projected 10 -> 5 bundles; comp_b: 1 per bundle, projected 3 -> 3 bundles + self.set_bin_projected_qty(comp_a, WAREHOUSE, 10) + self.set_bin_projected_qty(comp_b, WAREHOUSE, 3) + self.make_active_bundle(parent, [(comp_a, 2), (comp_b, 1)]) + + rows = self.report_rows_for(parent) + + # Exactly one (bundle, warehouse) row, and packable == MIN(5, 3) == 3. + self.assertEqual(len(rows), 1) + row = rows[0] + # row shape: [item_code, item_name, description, uom, warehouse, quantity] + self.assertEqual(row[4], WAREHOUSE) + self.assertEqual(flt(row[5]), 3.0) + + def test_per_warehouse_grouping(self): + comp_a = self.make_component() + comp_b = self.make_component() + parent = self.make_bundle_parent() + + other_wh = self.make_secondary_warehouse() + + # _Test Warehouse: comp_a 8/2=4, comp_b 6/1=6 -> MIN 4 + self.set_bin_projected_qty(comp_a, WAREHOUSE, 8) + self.set_bin_projected_qty(comp_b, WAREHOUSE, 6) + # other warehouse: comp_a 4/2=2, comp_b 9/1=9 -> MIN 2 + self.set_bin_projected_qty(comp_a, other_wh, 4) + self.set_bin_projected_qty(comp_b, other_wh, 9) + + self.make_active_bundle(parent, [(comp_a, 2), (comp_b, 1)]) + + rows = self.report_rows_for(parent) + by_warehouse = {row[4]: flt(row[5]) for row in rows} + + self.assertEqual(by_warehouse.get(WAREHOUSE), 4.0) + self.assertEqual(by_warehouse.get(other_wh), 2.0) + + def test_starved_component_drops_row(self): + comp_a = self.make_component() + comp_b = self.make_component() + parent = self.make_bundle_parent() + + # comp_a is plentiful, comp_b is absent in the warehouse (no Bin) -> MIN == 0 -> dropped. + self.set_bin_projected_qty(comp_a, WAREHOUSE, 50) + self.make_active_bundle(parent, [(comp_a, 2), (comp_b, 1)]) + + self.assertEqual(self.report_rows_for(parent), []) + + def test_zero_projected_component_drops_row(self): + comp_a = self.make_component() + comp_b = self.make_component() + parent = self.make_bundle_parent() + + # comp_b present but with projected 0 -> 0/1 == 0 -> MIN == 0 -> row dropped. + self.set_bin_projected_qty(comp_a, WAREHOUSE, 20) + self.set_bin_projected_qty(comp_b, WAREHOUSE, 0) + self.make_active_bundle(parent, [(comp_a, 2), (comp_b, 1)]) + + self.assertEqual(self.report_rows_for(parent), []) + + def test_inactive_bundle_excluded(self): + comp_a = self.make_component() + parent = self.make_bundle_parent() + + self.set_bin_projected_qty(comp_a, WAREHOUSE, 10) + bundle = self.make_active_bundle(parent, [(comp_a, 1)]) + + # Sanity: while active it shows up... + self.assertTrue(self.report_rows_for(parent)) + + # ...and disappears once cancelled (is_active cleared, docstatus 2). + bundle.cancel() + self.assertEqual(self.report_rows_for(parent), []) + + def make_secondary_warehouse(self): + """A second leaf warehouse under _Test Company so two warehouses can be asserted.""" + name = f"_Test Pack WH {random_string(6)}" + full_name = f"{name} - _TC" + if frappe.db.exists("Warehouse", full_name): + return full_name + wh = frappe.get_doc( + { + "doctype": "Warehouse", + "warehouse_name": name, + "company": "_Test Company", + } + ) + wh.insert() + return wh.name