fix: Auto format bom_update_log.py

(cherry picked from commit 79495679e2)
This commit is contained in:
marination
2022-03-30 13:08:58 +05:30
committed by mergify-bot
parent 9b069ed04b
commit 1f9ecb3397

View File

@@ -15,6 +15,7 @@ from erpnext.manufacturing.doctype.bom_update_tool.bom_update_tool import update
class BOMMissingError(frappe.ValidationError): class BOMMissingError(frappe.ValidationError):
pass pass
class BOMUpdateLog(Document): class BOMUpdateLog(Document):
def validate(self): def validate(self):
if self.update_type == "Replace BOM": if self.update_type == "Replace BOM":
@@ -28,7 +29,8 @@ class BOMUpdateLog(Document):
if self.update_type == "Replace BOM" and not (self.current_bom and self.new_bom): if self.update_type == "Replace BOM" and not (self.current_bom and self.new_bom):
frappe.throw( frappe.throw(
msg=_("Please mention the Current and New BOM for replacement."), msg=_("Please mention the Current and New BOM for replacement."),
title=_("Mandatory"), exc=BOMMissingError title=_("Mandatory"),
exc=BOMMissingError,
) )
def validate_same_bom(self): def validate_same_bom(self):
@@ -47,20 +49,22 @@ class BOMUpdateLog(Document):
return return
if self.update_type == "Replace BOM": if self.update_type == "Replace BOM":
boms = { boms = {"current_bom": self.current_bom, "new_bom": self.new_bom}
"current_bom": self.current_bom,
"new_bom": self.new_bom
}
frappe.enqueue( frappe.enqueue(
method="erpnext.manufacturing.doctype.bom_update_log.bom_update_log.run_bom_job", method="erpnext.manufacturing.doctype.bom_update_log.bom_update_log.run_bom_job",
doc=self, boms=boms, timeout=40000 doc=self,
boms=boms,
timeout=40000,
) )
else: else:
frappe.enqueue( frappe.enqueue(
method="erpnext.manufacturing.doctype.bom_update_log.bom_update_log.run_bom_job", method="erpnext.manufacturing.doctype.bom_update_log.bom_update_log.run_bom_job",
doc=self, update_type="Update Cost", timeout=40000 doc=self,
update_type="Update Cost",
timeout=40000,
) )
def replace_bom(boms: Dict) -> None: def replace_bom(boms: Dict) -> None:
"""Replace current BOM with new BOM in parent BOMs.""" """Replace current BOM with new BOM in parent BOMs."""
current_bom = boms.get("current_bom") current_bom = boms.get("current_bom")
@@ -69,13 +73,13 @@ def replace_bom(boms: Dict) -> None:
unit_cost = get_new_bom_unit_cost(new_bom) unit_cost = get_new_bom_unit_cost(new_bom)
update_new_bom(unit_cost, current_bom, new_bom) update_new_bom(unit_cost, current_bom, new_bom)
frappe.cache().delete_key('bom_children') frappe.cache().delete_key("bom_children")
parent_boms = get_parent_boms(new_bom) parent_boms = get_parent_boms(new_bom)
with click.progressbar(parent_boms) as parent_boms: with click.progressbar(parent_boms) as parent_boms:
pass pass
for bom in parent_boms: for bom in parent_boms:
bom_obj = frappe.get_cached_doc('BOM', bom) bom_obj = frappe.get_cached_doc("BOM", bom)
# this is only used for versioning and we do not want # this is only used for versioning and we do not want
# to make separate db calls by using load_doc_before_save # to make separate db calls by using load_doc_before_save
# which proves to be expensive while doing bulk replace # which proves to be expensive while doing bulk replace
@@ -85,34 +89,29 @@ def replace_bom(boms: Dict) -> None:
bom_obj.calculate_cost() bom_obj.calculate_cost()
bom_obj.update_parent_cost() bom_obj.update_parent_cost()
bom_obj.db_update() bom_obj.db_update()
if bom_obj.meta.get('track_changes') and not bom_obj.flags.ignore_version: if bom_obj.meta.get("track_changes") and not bom_obj.flags.ignore_version:
bom_obj.save_version() bom_obj.save_version()
def update_new_bom(unit_cost: float, current_bom: str, new_bom: str) -> None: def update_new_bom(unit_cost: float, current_bom: str, new_bom: str) -> None:
bom_item = frappe.qb.DocType("BOM Item") bom_item = frappe.qb.DocType("BOM Item")
frappe.qb.update(bom_item).set( frappe.qb.update(bom_item).set(bom_item.bom_no, new_bom).set(bom_item.rate, unit_cost).set(
bom_item.bom_no, new_bom
).set(
bom_item.rate, unit_cost
).set(
bom_item.amount, (bom_item.stock_qty * unit_cost) bom_item.amount, (bom_item.stock_qty * unit_cost)
).where( ).where(
(bom_item.bom_no == current_bom) (bom_item.bom_no == current_bom) & (bom_item.docstatus < 2) & (bom_item.parenttype == "BOM")
& (bom_item.docstatus < 2)
& (bom_item.parenttype == "BOM")
).run() ).run()
def get_parent_boms(new_bom: str, bom_list: Optional[List] = None) -> List: def get_parent_boms(new_bom: str, bom_list: Optional[List] = None) -> List:
bom_list = bom_list or [] bom_list = bom_list or []
bom_item = frappe.qb.DocType("BOM Item") bom_item = frappe.qb.DocType("BOM Item")
parents = frappe.qb.from_(bom_item).select( parents = (
bom_item.parent frappe.qb.from_(bom_item)
).where( .select(bom_item.parent)
(bom_item.bom_no == new_bom) .where((bom_item.bom_no == new_bom) & (bom_item.docstatus < 2) & (bom_item.parenttype == "BOM"))
& (bom_item.docstatus <2) .run(as_dict=True)
& (bom_item.parenttype == "BOM") )
).run(as_dict=True)
for d in parents: for d in parents:
if new_bom == d.parent: if new_bom == d.parent:
@@ -123,17 +122,19 @@ def get_parent_boms(new_bom: str, bom_list: Optional[List] = None) -> List:
return list(set(bom_list)) return list(set(bom_list))
def get_new_bom_unit_cost(new_bom: str) -> float: def get_new_bom_unit_cost(new_bom: str) -> float:
bom = frappe.qb.DocType("BOM") bom = frappe.qb.DocType("BOM")
new_bom_unitcost = frappe.qb.from_(bom).select( new_bom_unitcost = (
bom.total_cost / bom.quantity frappe.qb.from_(bom).select(bom.total_cost / bom.quantity).where(bom.name == new_bom).run()
).where( )
bom.name == new_bom
).run()
return flt(new_bom_unitcost[0][0]) return flt(new_bom_unitcost[0][0])
def run_bom_job(doc: "BOMUpdateLog", boms: Optional[Dict] = None, update_type: Optional[str] = "Replace BOM") -> None:
def run_bom_job(
doc: "BOMUpdateLog", boms: Optional[Dict] = None, update_type: Optional[str] = "Replace BOM"
) -> None:
try: try:
doc.db_set("status", "In Progress") doc.db_set("status", "In Progress")
if not frappe.flags.in_test: if not frappe.flags.in_test:
@@ -152,10 +153,7 @@ def run_bom_job(doc: "BOMUpdateLog", boms: Optional[Dict] = None, update_type: O
except (Exception, JobTimeoutException): except (Exception, JobTimeoutException):
frappe.db.rollback() frappe.db.rollback()
error_log = frappe.log_error( error_log = frappe.log_error(message=frappe.get_traceback(), title=_("BOM Update Tool Error"))
message=frappe.get_traceback(),
title=_("BOM Update Tool Error")
)
doc.db_set("status", "Failed") doc.db_set("status", "Failed")
doc.db_set("error_log", error_log.name) doc.db_set("error_log", error_log.name)