Merge pull request #56546 from mihir-kandoi/pg-rawsql-to-orm

refactor: convert convertible raw frappe.db.sql to ORM

erpnext/stock/doctype/material_request/test_material_request.py

@@ -746,11 +746,11 @@ class TestMaterialRequest(ERPNextTestSuite):
 		mr = frappe.get_doc("Material Request", mr.name)
 		mr.submit()
 		completed_qty = mr.items[0].ordered_qty
-		requested_qty = frappe.db.sql(
-			"""select indented_qty from `tabBin` where \
-			item_code= %s and warehouse= %s """,
-			(mr.items[0].item_code, mr.items[0].warehouse),
-		)[0][0]
+		requested_qty = frappe.db.get_value(
+			"Bin",
+			{"item_code": mr.items[0].item_code, "warehouse": mr.items[0].warehouse},
+			"indented_qty",
+		)
 
 		prod_order = raise_work_orders(mr.name, mr.company)
 		po = frappe.get_doc("Work Order", prod_order[0])
@@ -760,11 +760,11 @@ class TestMaterialRequest(ERPNextTestSuite):
 		mr = frappe.get_doc("Material Request", mr.name)
 		self.assertEqual(completed_qty + po.qty, mr.items[0].ordered_qty)
 
-		new_requested_qty = frappe.db.sql(
-			"""select indented_qty from `tabBin` where \
-			item_code= %s and warehouse= %s """,
-			(mr.items[0].item_code, mr.items[0].warehouse),
-		)[0][0]
+		new_requested_qty = frappe.db.get_value(
+			"Bin",
+			{"item_code": mr.items[0].item_code, "warehouse": mr.items[0].warehouse},
+			"indented_qty",
+		)
 
 		self.assertEqual(requested_qty - po.qty, new_requested_qty)
 
@@ -773,11 +773,11 @@ class TestMaterialRequest(ERPNextTestSuite):
 		mr = frappe.get_doc("Material Request", mr.name)
 		self.assertEqual(completed_qty, mr.items[0].ordered_qty)
 
-		new_requested_qty = frappe.db.sql(
-			"""select indented_qty from `tabBin` where \
-			item_code= %s and warehouse= %s """,
-			(mr.items[0].item_code, mr.items[0].warehouse),
-		)[0][0]
+		new_requested_qty = frappe.db.get_value(
+			"Bin",
+			{"item_code": mr.items[0].item_code, "warehouse": mr.items[0].warehouse},
+			"indented_qty",
+		)
 		self.assertEqual(requested_qty, new_requested_qty)
 
 	def test_requested_qty_multi_uom(self):

erpnext/stock/doctype/stock_entry/test_stock_entry.py

@@ -40,19 +40,12 @@ from erpnext.tests.utils import ERPNextTestSuite
 
 
 def get_sle(**args):
-	condition, values = "", []
-	for key, value in args.items():
-		condition += " and " if condition else " where "
-		condition += f"`{key}`=%s"
-		values.append(value)
-
-	return frappe.db.sql(
-		# posting_datetime is the precomputed date+time column; MySQL-only timestamp(date,time) errors on Postgres
-		"""select * from `tabStock Ledger Entry` %s
-		order by posting_datetime desc, creation desc limit 1"""
-		% condition,
-		values,
-		as_dict=1,
+	return frappe.get_all(
+		"Stock Ledger Entry",
+		filters=args,
+		fields=["*"],
+		order_by="posting_datetime desc, creation desc",
+		limit=1,
 	)
 
 
@@ -269,20 +262,10 @@ class TestStockEntry(ERPNextTestSuite):
 		mr.cancel()
 
 		self.assertTrue(
-			frappe.db.sql(
-				"""select * from `tabStock Ledger Entry`
-			where voucher_type='Stock Entry' and voucher_no=%s""",
-				mr.name,
-			)
+			frappe.db.exists("Stock Ledger Entry", {"voucher_type": "Stock Entry", "voucher_no": mr.name})
 		)
 
-		self.assertTrue(
-			frappe.db.sql(
-				"""select * from `tabGL Entry`
-			where voucher_type='Stock Entry' and voucher_no=%s""",
-				mr.name,
-			)
-		)
+		self.assertTrue(frappe.db.exists("GL Entry", {"voucher_type": "Stock Entry", "voucher_no": mr.name}))
 
 	def test_material_issue_gl_entry(self):
 		company = frappe.db.get_value("Warehouse", "Stores - TCP1", "company")
@@ -361,12 +344,7 @@ class TestStockEntry(ERPNextTestSuite):
 		if source_warehouse_account == target_warehouse_account:
 			# no gl entry as both source and target warehouse has linked to same account.
 			self.assertFalse(
-				frappe.db.sql(
-					"""select * from `tabGL Entry`
-				where voucher_type='Stock Entry' and voucher_no=%s""",
-					mtn.name,
-					as_dict=1,
-				)
+				frappe.db.exists("GL Entry", {"voucher_type": "Stock Entry", "voucher_no": mtn.name})
 			)
 
 		else:
@@ -460,14 +438,9 @@ class TestStockEntry(ERPNextTestSuite):
 			],
 		)
 
