fix: Type Annotations, Redundancy, etc.
- Renamed public function`update_new_bom` to `update_new_bom_in_bom_items`
- Replaced `get_cached_doc` with `get_doc`
- Removed click progress bar (drive through update log)
- Removed `bom_obj.update_new_bom()`, was redundant. Did same job as `update_new_bom_in_bom_items`
- Removed `update_new_bom()` in `bom.py`, unused.
- Prettier query formatting
- `update_type` annotated as non optional Literal
- Removed redundant use of JobTimeoutException
- Corrected type annotations in `create_bom_update_log()`
diff --git a/erpnext/manufacturing/doctype/bom/bom.py b/erpnext/manufacturing/doctype/bom/bom.py
index bf29474..8c0112b 100644
--- a/erpnext/manufacturing/doctype/bom/bom.py
+++ b/erpnext/manufacturing/doctype/bom/bom.py
@@ -697,15 +697,6 @@
self.scrap_material_cost = total_sm_cost
self.base_scrap_material_cost = base_total_sm_cost
- def update_new_bom(self, old_bom, new_bom, rate):
- for d in self.get("items"):
- if d.bom_no != old_bom:
- continue
-
- d.bom_no = new_bom
- d.rate = rate
- d.amount = (d.stock_qty or d.qty) * rate
-
def update_exploded_items(self, save=True):
"""Update Flat BOM, following will be correct data"""
self.get_exploded_items()
diff --git a/erpnext/manufacturing/doctype/bom_update_log/bom_update_log.py b/erpnext/manufacturing/doctype/bom_update_log/bom_update_log.py
index ce27743..139dcbc 100644
--- a/erpnext/manufacturing/doctype/bom_update_log/bom_update_log.py
+++ b/erpnext/manufacturing/doctype/bom_update_log/bom_update_log.py
@@ -1,13 +1,11 @@
# Copyright (c) 2022, Frappe Technologies Pvt. Ltd. and contributors
# For license information, please see license.txt
-from typing import Dict, List, Optional
+from typing import Dict, List, Literal, 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
@@ -71,20 +69,17 @@
new_bom = boms.get("new_bom")
unit_cost = get_new_bom_unit_cost(new_bom)
- update_new_bom(unit_cost, current_bom, new_bom)
+ update_new_bom_in_bom_items(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)
+ bom_obj = frappe.get_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()
@@ -93,12 +88,16 @@
bom_obj.save_version()
-def update_new_bom(unit_cost: float, current_bom: str, new_bom: str) -> None:
+def update_new_bom_in_bom_items(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")
+ (
+ 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()
@@ -133,7 +132,9 @@
def run_bom_job(
- doc: "BOMUpdateLog", boms: Optional[Dict] = None, update_type: Optional[str] = "Replace BOM"
+ doc: "BOMUpdateLog",
+ boms: Optional[Dict[str, str]] = None,
+ update_type: Literal["Replace BOM", "Update Cost"] = "Replace BOM",
) -> None:
try:
doc.db_set("status", "In Progress")
@@ -151,7 +152,7 @@
doc.db_set("status", "Completed")
- except (Exception, JobTimeoutException):
+ except Exception:
frappe.db.rollback()
error_log = frappe.log_error(message=frappe.get_traceback(), title=_("BOM Update Tool Error"))
diff --git a/erpnext/manufacturing/doctype/bom_update_tool/bom_update_tool.py b/erpnext/manufacturing/doctype/bom_update_tool/bom_update_tool.py
index a9189dd..b0e7da1 100644
--- a/erpnext/manufacturing/doctype/bom_update_tool/bom_update_tool.py
+++ b/erpnext/manufacturing/doctype/bom_update_tool/bom_update_tool.py
@@ -2,7 +2,7 @@
# For license information, please see license.txt
import json
-from typing import TYPE_CHECKING, Dict, Optional, Union
+from typing import TYPE_CHECKING, Dict, Literal, Optional, Union
if TYPE_CHECKING:
from erpnext.manufacturing.doctype.bom_update_log.bom_update_log import BOMUpdateLog
@@ -51,9 +51,11 @@
def create_bom_update_log(
- boms: Optional[Dict] = None, update_type: str = "Replace BOM"
+ boms: Optional[Dict[str, str]] = None,
+ update_type: Literal["Replace BOM", "Update Cost"] = "Replace BOM",
) -> "BOMUpdateLog":
"""Creates a BOM Update Log that handles the background job."""
+
boms = boms or {}
current_bom = boms.get("current_bom")
new_bom = boms.get("new_bom")