nsinnovations/erpnext
2
0
Fork 0
mirror of https://github.com/frappe/erpnext.git synced 2026-08-12 06:01:46 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(postgres): rebuild available_stock_for_packing_items report without raw SQL

Browse Source 
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Mihir Kandoi
2026-06-19 16:15:32 +05:30
parent f4d9869d7b
commit 9ce1b02e6e
2 changed files with 216 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files

93
erpnext/selling/report/available_stock_for_packing_items/available_stock_for_packing_items.py
View File

@@ -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

160
erpnext/selling/report/available_stock_for_packing_items/test_available_stock_for_packing_items.py Normal file
View File

@@ -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