Merge pull request #32280 from s-aga-r/refactor/report/item-shortage-report

refactor: rewrite `Item Shortage Report` queries in QB

erpnext/stock/report/item_shortage_report/item_shortage_report.py

@@ -8,8 +8,7 @@ from frappe import _
 
 def execute(filters=None):
 	columns = get_columns()
-	conditions = get_conditions(filters)
+	data = get_data(filters)
-	data = get_data(conditions, filters)
 
 	if not data:
 		return [], [], None, []
@@ -19,49 +18,39 @@ def execute(filters=None):
 	return columns, data, None, chart_data
 
 
-def get_conditions(filters):
+def get_data(filters):
-	conditions = ""
+	bin = frappe.qb.DocType("Bin")
+	wh = frappe.qb.DocType("Warehouse")
+	item = frappe.qb.DocType("Item")
 
-	if filters.get("warehouse"):
+	query = (
-		conditions += "AND warehouse in %(warehouse)s"
+		frappe.qb.from_(bin)
-	if filters.get("company"):
+		.from_(wh)
-		conditions += "AND company = %(company)s"
+		.from_(item)
-
+		.select(
-	return conditions
-
-
-def get_data(conditions, filters):
-	data = frappe.db.sql(
-		"""
-		SELECT
 			bin.warehouse,
 			bin.item_code,
-			bin.actual_qty ,
+			bin.actual_qty,
-			bin.ordered_qty ,
+			bin.ordered_qty,
-			bin.planned_qty ,
+			bin.planned_qty,
-			bin.reserved_qty ,
+			bin.reserved_qty,
 			bin.reserved_qty_for_production,
-			bin.projected_qty ,
+			bin.projected_qty,
-			warehouse.company,
+			wh.company,
-			item.item_name ,
+			item.item_name,
-			item.description
+			item.description,
-		FROM
+		)
-			`tabBin` bin,
+		.where((bin.projected_qty < 0) & (wh.name == bin.warehouse) & (bin.item_code == item.name))
-			`tabWarehouse` warehouse,
+		.orderby(bin.projected_qty)
-			`tabItem` item
-		WHERE
-			bin.projected_qty<0
-			AND warehouse.name = bin.warehouse
-			AND bin.item_code=item.name
-			{0}
-		ORDER BY bin.projected_qty;""".format(
-			conditions
-		),
-		filters,
-		as_dict=1,
 	)
 
-	return data
+	if filters.get("warehouse"):
+		query = query.where(bin.warehouse.isin(filters.get("warehouse")))
+
+	if filters.get("company"):
+		query = query.where(wh.company == filters.get("company"))
+
+	return query.run(as_dict=True)
 
 
 def get_chart_data(data):

erpnext/stock/report/item_shortage_report/test_item_shortage_report.py

@@ -0,0 +1,51 @@
+# Copyright (c) 2022, Frappe Technologies Pvt. Ltd. and contributors
+# For license information, please see license.txt
+
+import frappe
+from frappe.tests.utils import FrappeTestCase
+
+from erpnext.selling.doctype.sales_order.test_sales_order import make_sales_order
+from erpnext.stock.doctype.item.test_item import make_item
+from erpnext.stock.report.item_shortage_report.item_shortage_report import (
+	execute as item_shortage_report,
+)
+
+
+class TestItemShortageReport(FrappeTestCase):
+	def test_item_shortage_report(self):
+		item = make_item().name
+		so = make_sales_order(item_code=item)
+
+		reserved_qty, projected_qty = frappe.db.get_value(
+			"Bin",
+			{
+				"item_code": item,
+				"warehouse": so.items[0].warehouse,
+			},
+			["reserved_qty", "projected_qty"],
+		)
+		self.assertEqual(reserved_qty, so.items[0].qty)
+		self.assertEqual(projected_qty, -(so.items[0].qty))
+
+		filters = {
+			"company": so.company,
+		}
+		report_data = item_shortage_report(filters)[1]
+		item_code_list = [row.get("item_code") for row in report_data]
+		self.assertIn(item, item_code_list)
+
+		filters = {
+			"company": so.company,
+			"warehouse": [so.items[0].warehouse],
+		}
+		report_data = item_shortage_report(filters)[1]
+		item_code_list = [row.get("item_code") for row in report_data]
+		self.assertIn(item, item_code_list)
+
+		filters = {
+			"company": so.company,
+			"warehouse": ["Work In Progress - _TC"],
+		}
+		report_data = item_shortage_report(filters)[1]
+		item_code_list = [row.get("item_code") for row in report_data]
+		self.assertNotIn(item, item_code_list)