refactor: Add exception handling in background job within BOM Update Tool

erpnext/hooks.py

@@ -469,7 +469,7 @@
 	],
 	"daily_long": [
 		"erpnext.setup.doctype.email_digest.email_digest.send",
-		"erpnext.manufacturing.doctype.bom_update_tool.bom_update_tool.update_latest_price_in_all_boms",
+		"erpnext.manufacturing.doctype.bom_update_tool.bom_update_tool.auto_update_latest_price_in_all_boms",
 		"erpnext.hr.doctype.leave_ledger_entry.leave_ledger_entry.process_expired_allocation",
 		"erpnext.hr.utils.generate_leave_encashment",
 		"erpnext.hr.utils.allocate_earned_leaves",

erpnext/manufacturing/doctype/bom_update_log/bom_update_log.js

@@ -0,0 +1,8 @@
+// Copyright (c) 2022, Frappe Technologies Pvt. Ltd. and contributors
+// For license information, please see license.txt
+
+frappe.ui.form.on('BOM Update Log', {
+	// refresh: function(frm) {
+
+	// }
+});

erpnext/manufacturing/doctype/bom_update_log/bom_update_log.json

@@ -0,0 +1,106 @@
+{
+ "actions": [],
+ "autoname": "BOM-UPDT-LOG-.#####",
+ "creation": "2022-03-16 14:23:35.210155",
+ "description": "BOM Update Tool Log with job status maintained",
+ "doctype": "DocType",
+ "editable_grid": 1,
+ "engine": "InnoDB",
+ "field_order": [
+  "current_bom",
+  "new_bom",
+  "column_break_3",
+  "update_type",
+  "status",
+  "error_log",
+  "amended_from"
+ ],
+ "fields": [
+  {
+   "fieldname": "current_bom",
+   "fieldtype": "Link",
+   "in_list_view": 1,
+   "label": "Current BOM",
+   "options": "BOM"
+  },
+  {
+   "fieldname": "new_bom",
+   "fieldtype": "Link",
+   "in_list_view": 1,
+   "label": "New BOM",
+   "options": "BOM"
+  },
+  {
+   "fieldname": "column_break_3",
+   "fieldtype": "Column Break"
+  },
+  {
+   "fieldname": "update_type",
+   "fieldtype": "Select",
+   "label": "Update Type",
+   "options": "Replace BOM\nUpdate Cost"
+  },
+  {
+   "fieldname": "status",
+   "fieldtype": "Select",
+   "label": "Status",
+   "options": "Queued\nIn Progress\nCompleted\nFailed"
+  },
+  {
+   "fieldname": "amended_from",
+   "fieldtype": "Link",
+   "label": "Amended From",
+   "no_copy": 1,
+   "options": "BOM Update Log",
+   "print_hide": 1,
+   "read_only": 1
+  },
+  {
+   "fieldname": "error_log",
+   "fieldtype": "Link",
+   "label": "Error Log",
+   "options": "Error Log"
+  }
+ ],
+ "in_create": 1,
+ "index_web_pages_for_search": 1,
+ "is_submittable": 1,
+ "links": [],
+ "modified": "2022-03-17 12:51:28.067900",
+ "modified_by": "Administrator",
+ "module": "Manufacturing",
+ "name": "BOM Update Log",
+ "naming_rule": "Expression (old style)",
+ "owner": "Administrator",
+ "permissions": [
+  {
+   "create": 1,
+   "delete": 1,
+   "email": 1,
+   "export": 1,
+   "print": 1,
+   "read": 1,
+   "report": 1,
+   "role": "System Manager",
+   "share": 1,
+   "submit": 1,
+   "write": 1
+  },
+  {
+   "create": 1,
+   "email": 1,
+   "export": 1,
+   "print": 1,
+   "read": 1,
+   "report": 1,
+   "role": "Manufacturing Manager",
+   "share": 1,
+   "submit": 1,
+   "write": 1
+  }
+ ],
+ "sort_field": "modified",
+ "sort_order": "DESC",
+ "states": [],
+ "track_changes": 1
+}

erpnext/manufacturing/doctype/bom_update_log/bom_update_log.py

@@ -0,0 +1,163 @@
+# Copyright (c) 2022, Frappe Technologies Pvt. Ltd. and contributors
+# For license information, please see license.txt
+from typing import Dict, List, Optional
+
+import click
+import frappe
+from frappe import _
+from frappe.model.document import Document
+from frappe.utils import cstr, flt
+from rq.timeouts import JobTimeoutException
+
+from erpnext.manufacturing.doctype.bom_update_tool.bom_update_tool import update_cost
+
+
+class BOMMissingError(frappe.ValidationError):
+	pass
+
+
+class BOMUpdateLog(Document):
+	def validate(self):
+		if self.update_type == "Replace BOM":
+			self.validate_boms_are_specified()
+			self.validate_same_bom()
+			self.validate_bom_items()
+
+		self.status = "Queued"
+
+	def validate_boms_are_specified(self):
+		if self.update_type == "Replace BOM" and not (self.current_bom and self.new_bom):
+			frappe.throw(
+				msg=_("Please mention the Current and New BOM for replacement."),
+				title=_("Mandatory"),
+				exc=BOMMissingError,
+			)
+
+	def validate_same_bom(self):
+		if cstr(self.current_bom) == cstr(self.new_bom):
+			frappe.throw(_("Current BOM and New BOM can not be same"))
+
+	def validate_bom_items(self):
+		current_bom_item = frappe.db.get_value("BOM", self.current_bom, "item")
+		new_bom_item = frappe.db.get_value("BOM", self.new_bom, "item")
+
+		if current_bom_item != new_bom_item:
+			frappe.throw(_("The selected BOMs are not for the same item"))
+
+	def on_submit(self):
+		if frappe.flags.in_test:
+			return
+
+		if self.update_type == "Replace BOM":
+			boms = {"current_bom": self.current_bom, "new_bom": self.new_bom}
+			frappe.enqueue(
+				method="erpnext.manufacturing.doctype.bom_update_log.bom_update_log.run_bom_job",
+				doc=self,
+				boms=boms,
+				timeout=40000,
+			)
+		else:
+			frappe.enqueue(
+				method="erpnext.manufacturing.doctype.bom_update_log.bom_update_log.run_bom_job",
+				doc=self,
+				update_type="Update Cost",
+				timeout=40000,
+			)
+
+
+def replace_bom(boms: Dict) -> None:
+	"""Replace current BOM with new BOM in parent BOMs."""
+	current_bom = boms.get("current_bom")
+	new_bom = boms.get("new_bom")
+
+	unit_cost = get_new_bom_unit_cost(new_bom)
+	update_new_bom(unit_cost, current_bom, new_bom)
+
+	frappe.cache().delete_key("bom_children")
+	parent_boms = get_parent_boms(new_bom)
+
+	with click.progressbar(parent_boms) as parent_boms:
+		pass
+	for bom in parent_boms:
+		bom_obj = frappe.get_cached_doc("BOM", bom)
+		# this is only used for versioning and we do not want
+		# to make separate db calls by using load_doc_before_save
+		# which proves to be expensive while doing bulk replace
+		bom_obj._doc_before_save = bom_obj
+		bom_obj.update_new_bom(unit_cost, current_bom, new_bom)
+		bom_obj.update_exploded_items()
+		bom_obj.calculate_cost()
+		bom_obj.update_parent_cost()
+		bom_obj.db_update()
+		if bom_obj.meta.get("track_changes") and not bom_obj.flags.ignore_version:
+			bom_obj.save_version()
+
+
+def update_new_bom(unit_cost: float, current_bom: str, new_bom: str) -> None:
+	bom_item = frappe.qb.DocType("BOM Item")
+	frappe.qb.update(bom_item).set(bom_item.bom_no, new_bom).set(bom_item.rate, unit_cost).set(
+		bom_item.amount, (bom_item.stock_qty * unit_cost)
+	).where(
+		(bom_item.bom_no == current_bom) & (bom_item.docstatus < 2) & (bom_item.parenttype == "BOM")
+	).run()
+
+
+def get_parent_boms(new_bom: str, bom_list: Optional[List] = None) -> List:
+	bom_list = bom_list or []
+	bom_item = frappe.qb.DocType("BOM Item")
+
+	parents = (
+		frappe.qb.from_(bom_item)
+		.select(bom_item.parent)
+		.where((bom_item.bom_no == new_bom) & (bom_item.docstatus < 2) & (bom_item.parenttype == "BOM"))
+		.run(as_dict=True)
+	)
+
+	for d in parents:
+		if new_bom == d.parent:
+			frappe.throw(_("BOM recursion: {0} cannot be child of {1}").format(new_bom, d.parent))
+
+		bom_list.append(d.parent)
+		get_parent_boms(d.parent, bom_list)
+
+	return list(set(bom_list))
+
+
+def get_new_bom_unit_cost(new_bom: str) -> float:
+	bom = frappe.qb.DocType("BOM")
+	new_bom_unitcost = (
+		frappe.qb.from_(bom).select(bom.total_cost / bom.quantity).where(bom.name == new_bom).run()
+	)
+
+	return flt(new_bom_unitcost[0][0])
+
+
+def run_bom_job(
+	doc: "BOMUpdateLog", boms: Optional[Dict] = None, update_type: Optional[str] = "Replace BOM"
+) -> None:
+	try:
+		doc.db_set("status", "In Progress")
+		if not frappe.flags.in_test:
+			frappe.db.commit()
+
+		frappe.db.auto_commit_on_many_writes = 1
+
+		boms = frappe._dict(boms or {})
+
+		if update_type == "Replace BOM":
+			replace_bom(boms)
+		else:
+			update_cost()
+
+		doc.db_set("status", "Completed")
+
+	except (Exception, JobTimeoutException):
+		frappe.db.rollback()
+		error_log = frappe.log_error(message=frappe.get_traceback(), title=_("BOM Update Tool Error"))
+
+		doc.db_set("status", "Failed")
+		doc.db_set("error_log", error_log.name)
+
+	finally:
+		frappe.db.auto_commit_on_many_writes = 0
+		frappe.db.commit()  # nosemgrep

erpnext/manufacturing/doctype/bom_update_log/bom_update_log_list.js

@@ -0,0 +1,13 @@
+frappe.listview_settings['BOM Update Log'] = {
+	add_fields: ["status"],
+	get_indicator: function(doc) {
+		let status_map = {
+			"Queued": "orange",
+			"In Progress": "blue",
+			"Completed": "green",
+			"Failed": "red"
+		};
+
+		return [__(doc.status), status_map[doc.status], "status,=," + doc.status];
+	}
+};

erpnext/manufacturing/doctype/bom_update_log/test_bom_update_log.py

@@ -0,0 +1,90 @@
+# Copyright (c) 2022, Frappe Technologies Pvt. Ltd. and Contributors
+# See license.txt
+
+import frappe
+from frappe.tests.utils import FrappeTestCase
+
+from erpnext.manufacturing.doctype.bom_update_log.bom_update_log import (
+	BOMMissingError,
+	run_bom_job,
+)
+from erpnext.manufacturing.doctype.bom_update_tool.bom_update_tool import enqueue_replace_bom
+
+test_records = frappe.get_test_records("BOM")
+
+
+class TestBOMUpdateLog(FrappeTestCase):
+	"Test BOM Update Tool Operations via BOM Update Log."
+
+	def setUp(self):
+		bom_doc = frappe.copy_doc(test_records[0])
+		bom_doc.items[1].item_code = "_Test Item"
+		bom_doc.insert()
+
+		self.boms = frappe._dict(
+			current_bom="BOM-_Test Item Home Desktop Manufactured-001",
+			new_bom=bom_doc.name,
+		)
+
+		self.new_bom_doc = bom_doc
+
+	def tearDown(self):
+		frappe.db.rollback()
+
+		if self._testMethodName == "test_bom_update_log_completion":
+			# clear logs and delete BOM created via setUp
+			frappe.db.delete("BOM Update Log")
+			self.new_bom_doc.cancel()
+			self.new_bom_doc.delete()
+
+			# explicitly commit and restore to original state
+			frappe.db.commit()  # nosemgrep
+
+	def test_bom_update_log_validate(self):
+		"Test if BOM presence is validated."
+
+		with self.assertRaises(BOMMissingError):
+			enqueue_replace_bom(boms={})
+
+	def test_bom_update_log_queueing(self):
+		"Test if BOM Update Log is created and queued."
+
+		log = enqueue_replace_bom(
+			boms=self.boms,
+		)
+
+		self.assertEqual(log.docstatus, 1)
+		self.assertEqual(log.status, "Queued")
+
+	def test_bom_update_log_completion(self):
+		"Test if BOM Update Log handles job completion correctly."
+
+		log = enqueue_replace_bom(
+			boms=self.boms,
+		)
+
+		# Explicitly commits log, new bom (setUp) and replacement impact.
+		# Is run via background jobs IRL
+		run_bom_job(
+			doc=log,
+			boms=self.boms,
+			update_type="Replace BOM",
+		)
+		log.reload()
+
+		self.assertEqual(log.status, "Completed")
+
+		# teardown (undo replace impact) due to commit
+		boms = frappe._dict(
+			current_bom=self.boms.new_bom,
+			new_bom=self.boms.current_bom,
+		)
+		log2 = enqueue_replace_bom(
+			boms=self.boms,
+		)
+		run_bom_job(  # Explicitly commits
+			doc=log2,
+			boms=boms,
+			update_type="Replace BOM",
+		)
+		self.assertEqual(log2.status, "Completed")