refactor: convert query to QB and make creation optional

erpnext/stock/doctype/batch/test_batch.py

@@ -1,6 +1,8 @@
 # Copyright (c) 2015, Frappe Technologies Pvt. Ltd. and Contributors
 # License: GNU General Public License v3. See license.txt
 
+import json
+
 import frappe
 from frappe.exceptions import ValidationError
 from frappe.utils import cint, flt
@@ -347,7 +349,7 @@ class TestBatch(ERPNextTestCase):
 			self.assertAlmostEqual(sle.qty_after_transaction, qty_after_transaction)
 			self.assertAlmostEqual(sle.valuation_rate, stock_value / qty_after_transaction)
 
-			self.assertEqual(sle.stock_queue, [])  # queues don't apply on batched items
+			self.assertEqual(json.loads(sle.stock_queue), [])  # queues don't apply on batched items
 
 	def test_moving_batch_valuation_rates(self):
 		item_code = "_TestBatchWiseVal"

erpnext/stock/stock_ledger.py

@@ -8,7 +8,9 @@ from typing import Optional
 import frappe
 from frappe import _
 from frappe.model.meta import get_field_precision
+from frappe.query_builder.functions import Sum
 from frappe.utils import cint, cstr, flt, get_link_to_form, getdate, now, nowdate
+from pypika import CustomFunction
 
 import erpnext
 from erpnext.stock.doctype.bin.bin import update_qty as update_bin_qty
@@ -24,7 +26,6 @@ class NegativeStockError(frappe.ValidationError): pass
 class SerialNoExistsInFutureTransaction(frappe.ValidationError):
 	pass
 
-_exceptions = frappe.local('stockledger_exceptions')
 
 def make_sl_entries(sl_entries, allow_negative_stock=False, via_landed_cost_voucher=False):
 	from erpnext.controllers.stock_controller import future_sle_exists
@@ -917,32 +918,32 @@ def get_sle_by_voucher_detail_no(voucher_detail_no, excluded_sle=None):
 
 def get_batch_incoming_rate(item_code, warehouse, batch_no, posting_date, posting_time, creation=None):
 
-	batch_details = frappe.db.sql("""
+	Timestamp = CustomFunction('timestamp', ['date', 'time'])
-		select sum(stock_value_difference) as batch_value, sum(actual_qty) as batch_qty
+
-		from `tabStock Ledger Entry`
+	sle = frappe.qb.DocType("Stock Ledger Entry")
-		where
+
-			item_code = %(item_code)s
+	timestamp_condition = (Timestamp(sle.posting_date, sle.posting_time) < Timestamp(posting_date, posting_time))
-			and warehouse = %(warehouse)s
+	if creation:
-			and batch_no = %(batch_no)s
+		timestamp_condition |= (
-			and is_cancelled = 0
+				(Timestamp(sle.posting_date, sle.posting_time) == Timestamp(posting_date, posting_time))
-			and (
+				& (sle.creation < creation)
-				timestamp(posting_date, posting_time) < timestamp(%(posting_date)s, %(posting_time)s)
-				or (
-					timestamp(posting_date, posting_time) = timestamp(%(posting_date)s, %(posting_time)s)
-					and creation < %(creation)s
-				)
 			)
-		""",
+
-		{
+	batch_details = (
-			"item_code": item_code,
+		frappe.qb
-			"warehouse": warehouse,
+			.from_(sle)
-			"batch_no": batch_no,
+			.select(
-			"posting_date": posting_date,
+				Sum(sle.stock_value_difference).as_("batch_value"),
-			"posting_time": posting_time,
+				Sum(sle.actual_qty).as_("batch_qty")
-			"creation": creation,
+			)
-		},
+			.where(
-		as_dict=True
+				(sle.item_code == item_code)
-	)
+				& (sle.warehouse == warehouse)
+				& (sle.batch_no == batch_no)
+				& (sle.is_cancelled == 0)
+			)
+			.where(timestamp_condition)
+	).run(as_dict=True)
 
 	if batch_details and batch_details[0].batch_qty:
 		return batch_details[0].batch_value / batch_details[0].batch_qty