-		gl_entries = frappe.db.sql(
-			"""select account, debit, credit
-			from `tabGL Entry` where voucher_type='Stock Entry' and voucher_no=%s
-			order by account desc""",
-			repack.name,
-			as_dict=1,
+		self.assertFalse(
+			frappe.db.exists("GL Entry", {"voucher_type": "Stock Entry", "voucher_no": repack.name})
 		)
-		self.assertFalse(gl_entries)
 
 	def test_repack_with_additional_costs(self):
 		company = frappe.db.get_value("Warehouse", "Stores - TCP1", "company")
@@ -601,15 +574,15 @@ class TestStockEntry(ERPNextTestSuite):
 		expected_sle.sort(key=lambda x: x[1])
 
 		# check stock ledger entries
-		sle = frappe.db.sql(
-			"""select item_code, warehouse, actual_qty
-			from `tabStock Ledger Entry` where voucher_type = %s
-			and voucher_no = %s order by item_code, warehouse, actual_qty""",
-			(voucher_type, voucher_no),
-			as_list=1,
+		sle = frappe.get_all(
+			"Stock Ledger Entry",
+			filters={"voucher_type": voucher_type, "voucher_no": voucher_no},
+			fields=["item_code", "warehouse", "actual_qty"],
+			order_by="item_code, warehouse, actual_qty",
+			as_list=True,
 		)
 		self.assertTrue(sle)
-		sle.sort(key=lambda x: x[1])
+		sle = sorted(sle, key=lambda x: x[1])
 
 		for i, sle_value in enumerate(sle):
 			self.assertEqual(expected_sle[i][0], sle_value[0])
@@ -619,16 +592,16 @@ class TestStockEntry(ERPNextTestSuite):
 	def check_gl_entries(self, voucher_type, voucher_no, expected_gl_entries):
 		expected_gl_entries.sort(key=lambda x: x[0])
 
-		gl_entries = frappe.db.sql(
-			"""select account, debit, credit
-			from `tabGL Entry` where voucher_type=%s and voucher_no=%s
-			order by account asc, debit asc""",
-			(voucher_type, voucher_no),
-			as_list=1,
+		gl_entries = frappe.get_all(
+			"GL Entry",
+			filters={"voucher_type": voucher_type, "voucher_no": voucher_no},
+			fields=["account", "debit", "credit"],
+			order_by="account asc, debit asc",
+			as_list=True,
 		)
 
 		self.assertTrue(gl_entries)
-		gl_entries.sort(key=lambda x: x[0])
+		gl_entries = sorted(gl_entries, key=lambda x: x[0])
 		for i, gle in enumerate(gl_entries):
 			self.assertEqual(expected_gl_entries[i][0], gle[0])
 			self.assertEqual(expected_gl_entries[i][1], gle[1])

erpnext/stock/doctype/stock_reconciliation/test_stock_reconciliation.py

@@ -85,11 +85,10 @@ class TestStockReconciliation(ERPNextTestSuite, StockTestMixin):
 			)
 
 			# check stock value
-			sle = frappe.db.sql(
-				"""select * from `tabStock Ledger Entry`
-				where voucher_type='Stock Reconciliation' and voucher_no=%s""",
-				stock_reco.name,
-				as_dict=1,
+			sle = frappe.get_all(
+				"Stock Ledger Entry",
+				filters={"voucher_type": "Stock Reconciliation", "voucher_no": stock_reco.name},
+				fields=["qty_after_transaction", "stock_value"],
 			)
 
 			qty_after_transaction = flt(d[0]) if d[0] != "" else flt(last_sle.get("qty_after_transaction"))

erpnext/stock/doctype/warehouse/test_warehouse.py

@@ -19,11 +19,10 @@ class TestWarehouse(ERPNextTestSuite):
 	def test_warehouse_hierarchy(self):
 		p_warehouse = frappe.get_doc("Warehouse", "_Test Warehouse Group - _TC")
 
-		child_warehouses = frappe.db.sql(
-			"""select name, is_group, parent_warehouse from `tabWarehouse` wh
-			where wh.lft > %s and wh.rgt < %s""",
-			(p_warehouse.lft, p_warehouse.rgt),
-			as_dict=1,
+		child_warehouses = frappe.get_all(
+			"Warehouse",
+			filters={"lft": [">", p_warehouse.lft], "rgt": ["<", p_warehouse.rgt]},
+			fields=["name", "is_group", "parent_warehouse"],
 		)
 
 		for child_warehouse in child_warehouses: