Merge pull request #56179 from mihir-kandoi/pg-stock-masters

refactor(stock): port masters/settings/dashboards raw SQL to qb/ORM (Postgres compat)

erpnext/stock/dashboard/item_dashboard.py

@@ -24,13 +24,19 @@ def get_data(
 		filters.append(["warehouse", "=", warehouse])
 	if item_group:
 		lft, rgt = frappe.db.get_value("Item Group", item_group, ["lft", "rgt"])
-		items = frappe.db.sql_list(
-			"""
-			select i.name from `tabItem` i
-			where exists(select name from `tabItem Group`
-				where name=i.item_group and lft >=%s and rgt<=%s)
-		""",
-			(lft, rgt),
+		item = frappe.qb.DocType("Item")
+		item_group_dt = frappe.qb.DocType("Item Group")
+		items = (
+			frappe.qb.from_(item)
+			.select(item.name)
+			.where(
+				item.item_group.isin(
+					frappe.qb.from_(item_group_dt)
+					.select(item_group_dt.name)
+					.where((item_group_dt.lft >= lft) & (item_group_dt.rgt <= rgt))
+				)
+			)
+			.run(pluck="name")
 		)
 		filters.append(["item_code", "in", items])
 	try:

erpnext/stock/dashboard/warehouse_capacity_dashboard.py

@@ -40,12 +40,8 @@ def get_filters(item_code=None, warehouse=None, parent_warehouse=None, company=N
 		filters.append(["company", "=", company])
 	if parent_warehouse:
 		lft, rgt = frappe.db.get_value("Warehouse", parent_warehouse, ["lft", "rgt"])
-		warehouses = frappe.db.sql_list(
-			"""
-			select name from `tabWarehouse`
-			where lft >=%s and rgt<=%s
-		""",
-			(lft, rgt),
+		warehouses = frappe.get_all(
+			"Warehouse", filters={"lft": [">=", lft], "rgt": ["<=", rgt]}, pluck="name"
 		)
 		filters.append(["warehouse", "in", warehouses])
 	return filters

erpnext/stock/doctype/item_alternative/item_alternative.py

@@ -7,6 +7,7 @@ from typing import Any
 import frappe
 from frappe import _
 from frappe.model.document import Document
+from frappe.utils import cint
 
 
 class ItemAlternative(Document):
@@ -86,13 +87,24 @@ class ItemAlternative(Document):
 @frappe.whitelist()
 @frappe.validate_and_sanitize_search_inputs
 def get_alternative_items(doctype: Any, txt: str, searchfield: Any, start: int, page_len: int, filters: dict):
-	return frappe.db.sql(
-		f""" (select alternative_item_code from `tabItem Alternative`
-			where item_code = %(item_code)s and alternative_item_code like %(txt)s)
-		union
-			(select item_code from `tabItem Alternative`
-			where alternative_item_code = %(item_code)s and item_code like %(txt)s
-			and two_way = 1) limit {page_len} offset {start}
-		""",
-		{"item_code": filters.get("item_code"), "txt": "%" + txt + "%"},
+	item_code = filters.get("item_code")
+	search = f"%{txt}%"
+	# each leg has distinct values (validate_duplicate), so start+page_len rows per leg suffice
+	limit = cint(start) + cint(page_len)
+
+	alternatives = frappe.get_all(
+		"Item Alternative",
+		filters={"item_code": item_code, "alternative_item_code": ["like", search]},
+		pluck="alternative_item_code",
+		limit=limit,
 	)
+	alternatives += frappe.get_all(
+		"Item Alternative",
+		filters={"alternative_item_code": item_code, "item_code": ["like", search], "two_way": 1},
+		pluck="item_code",
+		limit=limit,
+	)
+
+	# union (dedupe, preserve order) + paginate
+	unique_items = list(dict.fromkeys(alternatives))
+	return [[item] for item in unique_items[start : start + page_len]]

erpnext/stock/doctype/item_alternative/test_item_alternative.py

@@ -2,7 +2,7 @@
 # See license.txt
 
 import frappe
-from frappe.utils import flt
+from frappe.utils import flt, random_string
 
 from erpnext.controllers.subcontracting_controller import make_rm_stock_entry
 from erpnext.controllers.tests.test_subcontracting_controller import (
@@ -14,6 +14,7 @@ from erpnext.manufacturing.doctype.production_plan.test_production_plan import m
 from erpnext.manufacturing.doctype.work_order.mapper import make_stock_entry
 from erpnext.manufacturing.doctype.work_order.test_work_order import make_wo_order_test_record
 from erpnext.stock.doctype.item.test_item import create_item
+from erpnext.stock.doctype.item_alternative.item_alternative import get_alternative_items
 from erpnext.stock.doctype.stock_reconciliation.stock_reconciliation import (
 	EmptyStockReconciliationItemsError,
 )
@@ -166,6 +167,192 @@ class TestItemAlternative(ERPNextTestSuite):
 		self.assertEqual(status, True)
 		ste1.submit()
 
+	def test_get_alternative_items_both_directions_and_dedup(self):
+		"""get_alternative_items must return forward alternatives, reverse-only
+		two_way alternatives, exclude one-way reverse rows, and dedupe an item
+		that matches in both the forward and reverse legs of the old UNION."""
+		suffix = random_string(8)
+		base = f"_Test IA Base {suffix}"
+		alt_fwd = f"_Test IA Fwd {suffix}"  # forward only (two_way=0)
+		alt_both = f"_Test IA Both {suffix}"  # forward (two_way=1)
+		alt_rev = f"_Test IA Rev {suffix}"  # reverse via two_way=1
+		alt_norev = f"_Test IA NoRev {suffix}"  # reverse but two_way=0 -> excluded
+		dup = f"_Test IA Dup {suffix}"  # forward AND reverse -> must dedupe
+
+		for item_code in (base, alt_fwd, alt_both, alt_rev, alt_norev, dup):
+			create_item(item_code)
+			item = frappe.get_doc("Item", item_code)
+			if not item.allow_alternative_item:
+				item.allow_alternative_item = 1
+				item.save()
+
+		# forward rows: item_code = base
+		make_item_alternative(base, alt_fwd, two_way=0)
+		make_item_alternative(base, alt_both, two_way=1)
+		make_item_alternative(base, dup, two_way=1)
+
+		# reverse rows: alternative_item_code = base
+		make_item_alternative(alt_rev, base, two_way=1)
+		make_item_alternative(alt_norev, base, two_way=0)
+		make_item_alternative(dup, base, two_way=1)
+
+		# txt = the shared suffix so the LIKE matches every alternate but not `base`
+		results = get_alternative_items("Item", suffix, "name", 0, 20, {"item_code": base})
+
+		# structure: list of single-element lists
+		self.assertTrue(all(isinstance(row, list) and len(row) == 1 for row in results))
+
+		returned = [row[0] for row in results]
+
+		# forward alternatives (both one-way and two_way) are returned
+		self.assertIn(alt_fwd, returned)
+		self.assertIn(alt_both, returned)
+
+		# reverse alternative is only returned when the row is two_way
+		self.assertIn(alt_rev, returned)
+		self.assertNotIn(alt_norev, returned)
+
+		# `base` itself is never an alternative of itself
+		self.assertNotIn(base, returned)
+
+		# an item matching both legs of the old UNION is deduped to a single row
+		self.assertIn(dup, returned)
+		self.assertEqual(returned.count(dup), 1)
+
+	def test_get_alternative_items_respects_txt_filter(self):
+		"""The txt LIKE filter must actually narrow the result set so a
+		non-matching alternate is excluded (guards against a broken WHERE)."""
+		suffix = random_string(8)
+		base = f"_Test IA Filter Base {suffix}"
+		matching = f"_Test IA Match {suffix}"
+		other = f"_Test IA Other {suffix}"
+
+		for item_code in (base, matching, other):
+			create_item(item_code)
+			item = frappe.get_doc("Item", item_code)
+			if not item.allow_alternative_item:
+				item.allow_alternative_item = 1
+				item.save()
+
+		make_item_alternative(base, matching, two_way=0)
+		make_item_alternative(base, other, two_way=0)
+
+		# search only for the `Match` alternate
+		results = get_alternative_items("Item", f"Match {suffix}", "name", 0, 20, {"item_code": base})
+		returned = [row[0] for row in results]
+
+		self.assertIn(matching, returned)
+		self.assertNotIn(other, returned)
+
+	def test_get_alternative_items_case_insensitive_match(self):
+		"""The txt match must stay case-insensitive on BOTH engines: MariaDB LIKE is
+		case-insensitive by default, and frappe compiles the `like` filter to ILIKE on
+		Postgres. A case-shifted search must still find an alternate whose stored code
+		differs in case — this guards against the conversion degrading to a case-sensitive
+		match (plain LIKE / ==) that would silently return nothing on Postgres."""
+		suffix = random_string(8)
+		base = f"_Test IA Case Base {suffix}"
+		# distinctive mixed-case token in the stored alternate's code
+		alt = f"_Test IA CaseToken AbCdE {suffix}"
+
+		for item_code in (base, alt):
+			create_item(item_code)
+			item = frappe.get_doc("Item", item_code)
+			if not item.allow_alternative_item:
+				item.allow_alternative_item = 1
+				item.save()
+
+		make_item_alternative(base, alt, two_way=0)
+
+		# search the LOWERCASED token ("abcde") against the stored "AbCdE"
+		results = get_alternative_items(
+			"Item", f"casetoken abcde {suffix}", "name", 0, 20, {"item_code": base}
+		)
+		returned = [row[0] for row in results]
+
+		self.assertIn(alt, returned)
+
+	def test_get_alternative_items_pagination(self):
+		"""start/page_len must slice the deduped, order-preserving result."""
+		suffix = random_string(8)
+		base = f"_Test IA Page Base {suffix}"
+		alts = [f"_Test IA Page {i} {suffix}" for i in range(3)]
+
+		create_item(base)
+		base_item = frappe.get_doc("Item", base)
+		if not base_item.allow_alternative_item:
+			base_item.allow_alternative_item = 1
+			base_item.save()
+
+		for alt in alts:
+			create_item(alt)
+			alt_item = frappe.get_doc("Item", alt)
+			if not alt_item.allow_alternative_item:
+				alt_item.allow_alternative_item = 1
+				alt_item.save()
+			make_item_alternative(base, alt, two_way=0)
+
+		full = [row[0] for row in get_alternative_items("Item", suffix, "name", 0, 20, {"item_code": base})]
+		self.assertEqual(len(full), 3)
+
+		page = [row[0] for row in get_alternative_items("Item", suffix, "name", 1, 1, {"item_code": base})]
+		self.assertEqual(len(page), 1)
+		self.assertEqual(page[0], full[1])
+
+	def test_get_alternative_items_pagination_is_bounded_and_exact(self):
+		"""Each get_all is bounded to start+page_len rows, so the DB round trip stays small
+		instead of fetching every alternative per keystroke. Walking the result in small pages
+		must still reconstruct the complete deduped set — including an alternate that appears in
+		BOTH legs (forward + reverse two_way) — with no item dropped or duplicated by the bound."""
+		suffix = random_string(8)
+		base = f"_Test IA Bound Base {suffix}"
+		forwards = [f"_Test IA Bound Fwd {i} {suffix}" for i in range(3)]
+		reverses = [f"_Test IA Bound Rev {i} {suffix}" for i in range(3)]
+		dup = f"_Test IA Bound Dup {suffix}"  # forward AND reverse two_way -> deduped across legs
+
+		for item_code in [base, dup, *forwards, *reverses]:
+			create_item(item_code)
+			item = frappe.get_doc("Item", item_code)
+			if not item.allow_alternative_item:
+				item.allow_alternative_item = 1
+				item.save()
+
+		for fwd in forwards:
+			make_item_alternative(base, fwd, two_way=0)
+		make_item_alternative(base, dup, two_way=1)  # dup via the forward leg
+		for rev in reverses:
+			make_item_alternative(rev, base, two_way=1)
+		make_item_alternative(dup, base, two_way=1)  # dup also via the reverse leg
+
+		full = [row[0] for row in get_alternative_items("Item", suffix, "name", 0, 50, {"item_code": base})]
+		# 3 forward + 3 reverse + the single deduped dup = 7 distinct
+		self.assertEqual(len(full), 7)
+		self.assertEqual(full.count(dup), 1)
+
+		# walk in pages of 2; bounded fetches must yield exactly the same set, once each
+		collected = []
+		for start in range(0, 8, 2):
+			collected += [
+				row[0] for row in get_alternative_items("Item", suffix, "name", start, 2, {"item_code": base})
+			]
+
+		self.assertEqual(len(collected), len(set(collected)))  # no duplicates introduced by paging
+		self.assertEqual(set(collected), set(full))  # nothing dropped by the per-leg limit
+		self.assertEqual(collected.count(dup), 1)  # the cross-leg dup survives exactly once
+
+
+def make_item_alternative(item_code, alternative_item_code, two_way=0):
+	doc = frappe.get_doc(
+		{
+			"doctype": "Item Alternative",
+			"item_code": item_code,
+			"alternative_item_code": alternative_item_code,
+			"two_way": two_way,
+		}
+	)
+	doc.insert()
+	return doc
+
 
 def make_items():
 	items = [

erpnext/stock/doctype/packing_slip/packing_slip.py

@@ -207,13 +207,24 @@ class PackingSlip(StatusUpdater):
 @frappe.whitelist()
 @frappe.validate_and_sanitize_search_inputs
 def item_details(doctype: str, txt: str, searchfield: str, start: int, page_len: int, filters: dict):
-	from erpnext.controllers.queries import get_match_cond
+	item = frappe.qb.DocType("Item")
+	dn_item = frappe.qb.DocType("Delivery Note Item")
+	delivery_note = (filters or {}).get("delivery_note")
 
-	return frappe.db.sql(
-		"""select name, item_name, description from `tabItem`
-				where name in ( select item_code FROM `tabDelivery Note Item`
-	 						where parent= {})
-	 			and {} like "{}" {}
-	 			limit  {} offset {} """.format("%s", searchfield, "%s", get_match_cond(doctype), "%s", "%s"),
-		((filters or {}).get("delivery_note"), "%%%s%%" % txt, page_len, start),
+	query = frappe.qb.get_query(
+		"Item",
+		fields=["name", "item_name", "description"],
+		ignore_permissions=False,
+	)
+
+	return (
+		query.where(
+			item.name.isin(
+				frappe.qb.from_(dn_item).select(dn_item.item_code).where(dn_item.parent == delivery_note)
+			)
+			& item[searchfield].like(f"%{txt}%")
+		)
+		.limit(page_len)
+		.offset(start)
+		.run()
 	)

erpnext/stock/doctype/price_list/price_list.py

@@ -5,7 +5,7 @@
 import frappe
 from frappe import _, throw
 from frappe.model.document import Document
-from frappe.utils import cint
+from frappe.utils import cint, now
 
 
 class PriceList(Document):
@@ -47,11 +47,15 @@ class PriceList(Document):
 				frappe.set_value("Buying Settings", "Buying Settings", "buying_price_list", self.name)
 
 	def update_item_price(self):
-		frappe.db.sql(
-			"""update `tabItem Price` set currency=%s,
-			buying=%s, selling=%s, modified=NOW() where price_list=%s""",
-			(self.currency, cint(self.buying), cint(self.selling), self.name),
-		)
+		item_price = frappe.qb.DocType("Item Price")
+		(
+			frappe.qb.update(item_price)
+			.set(item_price.currency, self.currency)
+			.set(item_price.buying, cint(self.buying))
+			.set(item_price.selling, cint(self.selling))
+			.set(item_price.modified, now())
+			.where(item_price.price_list == self.name)
+		).run()
 
 	def on_trash(self):
 		self.delete_price_list_details_key()

erpnext/stock/doctype/price_list/test_price_list.py

@@ -1,4 +1,83 @@
-# Copyright (c) 2015, Frappe Technologies Pvt. Ltd. and Contributors
+# Copyright (c) 2024, Frappe Technologies Pvt. Ltd. and Contributors
 # License: GNU General Public License v3. See license.txt
 
 
+import frappe
+from frappe.utils import random_string
+
+from erpnext.tests.utils import ERPNextTestSuite
+
+
+class TestPriceList(ERPNextTestSuite):
+	def make_price_list(self, currency="INR", buying=1, selling=1):
+		price_list = frappe.get_doc(
+			{
+				"doctype": "Price List",
+				"price_list_name": "_Test PL " + random_string(10),
+				"enabled": 1,
+				"currency": currency,
+				"buying": buying,
+				"selling": selling,
+			}
+		).insert()
+		return price_list
+
+	def make_item_price(self, price_list, item_code="_Test Item", rate=100):
+		return frappe.get_doc(
+			{
+				"doctype": "Item Price",
+				"item_code": item_code,
+				"price_list": price_list,
+				"price_list_rate": rate,
+			}
+		).insert()
+
+	def test_update_item_price_propagates_currency_and_flags(self):
+		# Price List starts in INR, applicable for both buying and selling.
+		price_list = self.make_price_list(currency="INR", buying=1, selling=1)
+
+		ip1 = self.make_item_price(price_list.name, item_code="_Test Item", rate=100)
+		ip2 = self.make_item_price(price_list.name, item_code="_Test Item 2", rate=250)
+
+		# Sanity: Item Price rows inherited the Price List's initial state.
+		for ip in (ip1, ip2):
+			row = frappe.db.get_value("Item Price", ip.name, ["currency", "buying", "selling"], as_dict=True)
+			self.assertEqual(row.currency, "INR")
+			self.assertEqual(row.buying, 1)
+			self.assertEqual(row.selling, 1)
+
+		# Change the Price List's currency and flip the buying flag off.
+		# on_update -> update_item_price() should bulk-UPDATE every Item Price
+		# linked to this Price List.
+		price_list.currency = "USD"
+		price_list.buying = 0
+		price_list.selling = 1
+		price_list.save()
+
+		for ip in (ip1, ip2):
+			row = frappe.db.get_value("Item Price", ip.name, ["currency", "buying", "selling"], as_dict=True)
+			self.assertEqual(row.currency, "USD")
+			self.assertEqual(row.buying, 0)
+			self.assertEqual(row.selling, 1)
+
+	def test_update_item_price_scoped_to_own_price_list(self):
+		# Two independent Price Lists; updating one must not touch the other's
+		# Item Price rows (the WHERE price_list == self.name clause).
+		pl_a = self.make_price_list(currency="INR", buying=1, selling=1)
+		pl_b = self.make_price_list(currency="INR", buying=1, selling=1)
+
+		ip_a = self.make_item_price(pl_a.name, item_code="_Test Item", rate=100)
+		ip_b = self.make_item_price(pl_b.name, item_code="_Test Item", rate=100)
+
+		pl_a.currency = "USD"
+		pl_a.buying = 0
+		pl_a.save()
+
+		row_a = frappe.db.get_value("Item Price", ip_a.name, ["currency", "buying"], as_dict=True)
+		self.assertEqual(row_a.currency, "USD")
+		self.assertEqual(row_a.buying, 0)
+
+		# pl_b was untouched, so its Item Price must keep the original values.
+		row_b = frappe.db.get_value("Item Price", ip_b.name, ["currency", "buying"], as_dict=True)
+		self.assertEqual(row_b.currency, "INR")
+		self.assertEqual(row_b.buying, 1)

erpnext/stock/doctype/quality_inspection/quality_inspection.py

@@ -215,14 +215,13 @@ class QualityInspection(Document):
 
 		if self.reference_type == "Job Card":
 			if self.reference_name:
-				frappe.db.sql(
-					f"""
-					UPDATE `tab{self.reference_type}`
-					SET quality_inspection = %s, modified = %s
-					WHERE name = %s and production_item = %s
-				""",
-					(quality_inspection, self.modified, self.reference_name, self.item_code),
-				)
+				ref = frappe.qb.DocType(self.reference_type)
+				(
+					frappe.qb.update(ref)
+					.set(ref.quality_inspection, quality_inspection)
+					.set(ref.modified, self.modified)
+					.where((ref.name == self.reference_name) & (ref.production_item == self.item_code))
+				).run()
 
 		else:
 			doctype = self.reference_type + " Item"

erpnext/stock/doctype/quality_inspection/test_quality_inspection.py

@@ -279,6 +279,65 @@ class TestQualityInspection(ERPNextTestSuite):
 
 		se.delete()
 
+	def test_qi_updates_job_card_reference(self):
+		"""Submitting a QI with reference_type 'Job Card' writes its name onto the
+		Job Card's quality_inspection field (the Job Card branch of
+		QualityInspection.update_qc_reference)."""
+		create_item("_Test Item")
+
+		# Job Card whose production_item matches the QI item_code -> must be updated.
+		matching_jc = make_minimal_job_card(production_item="_Test Item")
+		# Job Card with a different production_item -> the production_item filter must
+		# keep it untouched.
+		other_item = create_item("_Test Item for QC " + frappe.utils.random_string(6)).name
+		non_matching_jc = make_minimal_job_card(production_item=other_item)
+
+		qa = create_quality_inspection(
+			item_code="_Test Item",
+			reference_type="Job Card",
+			reference_name=matching_jc,
+		)
+
+		# The converted UPDATE wrote the QI name onto the matching Job Card.
+		self.assertEqual(
+			frappe.db.get_value("Job Card", matching_jc, "quality_inspection"),
+			qa.name,
+		)
+		# The production_item filter excluded the Job Card with a different item.
+		self.assertFalse(frappe.db.get_value("Job Card", non_matching_jc, "quality_inspection"))
+
+	def test_qi_job_card_reference_respects_production_item(self):
+		"""A QI referencing a Job Card by name but whose item_code does not match the
+		Job Card's production_item must NOT update that Job Card."""
+		create_item("_Test Item")
+		mismatch_item = create_item("_Test Item Mismatch QC " + frappe.utils.random_string(6)).name
+
+		# Job Card produces a different item than the QI's item_code.
+		jc = make_minimal_job_card(production_item=mismatch_item)
+
+		create_quality_inspection(
+			item_code="_Test Item",
+			reference_type="Job Card",
+			reference_name=jc,
+		)
+
+		# name matches but production_item != item_code, so the row is left untouched.
+		self.assertFalse(frappe.db.get_value("Job Card", jc, "quality_inspection"))
+
+
+def make_minimal_job_card(production_item):
+	"""db_insert a minimal submitted Job Card row carrying only the columns the
+	converted UPDATE reads (name, production_item, quality_inspection, modified)."""
+	jc = frappe.new_doc("Job Card")
+	jc.name = "_T-Job Card-" + frappe.utils.random_string(10)
+	jc.flags.name_set = True
+	jc.production_item = production_item
+	jc.company = "_Test Company"
+	jc.for_quantity = 1
+	jc.docstatus = 1
+	jc.db_insert()
+	return jc.name
+
 
 def create_quality_inspection(**args):
 	args = frappe._dict(args)

erpnext/stock/doctype/serial_no/serial_no.py

@@ -101,11 +101,10 @@ class SerialNo(StockController):
 			self.maintenance_status = "Under Warranty"
 
 	def on_trash(self):
-		sl_entries = frappe.db.sql(
-			"""select serial_no from `tabStock Ledger Entry`
-			where serial_no like %s and item_code=%s and is_cancelled=0""",
-			("%%%s%%" % self.name, self.item_code),
-			as_dict=True,
+		sl_entries = frappe.get_all(
+			"Stock Ledger Entry",
+			filters={"serial_no": ["like", f"%{self.name}%"], "item_code": self.item_code, "is_cancelled": 0},
+			fields=["serial_no"],
 		)
 
 		# Find the exact match
@@ -172,13 +171,14 @@ def clean_serial_no_string(serial_no: str) -> str:
 
 
 def update_maintenance_status():
-	serial_nos = frappe.db.sql(
-		"""select name from `tabSerial No` where (amc_expiry_date<%s or
-		warranty_expiry_date<%s) and maintenance_status not in ('Out of Warranty', 'Out of AMC')""",
-		(nowdate(), nowdate()),
+	serial_nos = frappe.get_all(
+		"Serial No",
+		filters={"maintenance_status": ["not in", ["Out of Warranty", "Out of AMC"]]},
+		or_filters=[["amc_expiry_date", "<", nowdate()], ["warranty_expiry_date", "<", nowdate()]],
+		pluck="name",
 	)
 	for serial_no in serial_nos:
-		doc = frappe.get_doc("Serial No", serial_no[0])
+		doc = frappe.get_doc("Serial No", serial_no)
 		doc.set_maintenance_status()
 		frappe.db.set_value("Serial No", doc.name, "maintenance_status", doc.maintenance_status)
 

erpnext/stock/doctype/serial_no/test_serial_no.py

@@ -7,6 +7,7 @@
 
 import frappe
 from frappe import _dict
+from frappe.utils import add_days, nowdate, random_string
 
 from erpnext.stock.doctype.delivery_note.test_delivery_note import create_delivery_note
 from erpnext.stock.doctype.item.test_item import make_item
@@ -16,6 +17,7 @@ from erpnext.stock.doctype.serial_and_batch_bundle.test_serial_and_batch_bundle
 	get_serial_nos_from_bundle,
 )
 from erpnext.stock.doctype.serial_no.serial_no import *
+from erpnext.stock.doctype.serial_no.serial_no import update_maintenance_status
 from erpnext.stock.doctype.stock_entry.stock_entry_utils import make_stock_entry
 from erpnext.stock.doctype.stock_entry.test_stock_entry import make_serialized_item
 from erpnext.stock.doctype.warehouse.test_warehouse import create_warehouse
@@ -331,6 +333,129 @@ class TestSerialNo(ERPNextTestSuite):
 
 		self.assertEqual(non_expired_serials, [])
 
+	def test_update_maintenance_status_expires_past_warranty(self):
+		"""update_maintenance_status() must pick up the past-warranty Serial No via or_filters and flip it Out of Warranty."""
+		item_code = "_Test Serialized Item"
+		past_date = add_days(nowdate(), -10)
+		future_date = add_days(nowdate(), 10)
+
+		# Serial No whose warranty has lapsed; force maintenance_status back to a
+		# value that passes the `not in [Out of Warranty, Out of AMC]` filter so the
+		# cron job is the thing that actually transitions it.
+		expired_sr = frappe.get_doc(
+			{
+				"doctype": "Serial No",
+				"item_code": item_code,
+				"serial_no": "_TCWARREXP" + random_string(6),
+				"company": "_Test Company",
+				"warranty_expiry_date": past_date,
+			}
+		).insert()
+		frappe.db.set_value("Serial No", expired_sr.name, "maintenance_status", "Under Warranty")
+		self.assertEqual(
+			frappe.db.get_value("Serial No", expired_sr.name, "maintenance_status"), "Under Warranty"
+		)
+
+		# Serial No whose warranty is still valid; it must stay Under Warranty.
+		active_sr = frappe.get_doc(
+			{
+				"doctype": "Serial No",
+				"item_code": item_code,
+				"serial_no": "_TCWARRACT" + random_string(6),
+				"company": "_Test Company",
+				"warranty_expiry_date": future_date,
+			}
+		).insert()
+		self.assertEqual(
+			frappe.db.get_value("Serial No", active_sr.name, "maintenance_status"), "Under Warranty"
+		)
+
+		update_maintenance_status()
+
+		# The lapsed Serial No was selected by the or_filters and re-evaluated.
+		self.assertEqual(
+			frappe.db.get_value("Serial No", expired_sr.name, "maintenance_status"), "Out of Warranty"
+		)
+		# The in-warranty Serial No keeps its correct Under Warranty status.
+		self.assertEqual(
+			frappe.db.get_value("Serial No", active_sr.name, "maintenance_status"), "Under Warranty"
+		)
+
+	def test_update_maintenance_status_excludes_out_of_amc(self):
+		"""The `not in [Out of Warranty, Out of AMC]` filter must skip rows already pinned to
+		those statuses, even when they match the expiry or_filters, while rows in any other
+		status ARE re-evaluated. The contrast makes the `not in` clause load-bearing."""
+		item_code = "_Test Serialized Item"
+		past_date = add_days(nowdate(), -10)
+		future_date = add_days(nowdate(), 10)
+
+		# Excluded row: matches or_filters (amc lapsed) AND is pinned to "Out of AMC", so the
+		# `not in` filter must skip it. Its warranty is in the FUTURE, so if the filter were
+		# broken and the row were re-evaluated, set_maintenance_status() would flip it to
+		# "Under Warranty" (the last cascade branch to match). Staying "Out of AMC" proves it.
+		excluded_sr = frappe.get_doc(
+			{
+				"doctype": "Serial No",
+				"item_code": item_code,
+				"serial_no": "_TCAMCEXCL" + random_string(6),
+				"company": "_Test Company",
+				"amc_expiry_date": past_date,
+				"warranty_expiry_date": future_date,
+			}
+		).insert()
+		frappe.db.set_value("Serial No", excluded_sr.name, "maintenance_status", "Out of AMC")
+
+		# Negative control: same lapsed amc date, but a status NOT in the excluded list, so it
+		# must be picked up and re-evaluated to "Out of AMC". This proves update_maintenance_status()
+		# actually processes candidates — i.e. the exclusion above is meaningful, not a no-op.
+		candidate_sr = frappe.get_doc(
+			{
+				"doctype": "Serial No",
+				"item_code": item_code,
+				"serial_no": "_TCAMCCAND" + random_string(6),
+				"company": "_Test Company",
+				"amc_expiry_date": past_date,
+			}
+		).insert()
+		frappe.db.set_value("Serial No", candidate_sr.name, "maintenance_status", "Under AMC")
+
+		update_maintenance_status()
+
+		# Excluded by the `not in` filter -> pinned status left untouched.
+		self.assertEqual(
+			frappe.db.get_value("Serial No", excluded_sr.name, "maintenance_status"), "Out of AMC"
+		)
+		# Not excluded -> re-evaluated; lapsed amc -> "Out of AMC".
+		self.assertEqual(
+			frappe.db.get_value("Serial No", candidate_sr.name, "maintenance_status"), "Out of AMC"
+		)
+
+	def test_update_maintenance_status_includes_null_status(self):
+		"""Converting the raw `maintenance_status not in (...)` to a get_all filter changes NULL
+		handling: frappe wraps the clause as `ifnull(maintenance_status, '') not in (...)`, so a
+		NULL-status row that matches the expiry or_filters is now re-evaluated (consistently on
+		MariaDB and Postgres). Pin that contract."""
+		item_code = "_Test Serialized Item"
+		past_date = add_days(nowdate(), -10)
+
+		null_sr = frappe.get_doc(
+			{
+				"doctype": "Serial No",
+				"item_code": item_code,
+				"serial_no": "_TCAMCNULL" + random_string(6),
+				"company": "_Test Company",
+				"amc_expiry_date": past_date,
+			}
+		).insert()
+		# Force a NULL maintenance_status while a lapsed amc date keeps the row in or_filters.
+		frappe.db.set_value("Serial No", null_sr.name, "maintenance_status", None)
+		self.assertIsNone(frappe.db.get_value("Serial No", null_sr.name, "maintenance_status"))
+
+		update_maintenance_status()
+
+		# Picked up (NULL -> '' -> not in the excluded list) and re-evaluated: lapsed amc -> "Out of AMC".
+		self.assertEqual(frappe.db.get_value("Serial No", null_sr.name, "maintenance_status"), "Out of AMC")
+
 
 def get_auto_serial_nos(kwargs):
 	from erpnext.stock.doctype.serial_and_batch_bundle.serial_and_batch_bundle import (

erpnext/stock/doctype/stock_settings/stock_settings.py

@@ -120,7 +120,7 @@ class StockSettings(Document):
 		if not doc_before_save:
 			return
 
-		if not frappe.get_all("Serial and Batch Bundle", filters={"docstatus": 1}, limit=1, pluck="name"):
+		if not frappe.db.exists("Serial and Batch Bundle", {"docstatus": 1}):
 			return
 
 		if doc_before_save.do_not_use_batchwise_valuation and not self.do_not_use_batchwise_valuation:
@@ -142,7 +142,7 @@ class StockSettings(Document):
 			doc_before_save.enable_serial_and_batch_no_for_item
 			and not self.enable_serial_and_batch_no_for_item
 		):
-			if frappe.get_all("Serial and Batch Bundle", filters={"docstatus": 1}, limit=1, pluck="name"):
+			if frappe.db.exists("Serial and Batch Bundle", {"docstatus": 1}):
 				frappe.throw(
 					_(
 						"Cannot disable Serial and Batch No for Item, as there are existing records for serial / batch."
@@ -170,11 +170,20 @@ class StockSettings(Document):
 		if previous_valuation_method and previous_valuation_method != self.valuation_method:
 			# check if there are any stock ledger entries against items
 			# which does not have it's own valuation method
-			sle = frappe.db.sql(
-				"""select name from `tabStock Ledger Entry` sle
-				where exists(select name from tabItem
-					where name=sle.item_code and (valuation_method is null or valuation_method='')) limit 1
-			"""
+			sle_dt = frappe.qb.DocType("Stock Ledger Entry")
+			item = frappe.qb.DocType("Item")
+			sle = (
+				frappe.qb.from_(sle_dt)
+				.select(sle_dt.name)
+				.where(
+					sle_dt.item_code.isin(
+						frappe.qb.from_(item)
+						.select(item.name)
+						.where(item.valuation_method.isnull() | (item.valuation_method == ""))
+					)
+				)
+				.limit(1)
+				.run()
 			)
 
 			if sle: