diff --git a/.git-blame-ignore-revs b/.git-blame-ignore-revs index 1b94ca5b40b..2a62e2a821d 100644 --- a/.git-blame-ignore-revs +++ b/.git-blame-ignore-revs @@ -38,3 +38,6 @@ ec74a5e56617bbd76ac402451468fd4668af543d # ruff formatting a308792ee7fda18a681e9181f4fd00b36385bc23 + +# noisy typing refactoring of get_item_details +7b7211ac79c248a79ba8a999ff34e734d874c0ae diff --git a/erpnext/accounts/doctype/pos_invoice/pos_invoice.py b/erpnext/accounts/doctype/pos_invoice/pos_invoice.py index 9c82e3132fe..03aabb7f3b1 100644 --- a/erpnext/accounts/doctype/pos_invoice/pos_invoice.py +++ b/erpnext/accounts/doctype/pos_invoice/pos_invoice.py @@ -533,7 +533,7 @@ class POSInvoice(SalesInvoice): def set_pos_fields(self, for_validate=False): """Set retail related fields from POS Profiles""" - from erpnext.stock.get_item_details import get_pos_profile, get_pos_profile_item_details + from erpnext.stock.get_item_details import get_pos_profile, get_pos_profile_item_details_ if not self.pos_profile: pos_profile = get_pos_profile(self.company) or {} @@ -602,8 +602,8 @@ class POSInvoice(SalesInvoice): # set pos values in items for item in self.get("items"): if item.get("item_code"): - profile_details = get_pos_profile_item_details( - profile.get("company"), frappe._dict(item.as_dict()), profile + profile_details = get_pos_profile_item_details_( + frappe._dict(item.as_dict()), profile.get("company"), profile ) for fname, val in profile_details.items(): if (not for_validate) or (for_validate and not item.get(fname)): diff --git a/erpnext/accounts/doctype/sales_invoice/sales_invoice.py b/erpnext/accounts/doctype/sales_invoice/sales_invoice.py index 5a28d815ecf..a69ffb21808 100644 --- a/erpnext/accounts/doctype/sales_invoice/sales_invoice.py +++ b/erpnext/accounts/doctype/sales_invoice/sales_invoice.py @@ -766,7 +766,7 @@ class SalesInvoice(SellingController): "Company", self.company, "default_cash_account" ) - from erpnext.stock.get_item_details import get_pos_profile, get_pos_profile_item_details + from erpnext.stock.get_item_details import get_pos_profile, get_pos_profile_item_details_ if not self.pos_profile and not self.flags.ignore_pos_profile: pos_profile = get_pos_profile(self.company) or {} @@ -834,8 +834,8 @@ class SalesInvoice(SellingController): # set pos values in items for item in self.get("items"): if item.get("item_code"): - profile_details = get_pos_profile_item_details( - pos, frappe._dict(item.as_dict()), pos, update_data=True + profile_details = get_pos_profile_item_details_( + frappe._dict(item.as_dict()), pos, pos, update_data=True ) for fname, val in profile_details.items(): if (not for_validate) or (for_validate and not item.get(fname)): diff --git a/erpnext/assets/doctype/asset_capitalization/asset_capitalization.py b/erpnext/assets/doctype/asset_capitalization/asset_capitalization.py index b38bd952bca..d7d289b618e 100644 --- a/erpnext/assets/doctype/asset_capitalization/asset_capitalization.py +++ b/erpnext/assets/doctype/asset_capitalization/asset_capitalization.py @@ -28,7 +28,7 @@ from erpnext.stock.doctype.item.item import get_item_defaults from erpnext.stock.get_item_details import ( get_default_cost_center, get_default_expense_account, - get_item_warehouse, + get_item_warehouse_, ) from erpnext.stock.stock_ledger import get_previous_sle from erpnext.stock.utils import get_incoming_rate @@ -785,11 +785,11 @@ def get_target_asset_details(asset=None, company=None): @frappe.whitelist() -def get_consumed_stock_item_details(args): - if isinstance(args, str): - args = json.loads(args) +def get_consumed_stock_item_details(args_): + if isinstance(args_, str): + args_ = json.loads(args_) - args = frappe._dict(args) + args = frappe._dict(args_) out = frappe._dict() item = frappe._dict() @@ -803,7 +803,7 @@ def get_consumed_stock_item_details(args): out.stock_qty = flt(args.stock_qty) or 1 out.stock_uom = item.stock_uom - out.warehouse = get_item_warehouse(item, args, overwrite_warehouse=True) if item else None + out.warehouse = get_item_warehouse_(args, item, overwrite_warehouse=True) if item else None # Cost Center item_defaults = get_item_defaults(item.name, args.company) diff --git a/erpnext/controllers/accounts_controller.py b/erpnext/controllers/accounts_controller.py index bf5770859b5..e4d6746702b 100644 --- a/erpnext/controllers/accounts_controller.py +++ b/erpnext/controllers/accounts_controller.py @@ -66,7 +66,7 @@ from erpnext.stock.get_item_details import ( get_conversion_factor, get_item_details, get_item_tax_map, - get_item_warehouse, + get_item_warehouse_, ) from erpnext.utilities.regional import temporary_flag from erpnext.utilities.transaction_base import TransactionBase @@ -3218,7 +3218,7 @@ def set_order_defaults(parent_doctype, parent_doctype_name, child_doctype, child child_item.update({date_fieldname: trans_item.get(date_fieldname) or p_doc.get(date_fieldname)}) child_item.stock_uom = item.stock_uom child_item.uom = trans_item.get("uom") or item.stock_uom - child_item.warehouse = get_item_warehouse(item, p_doc, overwrite_warehouse=True) + child_item.warehouse = get_item_warehouse_(p_doc, item, overwrite_warehouse=True) conversion_factor = flt(get_conversion_factor(item.item_code, child_item.uom).get("conversion_factor")) child_item.conversion_factor = flt(trans_item.get("conversion_factor")) or conversion_factor @@ -3227,7 +3227,7 @@ def set_order_defaults(parent_doctype, parent_doctype_name, child_doctype, child child_item.base_rate = 1 child_item.base_amount = 1 if child_doctype == "Sales Order Item": - child_item.warehouse = get_item_warehouse(item, p_doc, overwrite_warehouse=True) + child_item.warehouse = get_item_warehouse_(p_doc, item, overwrite_warehouse=True) if not child_item.warehouse: frappe.throw( _( diff --git a/erpnext/deprecation_dumpster.py b/erpnext/deprecation_dumpster.py index cafec6b9fa2..f780e93f521 100644 --- a/erpnext/deprecation_dumpster.py +++ b/erpnext/deprecation_dumpster.py @@ -125,3 +125,27 @@ def taxes_and_totals_get_itemised_taxable_amount(items): itemised_taxable_amount[item_code] += item.net_amount return itemised_taxable_amount + + +@deprecated( + "erpnext.stock.get_pos_profile_item_details", + "2024-11-19", + "v16", + "Use erpnext.stock.get_pos_profile_item_details_ with a flipped signature", +) +def get_pos_profile_item_details(company, ctx, pos_profile=None, update_data=False): + from erpnext.stock.get_item_details import get_pos_profile_item_details_ + + return get_pos_profile_item_details_(ctx, company, pos_profile=pos_profile, update_data=update_data) + + +@deprecated( + "erpnext.stock.get_item_warehouse", + "2024-11-19", + "v16", + "Use erpnext.stock.get_item_warehouse_ with a flipped signature", +) +def get_item_warehouse(item, ctx, overwrite_warehouse, defaults=None): + from erpnext.stock.get_item_details import get_item_warehouse_ + + return get_item_warehouse_(ctx, item, overwrite_warehouse, defaults=defaults) diff --git a/erpnext/selling/doctype/sales_order/sales_order.py b/erpnext/selling/doctype/sales_order/sales_order.py index 9e7d94f1885..450909dd5ea 100755 --- a/erpnext/selling/doctype/sales_order/sales_order.py +++ b/erpnext/selling/doctype/sales_order/sales_order.py @@ -860,7 +860,7 @@ def make_material_request(source_name, target_doc=None): ) target.rate = flt( - get_price_list_rate(args=args, item_doc=frappe.get_cached_doc("Item", target.item_code)).get( + get_price_list_rate(args, item_doc=frappe.get_cached_doc("Item", target.item_code)).get( "price_list_rate" ) ) diff --git a/erpnext/stock/doctype/item_price/test_item_price.py b/erpnext/stock/doctype/item_price/test_item_price.py index 69827f8a157..eb2efaa9396 100644 --- a/erpnext/stock/doctype/item_price/test_item_price.py +++ b/erpnext/stock/doctype/item_price/test_item_price.py @@ -7,7 +7,7 @@ from frappe.tests import IntegrationTestCase, UnitTestCase from frappe.tests.utils import make_test_records_for_doctype from erpnext.stock.doctype.item_price.item_price import ItemPriceDuplicateItem -from erpnext.stock.get_item_details import get_price_list_rate_for, process_args +from erpnext.stock.get_item_details import get_price_list_rate_for class UnitTestItemPrice(UnitTestCase): @@ -88,7 +88,7 @@ class TestItemPrice(IntegrationTestCase): "qty": 10, } - price = get_price_list_rate_for(process_args(args), doc.item_code) + price = get_price_list_rate_for(args, doc.item_code) self.assertEqual(price, 20.0) def test_price_with_no_qty(self): diff --git a/erpnext/stock/get_item_details.py b/erpnext/stock/get_item_details.py index 388093ad62d..da683af3b42 100644 --- a/erpnext/stock/get_item_details.py +++ b/erpnext/stock/get_item_details.py @@ -3,10 +3,12 @@ import json +from functools import WRAPPER_ASSIGNMENTS, wraps import frappe from frappe import _, throw from frappe.model import child_table_fields, default_fields +from frappe.model.document import Document from frappe.model.meta import get_field_precision from frappe.model.utils import get_fetch_values from frappe.query_builder.functions import IfNull, Sum @@ -24,6 +26,10 @@ from erpnext.stock.doctype.item.item import get_item_defaults, get_uom_conv_fact from erpnext.stock.doctype.item_manufacturer.item_manufacturer import get_item_manufacturer_part_no from erpnext.stock.doctype.price_list.price_list import get_price_list_details +ItemDetails = frappe._dict +ItemDetailsCtx = frappe._dict +ItemPriceCtx = frappe._dict + sales_doctypes = ["Quotation", "Sales Order", "Delivery Note", "Sales Invoice", "POS Invoice"] purchase_doctypes = [ "Material Request", @@ -34,10 +40,40 @@ purchase_doctypes = [ ] +def type_narrow_ctx_arg(func: callable) -> callable: + # conserve annotations for frappe.utils.typing_validations + @wraps(func, assigned=(a for a in WRAPPER_ASSIGNMENTS if a != "__annotations__")) + def wrapper(ctx: ItemDetailsCtx | Document | dict | str, *args, **kwargs): + ctx: ItemDetailsCtx = parse_json(ctx) + if isinstance(ctx, Document): + ctx = ctx.as_dict() + + return func(ctx, *args, **kwargs) + + # set annotations from function + wrapper.__annotations__.update({k: v for k, v in func.__annotations__.items() if k != "ctx"}) + return wrapper + + +def _preprocess_ctx(ctx): + if not ctx.price_list: + ctx.price_list = ctx.selling_price_list or ctx.buying_price_list + + if not ctx.item_code and ctx.barcode: + ctx.item_code = get_item_code(barcode=ctx.barcode) + elif not ctx.item_code and ctx.serial_no: + ctx.item_code = get_item_code(serial_no=ctx.serial_no) + + set_transaction_type(ctx) + + @frappe.whitelist() -def get_item_details(args, doc=None, for_validate=False, overwrite_warehouse=True): +@type_narrow_ctx_arg +def get_item_details( + ctx: ItemDetailsCtx, doc=None, for_validate=False, overwrite_warehouse=True +) -> ItemDetails: """ - args = { + ctx = { "item_code": "", "warehouse": None, "customer": "", @@ -57,92 +93,90 @@ def get_item_details(args, doc=None, for_validate=False, overwrite_warehouse=Tru "set_warehouse": "" } """ - - args = process_args(args) - for_validate = process_string_args(for_validate) - overwrite_warehouse = process_string_args(overwrite_warehouse) - item = frappe.get_cached_doc("Item", args.item_code) - validate_item_details(args, item) + _preprocess_ctx(ctx) + for_validate = parse_json(for_validate) + overwrite_warehouse = parse_json(overwrite_warehouse) + item = frappe.get_cached_doc("Item", ctx.item_code) + validate_item_details(ctx, item) if isinstance(doc, str): doc = json.loads(doc) if doc: - args["transaction_date"] = doc.get("transaction_date") or doc.get("posting_date") + ctx.transaction_date = doc.get("transaction_date") or doc.get("posting_date") if doc.get("doctype") == "Purchase Invoice": - args["bill_date"] = doc.get("bill_date") + ctx.bill_date = doc.get("bill_date") - out = get_basic_details(args, item, overwrite_warehouse) - get_item_tax_template(args, item, out) - out["item_tax_rate"] = get_item_tax_map( - args.company, - args.get("item_tax_template") - if out.get("item_tax_template") is None - else out.get("item_tax_template"), + out: ItemDetails = get_basic_details(ctx, item, overwrite_warehouse) + + get_item_tax_template(ctx, item, out) + out.item_tax_rate = get_item_tax_map( + ctx.company, + out.item_tax_template or ctx.item_tax_template, as_json=True, ) - get_party_item_code(args, item, out) + get_party_item_code(ctx, item, out) - if args.get("doctype") in ["Sales Order", "Quotation"]: - set_valuation_rate(out, args) + if ctx.doctype in ["Sales Order", "Quotation"]: + set_valuation_rate(out, ctx) - update_party_blanket_order(args, out) + update_party_blanket_order(ctx, out) # Never try to find a customer price if customer is set in these Doctype - current_customer = args.customer - if args.get("doctype") in ["Purchase Order", "Purchase Receipt", "Purchase Invoice"]: - args.customer = None + current_customer = ctx.customer + if ctx.doctype in ["Purchase Order", "Purchase Receipt", "Purchase Invoice"]: + ctx.customer = None - out.update(get_price_list_rate(args, item)) + out.update(get_price_list_rate(ctx, item)) - args.customer = current_customer + ctx.customer = current_customer - if args.customer and cint(args.is_pos): - out.update(get_pos_profile_item_details(args.company, args, update_data=True)) + if ctx.customer and cint(ctx.is_pos): + out.update(get_pos_profile_item_details_(ctx, ctx.company, update_data=True)) if item.is_stock_item: - update_bin_details(args, out, doc) + update_bin_details(ctx, out, doc) - # update args with out, if key or value not exists + # update ctx with out, if key or value not exists for key, value in out.items(): - if args.get(key) is None: - args[key] = value + if ctx.get(key) is None: + ctx[key] = value - data = get_pricing_rule_for_item(args, doc=doc, for_validate=for_validate) + data = get_pricing_rule_for_item(ctx, doc=doc, for_validate=for_validate) out.update(data) - if args.transaction_date and item.lead_time_days: - out.schedule_date = out.lead_time_date = add_days(args.transaction_date, item.lead_time_days) + if ctx.transaction_date and item.lead_time_days: + out.schedule_date = out.lead_time_date = add_days(ctx.transaction_date, item.lead_time_days) - if args.get("is_subcontracted"): - out.bom = args.get("bom") or get_default_bom(args.item_code) + if ctx.is_subcontracted: + out.bom = ctx.bom or get_default_bom(ctx.item_code) get_gross_profit(out) - if args.doctype == "Material Request": - out.rate = args.rate or out.price_list_rate - out.amount = flt(args.qty) * flt(out.rate) + if ctx.doctype == "Material Request": + out.rate = ctx.rate or out.price_list_rate + out.amount = flt(ctx.qty) * flt(out.rate) out = remove_standard_fields(out) return out -def remove_standard_fields(details): +def remove_standard_fields(out: ItemDetails): for key in child_table_fields + default_fields: - details.pop(key, None) - return details + out.pop(key, None) + return out -def set_valuation_rate(out, args): - if frappe.db.exists("Product Bundle", {"name": args.item_code, "disabled": 0}, cache=True): +def set_valuation_rate(out: ItemDetails | dict, ctx: ItemDetailsCtx): + if frappe.db.exists("Product Bundle", {"name": ctx.item_code, "disabled": 0}, cache=True): valuation_rate = 0.0 - bundled_items = frappe.get_doc("Product Bundle", args.item_code) + bundled_items = frappe.get_doc("Product Bundle", ctx.item_code) for bundle_item in bundled_items.items: valuation_rate += flt( - get_valuation_rate(bundle_item.item_code, args.company, out.get("warehouse")).get( + get_valuation_rate(bundle_item.item_code, ctx.company, out.get("warehouse")).get( "valuation_rate" ) * bundle_item.qty @@ -151,46 +185,22 @@ def set_valuation_rate(out, args): out.update({"valuation_rate": valuation_rate}) else: - out.update(get_valuation_rate(args.item_code, args.company, out.get("warehouse"))) + out.update(get_valuation_rate(ctx.item_code, ctx.company, out.get("warehouse"))) -def update_bin_details(args, out, doc): - if args.get("doctype") == "Material Request" and args.get("material_request_type") == "Material Transfer": - out.update(get_bin_details(args.item_code, args.get("from_warehouse"))) +def update_bin_details(ctx: ItemDetailsCtx, out: ItemDetails, doc): + if ctx.doctype == "Material Request" and ctx.material_request_type == "Material Transfer": + out.update(get_bin_details(ctx.item_code, ctx.from_warehouse)) elif out.get("warehouse"): - company = args.company if (doc and doc.get("doctype") == "Purchase Order") else None + company = ctx.company if (doc and doc.get("doctype") == "Purchase Order") else None # calculate company_total_stock only for po - bin_details = get_bin_details(args.item_code, out.warehouse, company, include_child_warehouses=True) + bin_details = get_bin_details(ctx.item_code, out.warehouse, company, include_child_warehouses=True) out.update(bin_details) -def process_args(args): - if isinstance(args, str): - args = json.loads(args) - - args = frappe._dict(args) - - if not args.get("price_list"): - args.price_list = args.get("selling_price_list") or args.get("buying_price_list") - - if not args.item_code and args.barcode: - args.item_code = get_item_code(barcode=args.barcode) - elif not args.item_code and args.serial_no: - args.item_code = get_item_code(serial_no=args.serial_no) - - set_transaction_type(args) - return args - - -def process_string_args(args): - if isinstance(args, str): - args = json.loads(args) - return args - - def get_item_code(barcode=None, serial_no=None): if barcode: item_code = frappe.db.get_value("Item Barcode", {"barcode": barcode}, fieldname=["parent"]) @@ -204,8 +214,8 @@ def get_item_code(barcode=None, serial_no=None): return item_code -def validate_item_details(args, item): - if not args.company: +def validate_item_details(ctx: ItemDetailsCtx, item): + if not ctx.company: throw(_("Please specify Company")) from erpnext.stock.doctype.item.item import validate_end_of_life @@ -217,9 +227,9 @@ def validate_item_details(args, item): throw(_(msg), title=_("Template Item Selected")) - elif args.transaction_type == "buying" and args.doctype != "Material Request": - if args.get("is_subcontracted"): - if args.get("is_old_subcontracting_flow"): + elif ctx.transaction_type == "buying" and ctx.doctype != "Material Request": + if ctx.is_subcontracted: + if ctx.is_old_subcontracting_flow: if item.is_sub_contracted_item != 1: throw(_("Item {0} must be a Sub-contracted Item").format(item.name)) else: @@ -227,9 +237,9 @@ def validate_item_details(args, item): throw(_("Item {0} must be a Non-Stock Item").format(item.name)) -def get_basic_details(args, item, overwrite_warehouse=True): +def get_basic_details(ctx: ItemDetailsCtx, item, overwrite_warehouse=True) -> ItemDetails: """ - :param args: { + :param ctx: { "item_code": "", "warehouse": None, "customer": "", @@ -266,14 +276,14 @@ def get_basic_details(args, item, overwrite_warehouse=True): """ if not item: - item = frappe.get_doc("Item", args.get("item_code")) + item = frappe.get_doc("Item", ctx.item_code) if item.variant_of and not item.taxes and frappe.db.exists("Item Tax", {"parent": item.variant_of}): item.update_template_tables() - item_defaults = get_item_defaults(item.name, args.company) - item_group_defaults = get_item_group_defaults(item.name, args.company) - brand_defaults = get_brand_defaults(item.name, args.company) + item_defaults = get_item_defaults(item.name, ctx.company) + item_group_defaults = get_item_group_defaults(item.name, ctx.company) + brand_defaults = get_brand_defaults(item.name, ctx.company) defaults = frappe._dict( { @@ -283,11 +293,11 @@ def get_basic_details(args, item, overwrite_warehouse=True): } ) - warehouse = get_item_warehouse(item, args, overwrite_warehouse, defaults) + warehouse = get_item_warehouse_(ctx, item, overwrite_warehouse, defaults) - if args.get("doctype") == "Material Request" and not args.get("material_request_type"): - args["material_request_type"] = frappe.db.get_value( - "Material Request", args.get("name"), "material_request_type", cache=True + if ctx.doctype == "Material Request" and not ctx.material_request_type: + ctx["material_request_type"] = frappe.db.get_value( + "Material Request", ctx.name, "material_request_type", cache=True ) expense_account = None @@ -299,9 +309,9 @@ def get_basic_details(args, item, overwrite_warehouse=True): expense_account = get_asset_account( "capital_work_in_progress_account", asset_category=item.asset_category, - company=args.company, + company=ctx.company, ) - elif args.get("doctype") in ( + elif ctx.doctype in ( "Purchase Invoice", "Purchase Receipt", "Purchase Order", @@ -310,27 +320,27 @@ def get_basic_details(args, item, overwrite_warehouse=True): from erpnext.assets.doctype.asset_category.asset_category import get_asset_category_account expense_account = get_asset_category_account( - fieldname="fixed_asset_account", item=args.item_code, company=args.company + fieldname="fixed_asset_account", item=ctx.item_code, company=ctx.company ) # Set the UOM to the Default Sales UOM or Default Purchase UOM if configured in the Item Master - if not args.get("uom"): - if args.get("doctype") in sales_doctypes: - args.uom = item.sales_uom if item.sales_uom else item.stock_uom - elif (args.get("doctype") in ["Purchase Order", "Purchase Receipt", "Purchase Invoice"]) or ( - args.get("doctype") == "Material Request" and args.get("material_request_type") == "Purchase" + if not ctx.uom: + if ctx.doctype in sales_doctypes: + ctx.uom = item.sales_uom if item.sales_uom else item.stock_uom + elif (ctx.doctype in ["Purchase Order", "Purchase Receipt", "Purchase Invoice"]) or ( + ctx.doctype == "Material Request" and ctx.material_request_type == "Purchase" ): - args.uom = item.purchase_uom if item.purchase_uom else item.stock_uom + ctx.uom = item.purchase_uom if item.purchase_uom else item.stock_uom else: - args.uom = item.stock_uom + ctx.uom = item.stock_uom - # Set stock UOM in args, so that it can be used while fetching item price - args.stock_uom = item.stock_uom + # Set stock UOM in ctx, so that it can be used while fetching item price + ctx.stock_uom = item.stock_uom - if args.get("batch_no") and item.name != frappe.get_cached_value("Batch", args.get("batch_no"), "item"): - args["batch_no"] = "" + if ctx.batch_no and item.name != frappe.get_cached_value("Batch", ctx.batch_no, "item"): + ctx.batch_no = "" - out = frappe._dict( + out = ItemDetails( { "item_code": item.name, "item_name": item.item_name, @@ -338,25 +348,25 @@ def get_basic_details(args, item, overwrite_warehouse=True): "image": cstr(item.image).strip(), "warehouse": warehouse, "income_account": get_default_income_account( - args, item_defaults, item_group_defaults, brand_defaults + ctx, item_defaults, item_group_defaults, brand_defaults ), "expense_account": expense_account - or get_default_expense_account(args, item_defaults, item_group_defaults, brand_defaults), + or get_default_expense_account(ctx, item_defaults, item_group_defaults, brand_defaults), "discount_account": get_default_discount_account( - args, item_defaults, item_group_defaults, brand_defaults + ctx, item_defaults, item_group_defaults, brand_defaults ), "provisional_expense_account": get_provisional_account( - args, item_defaults, item_group_defaults, brand_defaults + ctx, item_defaults, item_group_defaults, brand_defaults ), - "cost_center": get_default_cost_center(args, item_defaults, item_group_defaults, brand_defaults), + "cost_center": get_default_cost_center(ctx, item_defaults, item_group_defaults, brand_defaults), "has_serial_no": item.has_serial_no, "has_batch_no": item.has_batch_no, - "batch_no": args.get("batch_no"), - "uom": args.uom, + "batch_no": ctx.batch_no, + "uom": ctx.uom, "stock_uom": item.stock_uom, - "min_order_qty": flt(item.min_order_qty) if args.doctype == "Material Request" else "", - "qty": flt(args.qty) or 1.0, - "stock_qty": flt(args.qty) or 1.0, + "min_order_qty": flt(item.min_order_qty) if ctx.doctype == "Material Request" else "", + "qty": flt(ctx.qty) or 1.0, + "stock_qty": flt(ctx.qty) or 1.0, "price_list_rate": 0.0, "base_price_list_rate": 0.0, "rate": 0.0, @@ -366,51 +376,49 @@ def get_basic_details(args, item, overwrite_warehouse=True): "net_rate": 0.0, "net_amount": 0.0, "discount_percentage": 0.0, - "discount_amount": flt(args.discount_amount) or 0.0, - "update_stock": args.get("update_stock") - if args.get("doctype") in ["Sales Invoice", "Purchase Invoice"] - else 0, + "discount_amount": flt(ctx.discount_amount) or 0.0, + "update_stock": ctx.update_stock if ctx.doctype in ["Sales Invoice", "Purchase Invoice"] else 0, "delivered_by_supplier": item.delivered_by_supplier - if args.get("doctype") in ["Sales Order", "Sales Invoice"] + if ctx.doctype in ["Sales Order", "Sales Invoice"] else 0, "is_fixed_asset": item.is_fixed_asset, - "last_purchase_rate": item.last_purchase_rate if args.get("doctype") in ["Purchase Order"] else 0, - "transaction_date": args.get("transaction_date"), - "against_blanket_order": args.get("against_blanket_order"), + "last_purchase_rate": item.last_purchase_rate if ctx.doctype in ["Purchase Order"] else 0, + "transaction_date": ctx.transaction_date, + "against_blanket_order": ctx.against_blanket_order, "bom_no": item.get("default_bom"), - "weight_per_unit": args.get("weight_per_unit") or item.get("weight_per_unit"), - "weight_uom": args.get("weight_uom") or item.get("weight_uom"), + "weight_per_unit": ctx.weight_per_unit or item.get("weight_per_unit"), + "weight_uom": ctx.weight_uom or item.get("weight_uom"), "grant_commission": item.get("grant_commission"), } ) - default_supplier = get_default_supplier(args, item_defaults, item_group_defaults, brand_defaults) + default_supplier = get_default_supplier(ctx, item_defaults, item_group_defaults, brand_defaults) if default_supplier: out.supplier = default_supplier if item.get("enable_deferred_revenue") or item.get("enable_deferred_expense"): - out.update(calculate_service_end_date(args, item)) + out.update(calculate_service_end_date(ctx, item)) # calculate conversion factor - if item.stock_uom == args.uom: + if item.stock_uom == ctx.uom: out.conversion_factor = 1.0 else: - out.conversion_factor = args.conversion_factor or get_conversion_factor(item.name, args.uom).get( + out.conversion_factor = ctx.conversion_factor or get_conversion_factor(item.name, ctx.uom).get( "conversion_factor" ) - args.conversion_factor = out.conversion_factor + ctx.conversion_factor = out.conversion_factor out.stock_qty = out.qty * out.conversion_factor - args.stock_qty = out.stock_qty + ctx.stock_qty = out.stock_qty # calculate last purchase rate - if args.get("doctype") in purchase_doctypes and not frappe.db.get_single_value( + if ctx.doctype in purchase_doctypes and not frappe.db.get_single_value( "Buying Settings", "disable_last_purchase_rate" ): from erpnext.buying.doctype.purchase_order.purchase_order import item_last_purchase_rate out.last_purchase_rate = item_last_purchase_rate( - args.name, args.conversion_rate, item.name, out.conversion_factor + ctx.name, ctx.conversion_rate, item.name, out.conversion_factor ) # if default specified in item is for another company, fetch from company @@ -421,18 +429,18 @@ def get_basic_details(args, item, overwrite_warehouse=True): ["Warehouse", "warehouse", ""], ]: if not out[d[1]]: - out[d[1]] = frappe.get_cached_value("Company", args.company, d[2]) if d[2] else None + out[d[1]] = frappe.get_cached_value("Company", ctx.company, d[2]) if d[2] else None for fieldname in ("item_name", "item_group", "brand", "stock_uom"): out[fieldname] = item.get(fieldname) - if args.get("manufacturer"): - part_no = get_item_manufacturer_part_no(args.get("item_code"), args.get("manufacturer")) + if ctx.manufacturer: + part_no = get_item_manufacturer_part_no(ctx.item_code, ctx.manufacturer) if part_no: - out["manufacturer_part_no"] = part_no + out.manufacturer_part_no = part_no else: - out["manufacturer_part_no"] = None - out["manufacturer"] = None + out.manufacturer_part_no = None + out.manufacturer = None else: data = frappe.get_value( "Item", item.name, ["default_item_manufacturer", "default_manufacturer_part_no"], as_dict=1 @@ -446,50 +454,54 @@ def get_basic_details(args, item, overwrite_warehouse=True): } ) - child_doctype = args.doctype + " Item" + child_doctype = ctx.doctype + " Item" meta = frappe.get_meta(child_doctype) if meta.get_field("barcode"): update_barcode_value(out) - if out.get("weight_per_unit"): - out["total_weight"] = out.weight_per_unit * out.stock_qty + if out.weight_per_unit: + out.total_weight = out.weight_per_unit * out.stock_qty return out -def get_item_warehouse(item, args, overwrite_warehouse, defaults=None): +from erpnext.deprecation_dumpster import get_item_warehouse + + +@type_narrow_ctx_arg +def get_item_warehouse_(ctx: ItemDetailsCtx, item, overwrite_warehouse, defaults=None): if not defaults: defaults = frappe._dict( { - "item_defaults": get_item_defaults(item.name, args.company), - "item_group_defaults": get_item_group_defaults(item.name, args.company), - "brand_defaults": get_brand_defaults(item.name, args.company), + "item_defaults": get_item_defaults(item.name, ctx.company), + "item_group_defaults": get_item_group_defaults(item.name, ctx.company), + "brand_defaults": get_brand_defaults(item.name, ctx.company), } ) - if overwrite_warehouse or not args.warehouse: + if overwrite_warehouse or not ctx.warehouse: warehouse = ( - args.get("set_warehouse") + ctx.set_warehouse or defaults.item_defaults.get("default_warehouse") or defaults.item_group_defaults.get("default_warehouse") or defaults.brand_defaults.get("default_warehouse") - or args.get("warehouse") + or ctx.warehouse ) if not warehouse: defaults = frappe.defaults.get_defaults() or {} warehouse_exists = frappe.db.exists( - "Warehouse", {"name": defaults.default_warehouse, "company": args.company} + "Warehouse", {"name": defaults.default_warehouse, "company": ctx.company} ) if defaults.get("default_warehouse") and warehouse_exists: warehouse = defaults.default_warehouse else: - warehouse = args.get("warehouse") + warehouse = ctx.warehouse if not warehouse: default_warehouse = frappe.db.get_single_value("Stock Settings", "default_warehouse") - if frappe.db.get_value("Warehouse", default_warehouse, "company") == args.company: + if frappe.db.get_value("Warehouse", default_warehouse, "company") == ctx.company: return default_warehouse return warehouse @@ -554,16 +566,16 @@ def get_item_tax_info(company, tax_category, item_codes, item_rates=None, item_t out[item_code[1]] = {} item = frappe.get_cached_doc("Item", item_code[0]) - args = { + ctx: ItemDetailsCtx = { "company": company, "tax_category": tax_category, "base_net_rate": item_rates.get(item_code[1]), } if item_tax_templates: - args.update({"item_tax_template": item_tax_templates.get(item_code[1])}) + ctx.update({"item_tax_template": item_tax_templates.get(item_code[1])}) - get_item_tax_template(args, item, out[item_code[1]]) + get_item_tax_template(ctx, item, out[item_code[1]]) out[item_code[1]]["item_tax_rate"] = get_item_tax_map( company, out[item_code[1]].get("item_tax_template"), as_json=True ) @@ -571,43 +583,72 @@ def get_item_tax_info(company, tax_category, item_codes, item_rates=None, item_t return out -def get_item_tax_template(args, item, out): +@type_narrow_ctx_arg +def get_item_tax_template(ctx: ItemDetailsCtx, item, out: ItemDetails): """ - args = { + Determines item_tax template from item or parent item groups. + + Accesses: + ctx = { + "child_doctype": str + } + Passes: + ctx = { + "company": str + "bill_date": str + "transaction_date": str "tax_category": None "item_tax_template": None - } + "base_net_rate": float + } """ item_tax_template = None if item.taxes: - item_tax_template = _get_item_tax_template(args, item.taxes, out) + item_tax_template = _get_item_tax_template(ctx, item.taxes, out) if not item_tax_template: item_group = item.item_group while item_group and not item_tax_template: item_group_doc = frappe.get_cached_doc("Item Group", item_group) - item_tax_template = _get_item_tax_template(args, item_group_doc.taxes, out) + item_tax_template = _get_item_tax_template(ctx, item_group_doc.taxes, out) item_group = item_group_doc.parent_item_group - if args.get("child_doctype") and item_tax_template: - out.update(get_fetch_values(args.get("child_doctype"), "item_tax_template", item_tax_template)) + if ctx.get("child_doctype") and item_tax_template: + out.update(get_fetch_values(ctx.get("child_doctype"), "item_tax_template", item_tax_template)) -def _get_item_tax_template(args, taxes, out=None, for_validate=False): +@type_narrow_ctx_arg +def _get_item_tax_template( + ctx: ItemDetailsCtx, taxes, out: ItemDetails | None = None, for_validate=False +) -> None | str | list[str]: + """ + Accesses: + ctx = { + "company": str + "bill_date": str + "transaction_date": str + "tax_category": None + "item_tax_template": None + } + Passes: + ctx = { + "base_net_rate": float + } + """ if out is None: - out = {} + out = ItemDetails() taxes_with_validity = [] taxes_with_no_validity = [] for tax in taxes: tax_company = frappe.get_cached_value("Item Tax Template", tax.item_tax_template, "company") - if tax_company == args["company"]: + if tax_company == ctx["company"]: if tax.valid_from or tax.maximum_net_rate: # In purchase Invoice first preference will be given to supplier invoice date # if supplier date is not present then posting date - validation_date = args.get("bill_date") or args.get("transaction_date") + validation_date = ctx.get("bill_date") or ctx.get("transaction_date") - if getdate(tax.valid_from) <= getdate(validation_date) and is_within_valid_range(args, tax): + if getdate(tax.valid_from) <= getdate(validation_date) and is_within_valid_range(ctx, tax): taxes_with_validity.append(tax) else: taxes_with_no_validity.append(tax) @@ -622,7 +663,7 @@ def _get_item_tax_template(args, taxes, out=None, for_validate=False): tax.item_tax_template for tax in taxes if ( - cstr(tax.tax_category) == cstr(args.get("tax_category")) + cstr(tax.tax_category) == cstr(ctx.get("tax_category")) and (tax.item_tax_template not in taxes) ) ] @@ -632,22 +673,30 @@ def _get_item_tax_template(args, taxes, out=None, for_validate=False): return None # do not change if already a valid template - if args.get("item_tax_template") in {t.item_tax_template for t in taxes}: - out["item_tax_template"] = args.get("item_tax_template") - return args.get("item_tax_template") + if ctx.get("item_tax_template") in {t.item_tax_template for t in taxes}: + out.item_tax_template = ctx.get("item_tax_template") + return ctx.get("item_tax_template") for tax in taxes: - if cstr(tax.tax_category) == cstr(args.get("tax_category")): - out["item_tax_template"] = tax.item_tax_template + if cstr(tax.tax_category) == cstr(ctx.get("tax_category")): + out.item_tax_template = tax.item_tax_template return tax.item_tax_template return None -def is_within_valid_range(args, tax): +@type_narrow_ctx_arg +def is_within_valid_range(ctx: ItemDetailsCtx, tax) -> bool: + """ + Accesses: + ctx = { + "base_net_rate": float + } + """ + if not flt(tax.maximum_net_rate): # No range specified, just ignore return True - elif flt(tax.minimum_net_rate) <= flt(args.get("base_net_rate")) <= flt(tax.maximum_net_rate): + elif flt(tax.minimum_net_rate) <= flt(ctx.get("base_net_rate")) <= flt(tax.maximum_net_rate): return True return False @@ -666,12 +715,13 @@ def get_item_tax_map(company, item_tax_template, as_json=True): @frappe.whitelist() -def calculate_service_end_date(args, item=None): - args = process_args(args) +@type_narrow_ctx_arg +def calculate_service_end_date(ctx: ItemDetailsCtx, item=None): + _preprocess_ctx(ctx) if not item: - item = frappe.get_cached_doc("Item", args.item_code) + item = frappe.get_cached_doc("Item", ctx.item_code) - doctype = args.get("parenttype") or args.get("doctype") + doctype = ctx.parenttype or ctx.doctype if doctype == "Sales Invoice": enable_deferred = "enable_deferred_revenue" no_of_months = "no_of_months" @@ -681,77 +731,78 @@ def calculate_service_end_date(args, item=None): no_of_months = "no_of_months_exp" account = "deferred_expense_account" - service_start_date = args.service_start_date if args.service_start_date else args.transaction_date + service_start_date = ctx.service_start_date if ctx.service_start_date else ctx.transaction_date service_end_date = add_months(service_start_date, item.get(no_of_months)) deferred_detail = {"service_start_date": service_start_date, "service_end_date": service_end_date} deferred_detail[enable_deferred] = item.get(enable_deferred) - deferred_detail[account] = get_default_deferred_account(args, item, fieldname=account) + deferred_detail[account] = get_default_deferred_account(ctx, item, fieldname=account) return deferred_detail -def get_default_income_account(args, item, item_group, brand): +def get_default_income_account(ctx: ItemDetailsCtx, item, item_group, brand): return ( item.get("income_account") or item_group.get("income_account") or brand.get("income_account") - or args.income_account + or ctx.income_account ) -def get_default_expense_account(args, item, item_group, brand): +def get_default_expense_account(ctx: ItemDetailsCtx, item, item_group, brand): return ( item.get("expense_account") or item_group.get("expense_account") or brand.get("expense_account") - or args.expense_account + or ctx.expense_account ) -def get_provisional_account(args, item, item_group, brand): +def get_provisional_account(ctx: ItemDetailsCtx, item, item_group, brand): return ( item.get("default_provisional_account") or item_group.get("default_provisional_account") or brand.get("default_provisional_account") - or args.default_provisional_account + or ctx.default_provisional_account ) -def get_default_discount_account(args, item, item_group, brand): +def get_default_discount_account(ctx: ItemDetailsCtx, item, item_group, brand): return ( item.get("default_discount_account") or item_group.get("default_discount_account") or brand.get("default_discount_account") - or args.discount_account + or ctx.discount_account ) -def get_default_deferred_account(args, item, fieldname=None): +def get_default_deferred_account(ctx: ItemDetailsCtx, item, fieldname=None): if item.get("enable_deferred_revenue") or item.get("enable_deferred_expense"): return ( frappe.get_cached_value( "Item Default", - {"parent": args.item_code, "company": args.get("company")}, + {"parent": ctx.item_code, "company": ctx.company}, fieldname, ) - or args.get(fieldname) - or frappe.get_cached_value("Company", args.company, "default_" + fieldname) + or ctx.get(fieldname) + or frappe.get_cached_value("Company", ctx.company, "default_" + fieldname) ) else: return None -def get_default_cost_center(args, item=None, item_group=None, brand=None, company=None): +@type_narrow_ctx_arg +def get_default_cost_center(ctx: ItemDetailsCtx, item=None, item_group=None, brand=None, company=None): cost_center = None - if not company and args.get("company"): - company = args.get("company") + if not company and ctx.get("company"): + company = ctx.get("company") - if args.get("project"): - cost_center = frappe.db.get_value("Project", args.get("project"), "cost_center", cache=True) + if ctx.get("project"): + cost_center = frappe.db.get_value("Project", ctx.get("project"), "cost_center", cache=True) if not cost_center and (item and item_group and brand): - if args.get("customer"): + if ctx.get("customer"): cost_center = ( item.get("selling_cost_center") or item_group.get("selling_cost_center") @@ -764,22 +815,22 @@ def get_default_cost_center(args, item=None, item_group=None, brand=None, compan or brand.get("buying_cost_center") ) - elif not cost_center and args.get("item_code") and company: + elif not cost_center and ctx.get("item_code") and company: for method in ["get_item_defaults", "get_item_group_defaults", "get_brand_defaults"]: path = f"erpnext.stock.get_item_details.{method}" - data = frappe.get_attr(path)(args.get("item_code"), company) + data = frappe.get_attr(path)(ctx.get("item_code"), company) if data and (data.selling_cost_center or data.buying_cost_center): - if args.get("customer") and data.selling_cost_center: + if ctx.get("customer") and data.selling_cost_center: return data.selling_cost_center - elif args.get("supplier") and data.buying_cost_center: + elif ctx.get("supplier") and data.buying_cost_center: return data.buying_cost_center return data.selling_cost_center or data.buying_cost_center - if not cost_center and args.get("cost_center"): - cost_center = args.get("cost_center") + if not cost_center and ctx.get("cost_center"): + cost_center = ctx.get("cost_center") if company and cost_center and frappe.get_cached_value("Cost Center", cost_center, "company") != company: return None @@ -790,85 +841,76 @@ def get_default_cost_center(args, item=None, item_group=None, brand=None, compan return cost_center -def get_default_supplier(args, item, item_group, brand): +def get_default_supplier(_ctx: ItemDetailsCtx, item, item_group, brand): return item.get("default_supplier") or item_group.get("default_supplier") or brand.get("default_supplier") -def get_price_list_rate(args, item_doc, out=None): +def get_price_list_rate(ctx: ItemDetailsCtx, item_doc, out: ItemDetails = None): if out is None: - out = frappe._dict() + out = ItemDetails() - meta = frappe.get_meta(args.parenttype or args.doctype) + meta = frappe.get_meta(ctx.parenttype or ctx.doctype) - if meta.get_field("currency") or args.get("currency"): - if not args.get("price_list_currency") or not args.get("plc_conversion_rate"): + if meta.get_field("currency") or ctx.get("currency"): + if not ctx.price_list_currency or not ctx.plc_conversion_rate: # if currency and plc_conversion_rate exist then # `get_price_list_currency_and_exchange_rate` has already been called - pl_details = get_price_list_currency_and_exchange_rate(args) - args.update(pl_details) + pl_details = get_price_list_currency_and_exchange_rate(ctx) + ctx.update(pl_details) if meta.get_field("currency"): - validate_conversion_rate(args, meta) + validate_conversion_rate(ctx, meta) - price_list_rate = get_price_list_rate_for(args, item_doc.name) + price_list_rate = get_price_list_rate_for(ctx, item_doc.name) # variant if price_list_rate is None and item_doc.variant_of: - price_list_rate = get_price_list_rate_for(args, item_doc.variant_of) + price_list_rate = get_price_list_rate_for(ctx, item_doc.variant_of) # insert in database if price_list_rate is None or frappe.db.get_single_value( "Stock Settings", "update_existing_price_list_rate" ): - insert_item_price(args) + insert_item_price(ctx) if price_list_rate is None: return out - out.price_list_rate = flt(price_list_rate) * flt(args.plc_conversion_rate) / flt(args.conversion_rate) + out.price_list_rate = flt(price_list_rate) * flt(ctx.plc_conversion_rate) / flt(ctx.conversion_rate) if frappe.db.get_single_value("Buying Settings", "disable_last_purchase_rate"): return out - if ( - not args.get("is_internal_supplier") - and not out.price_list_rate - and args.transaction_type == "buying" - ): + if not ctx.is_internal_supplier and not out.price_list_rate and ctx.transaction_type == "buying": from erpnext.stock.doctype.item.item import get_last_purchase_details - out.update(get_last_purchase_details(item_doc.name, args.name, args.conversion_rate)) + out.update(get_last_purchase_details(item_doc.name, ctx.name, ctx.conversion_rate)) return out -def insert_item_price(args): +def insert_item_price(ctx: ItemDetailsCtx): """Insert Item Price if Price List and Price List Rate are specified and currency is the same""" - if ( - not args.price_list - or not args.rate - or args.get("is_internal_supplier") - or args.get("is_internal_customer") - ): + if not ctx.price_list or not ctx.rate or ctx.is_internal_supplier or ctx.is_internal_customer: return - if frappe.db.get_value("Price List", args.price_list, "currency", cache=True) == args.currency and cint( + if frappe.db.get_value("Price List", ctx.price_list, "currency", cache=True) == ctx.currency and cint( frappe.db.get_single_value("Stock Settings", "auto_insert_price_list_rate_if_missing") ): if frappe.has_permission("Item Price", "write"): price_list_rate = ( - (flt(args.rate) + flt(args.discount_amount)) / args.get("conversion_factor") - if args.get("conversion_factor") - else (flt(args.rate) + flt(args.discount_amount)) + (flt(ctx.rate) + flt(ctx.discount_amount)) / ctx.conversion_factor + if ctx.conversion_factor + else (flt(ctx.rate) + flt(ctx.discount_amount)) ) item_price = frappe.db.get_value( "Item Price", { - "item_code": args.item_code, - "price_list": args.price_list, - "currency": args.currency, - "uom": args.stock_uom, + "item_code": ctx.item_code, + "price_list": ctx.price_list, + "currency": ctx.currency, + "uom": ctx.stock_uom, }, ["name", "price_list_rate"], as_dict=1, @@ -880,7 +922,7 @@ def insert_item_price(args): frappe.db.set_value("Item Price", item_price.name, "price_list_rate", price_list_rate) frappe.msgprint( _("Item Price updated for {0} in Price List {1}").format( - args.item_code, args.price_list + ctx.item_code, ctx.price_list ), alert=True, ) @@ -888,28 +930,31 @@ def insert_item_price(args): item_price = frappe.get_doc( { "doctype": "Item Price", - "price_list": args.price_list, - "item_code": args.item_code, - "currency": args.currency, + "price_list": ctx.price_list, + "item_code": ctx.item_code, + "currency": ctx.currency, "price_list_rate": price_list_rate, - "uom": args.stock_uom, + "uom": ctx.stock_uom, } ) item_price.insert() frappe.msgprint( - _("Item Price added for {0} in Price List {1}").format(args.item_code, args.price_list), + _("Item Price added for {0} in Price List {1}").format(ctx.item_code, ctx.price_list), alert=True, ) -def get_item_price(args, item_code, ignore_party=False, force_batch_no=False) -> list[dict]: +def get_item_price( + pctx: ItemPriceCtx | dict, item_code, ignore_party=False, force_batch_no=False +) -> list[dict]: """ Get name, price_list_rate from Item Price based on conditions Check if the desired qty is within the increment of the packing list. - :param args: dict (or frappe._dict) with mandatory fields price_list, uom + :param pctx: dict (or frappe._dict) with mandatory fields price_list, uom optional fields transaction_date, customer, supplier :param item_code: str, Item Doctype field item_code """ + pctx: ItemPriceCtx = frappe._dict(pctx) ip = frappe.qb.DocType("Item Price") query = ( @@ -917,8 +962,8 @@ def get_item_price(args, item_code, ignore_party=False, force_batch_no=False) -> .select(ip.name, ip.price_list_rate, ip.uom) .where( (ip.item_code == item_code) - & (ip.price_list == args.get("price_list")) - & (IfNull(ip.uom, "").isin(["", args.get("uom")])) + & (ip.price_list == pctx.price_list) + & (IfNull(ip.uom, "").isin(["", pctx.uom])) ) .orderby(ip.valid_from, order=frappe.qb.desc) .orderby(IfNull(ip.batch_no, ""), order=frappe.qb.desc) @@ -927,43 +972,43 @@ def get_item_price(args, item_code, ignore_party=False, force_batch_no=False) -> ) if force_batch_no: - query = query.where(ip.batch_no == args.get("batch_no")) + query = query.where(ip.batch_no == pctx.batch_no) else: - query = query.where(IfNull(ip.batch_no, "").isin(["", args.get("batch_no")])) + query = query.where(IfNull(ip.batch_no, "").isin(["", pctx.batch_no])) if not ignore_party: - if args.get("customer"): - query = query.where(ip.customer == args.get("customer")) - elif args.get("supplier"): - query = query.where(ip.supplier == args.get("supplier")) + if pctx.customer: + query = query.where(ip.customer == pctx.customer) + elif pctx.supplier: + query = query.where(ip.supplier == pctx.supplier) else: query = query.where((IfNull(ip.customer, "") == "") & (IfNull(ip.supplier, "") == "")) - if args.get("transaction_date"): + if pctx.transaction_date: query = query.where( - (IfNull(ip.valid_from, "2000-01-01") <= args["transaction_date"]) - & (IfNull(ip.valid_upto, "2500-12-31") >= args["transaction_date"]) + (IfNull(ip.valid_from, "2000-01-01") <= pctx.transaction_date) + & (IfNull(ip.valid_upto, "2500-12-31") >= pctx.transaction_date) ) return query.run(as_dict=True) @frappe.whitelist() -def get_batch_based_item_price(params, item_code) -> float: - if isinstance(params, str): - params = parse_json(params) +def get_batch_based_item_price(pctx: ItemPriceCtx | dict | str, item_code) -> float: + pctx = parse_json(pctx) - item_price = get_item_price(params, item_code, force_batch_no=True) + item_price = get_item_price(pctx, item_code, force_batch_no=True) if not item_price: - item_price = get_item_price(params, item_code, ignore_party=True, force_batch_no=True) + item_price = get_item_price(pctx, item_code, ignore_party=True, force_batch_no=True) - if item_price and item_price[0].uom == params.get("uom"): + if item_price and item_price[0].uom == pctx.uom: return item_price[0].price_list_rate return 0.0 -def get_price_list_rate_for(args, item_code): +@type_narrow_ctx_arg +def get_price_list_rate_for(ctx: ItemDetailsCtx, item_code): """ :param customer: link to Customer DocType :param supplier: link to Supplier DocType @@ -972,44 +1017,42 @@ def get_price_list_rate_for(args, item_code): :param qty: Desired Qty :param transaction_date: Date of the price """ - item_price_args = { - "item_code": item_code, - "price_list": args.get("price_list"), - "customer": args.get("customer"), - "supplier": args.get("supplier"), - "uom": args.get("uom"), - "transaction_date": args.get("transaction_date"), - "batch_no": args.get("batch_no"), - } + pctx = ItemPriceCtx( + { + "item_code": item_code, + "price_list": ctx.get("price_list"), + "customer": ctx.get("customer"), + "supplier": ctx.get("supplier"), + "uom": ctx.get("uom"), + "transaction_date": ctx.get("transaction_date"), + "batch_no": ctx.get("batch_no"), + } + ) item_price_data = 0 - price_list_rate = get_item_price(item_price_args, item_code) + price_list_rate = get_item_price(pctx, item_code) if price_list_rate: - desired_qty = args.get("qty") + desired_qty = ctx.get("qty") if desired_qty and check_packing_list(price_list_rate[0].name, desired_qty, item_code): item_price_data = price_list_rate else: for field in ["customer", "supplier"]: - del item_price_args[field] + del pctx[field] - general_price_list_rate = get_item_price( - item_price_args, item_code, ignore_party=args.get("ignore_party") - ) + general_price_list_rate = get_item_price(pctx, item_code, ignore_party=ctx.get("ignore_party")) - if not general_price_list_rate and args.get("uom") != args.get("stock_uom"): - item_price_args["uom"] = args.get("stock_uom") - general_price_list_rate = get_item_price( - item_price_args, item_code, ignore_party=args.get("ignore_party") - ) + if not general_price_list_rate and ctx.get("uom") != ctx.get("stock_uom"): + pctx.uom = ctx.get("stock_uom") + general_price_list_rate = get_item_price(pctx, item_code, ignore_party=ctx.get("ignore_party")) if general_price_list_rate: item_price_data = general_price_list_rate if item_price_data: - if item_price_data[0].uom == args.get("uom"): + if item_price_data[0].uom == ctx.get("uom"): return item_price_data[0].price_list_rate - elif not args.get("price_list_uom_dependant"): - return flt(item_price_data[0].price_list_rate * flt(args.get("conversion_factor", 1))) + elif not ctx.get("price_list_uom_dependant"): + return flt(item_price_data[0].price_list_rate * flt(ctx.get("conversion_factor", 1))) else: return item_price_data[0].price_list_rate @@ -1034,91 +1077,93 @@ def check_packing_list(price_list_rate_name, desired_qty, item_code): return flag -def validate_conversion_rate(args, meta): +def validate_conversion_rate(ctx: ItemDetailsCtx, meta): from erpnext.controllers.accounts_controller import validate_conversion_rate - company_currency = frappe.get_cached_value("Company", args.company, "default_currency") - if not args.conversion_rate and args.currency == company_currency: - args.conversion_rate = 1.0 + company_currency = frappe.get_cached_value("Company", ctx.company, "default_currency") + if not ctx.conversion_rate and ctx.currency == company_currency: + ctx.conversion_rate = 1.0 - if not args.ignore_conversion_rate and args.conversion_rate == 1 and args.currency != company_currency: - args.conversion_rate = ( - get_exchange_rate(args.currency, company_currency, args.transaction_date, "for_buying") or 1.0 + if not ctx.ignore_conversion_rate and ctx.conversion_rate == 1 and ctx.currency != company_currency: + ctx.conversion_rate = ( + get_exchange_rate(ctx.currency, company_currency, ctx.transaction_date, "for_buying") or 1.0 ) # validate currency conversion rate validate_conversion_rate( - args.currency, args.conversion_rate, meta.get_label("conversion_rate"), args.company + ctx.currency, ctx.conversion_rate, meta.get_label("conversion_rate"), ctx.company ) - args.conversion_rate = flt( - args.conversion_rate, - get_field_precision(meta.get_field("conversion_rate"), frappe._dict({"fields": args})), + ctx.conversion_rate = flt( + ctx.conversion_rate, + get_field_precision(meta.get_field("conversion_rate"), frappe._dict({"fields": ctx})), ) - if args.price_list: - if not args.plc_conversion_rate and args.price_list_currency == frappe.db.get_value( - "Price List", args.price_list, "currency", cache=True + if ctx.price_list: + if not ctx.plc_conversion_rate and ctx.price_list_currency == frappe.db.get_value( + "Price List", ctx.price_list, "currency", cache=True ): - args.plc_conversion_rate = 1.0 + ctx.plc_conversion_rate = 1.0 # validate price list currency conversion rate - if not args.get("price_list_currency"): + if not ctx.price_list_currency: throw(_("Price List Currency not selected")) else: validate_conversion_rate( - args.price_list_currency, - args.plc_conversion_rate, + ctx.price_list_currency, + ctx.plc_conversion_rate, meta.get_label("plc_conversion_rate"), - args.company, + ctx.company, ) if meta.get_field("plc_conversion_rate"): - args.plc_conversion_rate = flt( - args.plc_conversion_rate, - get_field_precision( - meta.get_field("plc_conversion_rate"), frappe._dict({"fields": args}) - ), + ctx.plc_conversion_rate = flt( + ctx.plc_conversion_rate, + get_field_precision(meta.get_field("plc_conversion_rate"), frappe._dict({"fields": ctx})), ) -def get_party_item_code(args, item_doc, out): - if args.transaction_type == "selling" and args.customer: +def get_party_item_code(ctx: ItemDetailsCtx, item_doc, out: ItemDetails): + if ctx.transaction_type == "selling" and ctx.customer: out.customer_item_code = None - if args.quotation_to and args.quotation_to != "Customer": + if ctx.quotation_to and ctx.quotation_to != "Customer": return - customer_item_code = item_doc.get("customer_items", {"customer_name": args.customer}) + customer_item_code = item_doc.get("customer_items", {"customer_name": ctx.customer}) if customer_item_code: out.customer_item_code = customer_item_code[0].ref_code else: - customer_group = frappe.get_cached_value("Customer", args.customer, "customer_group") + customer_group = frappe.get_cached_value("Customer", ctx.customer, "customer_group") customer_group_item_code = item_doc.get("customer_items", {"customer_group": customer_group}) if customer_group_item_code and not customer_group_item_code[0].customer_name: out.customer_item_code = customer_group_item_code[0].ref_code - if args.transaction_type == "buying" and args.supplier: - item_supplier = item_doc.get("supplier_items", {"supplier": args.supplier}) + if ctx.transaction_type == "buying" and ctx.supplier: + item_supplier = item_doc.get("supplier_items", {"supplier": ctx.supplier}) out.supplier_part_no = item_supplier[0].supplier_part_no if item_supplier else None -def get_pos_profile_item_details(company, args, pos_profile=None, update_data=False): +from erpnext.deprecation_dumpster import get_pos_profile_item_details + + +@type_narrow_ctx_arg +def get_pos_profile_item_details_(ctx: ItemDetailsCtx, company, pos_profile=None, update_data=False): res = frappe._dict() if not frappe.flags.pos_profile and not pos_profile: - pos_profile = frappe.flags.pos_profile = get_pos_profile(company, args.get("pos_profile")) + pos_profile = frappe.flags.pos_profile = get_pos_profile(company, ctx.pos_profile) if pos_profile: for fieldname in ("income_account", "cost_center", "warehouse", "expense_account"): - if (not args.get(fieldname) or update_data) and pos_profile.get(fieldname): + if (not ctx.get(fieldname) or update_data) and pos_profile.get(fieldname): res[fieldname] = pos_profile.get(fieldname) if res.get("warehouse"): - res.actual_qty = get_bin_details( - args.item_code, res.warehouse, include_child_warehouses=True - ).get("actual_qty") + res.actual_qty = get_bin_details(ctx.item_code, res.warehouse, include_child_warehouses=True).get( + "actual_qty" + ) return res @@ -1233,14 +1278,15 @@ def get_batch_qty(batch_no, warehouse, item_code): @frappe.whitelist() -def apply_price_list(args, as_doc=False, doc=None): +@type_narrow_ctx_arg +def apply_price_list(ctx: ItemDetailsCtx, as_doc=False, doc=None): """Apply pricelist on a document-like dict object and return as {'parent': dict, 'children': list} - :param args: See below + :param ctx: See below :param as_doc: Updates value in the passed dict - args = { + ctx = { "doctype": "", "name": "", "items": [{"doctype": "", "name": "", "item_code": "", "brand": "", "item_group": ""}, ...], @@ -1258,71 +1304,68 @@ def apply_price_list(args, as_doc=False, doc=None): "ignore_pricing_rule": 0/1 } """ - args = process_args(args) - - parent = get_price_list_currency_and_exchange_rate(args) - args.update(parent) + _preprocess_ctx(ctx) + parent = get_price_list_currency_and_exchange_rate(ctx) + ctx.update(parent) children = [] - if "items" in args: - item_list = args.get("items") - args.update(parent) + if "items" in ctx: + item_list = ctx.items + ctx.update(parent) for item in item_list: - args_copy = frappe._dict(args.copy()) - args_copy.update(item) - item_details = apply_price_list_on_item(args_copy, doc=doc) + ctx_copy = ItemDetailsCtx(ctx.copy()) + ctx_copy.update(item) + item_details = apply_price_list_on_item(ctx_copy, doc=doc) children.append(item_details) if as_doc: - args.price_list_currency = (parent.price_list_currency,) - args.plc_conversion_rate = parent.plc_conversion_rate - if args.get("items"): - for i, item in enumerate(args.get("items")): + ctx.price_list_currency = (parent.price_list_currency,) + ctx.plc_conversion_rate = parent.plc_conversion_rate + if ctx.items: + for i, item in enumerate(ctx.items): for fieldname in children[i]: # if the field exists in the original doc # update the value if fieldname in item and fieldname not in ("name", "doctype"): item[fieldname] = children[i][fieldname] - return args + return ctx else: return {"parent": parent, "children": children} -def apply_price_list_on_item(args, doc=None): - item_doc = frappe.db.get_value("Item", args.item_code, ["name", "variant_of"], as_dict=1) - item_details = get_price_list_rate(args, item_doc) - item_details.update(get_pricing_rule_for_item(args, doc=doc)) +def apply_price_list_on_item(ctx, doc=None): + item_doc = frappe.db.get_value("Item", ctx.item_code, ["name", "variant_of"], as_dict=1) + item_details = get_price_list_rate(ctx, item_doc) + item_details.update(get_pricing_rule_for_item(ctx, doc=doc)) return item_details -def get_price_list_currency_and_exchange_rate(args): - if not args.price_list: +def get_price_list_currency_and_exchange_rate(ctx: ItemDetailsCtx): + if not ctx.price_list: return {} - if args.doctype in ["Quotation", "Sales Order", "Delivery Note", "Sales Invoice"]: - args.update({"exchange_rate": "for_selling"}) - elif args.doctype in ["Purchase Order", "Purchase Receipt", "Purchase Invoice"]: - args.update({"exchange_rate": "for_buying"}) + if ctx.doctype in ["Quotation", "Sales Order", "Delivery Note", "Sales Invoice"]: + ctx.update({"exchange_rate": "for_selling"}) + elif ctx.doctype in ["Purchase Order", "Purchase Receipt", "Purchase Invoice"]: + ctx.update({"exchange_rate": "for_buying"}) - price_list_details = get_price_list_details(args.price_list) + price_list_details = get_price_list_details(ctx.price_list) price_list_currency = price_list_details.get("currency") price_list_uom_dependant = price_list_details.get("price_list_uom_dependant") - plc_conversion_rate = args.plc_conversion_rate - company_currency = get_company_currency(args.company) + plc_conversion_rate = ctx.plc_conversion_rate + company_currency = get_company_currency(ctx.company) if (not plc_conversion_rate) or ( - price_list_currency and args.price_list_currency and price_list_currency != args.price_list_currency + price_list_currency and ctx.price_list_currency and price_list_currency != ctx.price_list_currency ): # cksgb 19/09/2016: added args.transaction_date as posting_date argument for get_exchange_rate plc_conversion_rate = ( - get_exchange_rate( - price_list_currency, company_currency, args.transaction_date, args.exchange_rate - ) + get_exchange_rate(price_list_currency, company_currency, ctx.transaction_date, ctx.exchange_rate) or plc_conversion_rate ) @@ -1388,7 +1431,7 @@ def get_valuation_rate(item_code, company, warehouse=None): return {"valuation_rate": 0.0} -def get_gross_profit(out): +def get_gross_profit(out: ItemDetails): if out.valuation_rate: out.update({"gross_profit": ((out.base_rate - out.valuation_rate) * out.stock_qty)}) @@ -1396,26 +1439,24 @@ def get_gross_profit(out): @frappe.whitelist() -def get_serial_no(args, serial_nos=None, sales_order=None): +def get_serial_no(_args, serial_nos=None, sales_order=None): serial_nos = serial_nos or [] return serial_nos -def update_party_blanket_order(args, out): +def update_party_blanket_order(ctx: ItemDetailsCtx, out: ItemDetails | dict): if out["against_blanket_order"]: - blanket_order_details = get_blanket_order_details(args) + blanket_order_details = get_blanket_order_details(ctx) if blanket_order_details: out.update(blanket_order_details) @frappe.whitelist() -def get_blanket_order_details(args): - if isinstance(args, str): - args = frappe._dict(json.loads(args)) - +@type_narrow_ctx_arg +def get_blanket_order_details(ctx: ItemDetailsCtx): blanket_order_details = None - if args.item_code: + if ctx.item_code: bo = frappe.qb.DocType("Blanket Order") bo_item = frappe.qb.DocType("Blanket Order Item") @@ -1424,21 +1465,21 @@ def get_blanket_order_details(args): .from_(bo_item) .select(bo_item.rate.as_("blanket_order_rate"), bo.name.as_("blanket_order")) .where( - (bo.company == args.company) - & (bo_item.item_code == args.item_code) + (bo.company == ctx.company) + & (bo_item.item_code == ctx.item_code) & (bo.docstatus == 1) & (bo.name == bo_item.parent) ) ) - if args.customer and args.doctype == "Sales Order": - query = query.where(bo.customer == args.customer) - elif args.supplier and args.doctype == "Purchase Order": - query = query.where(bo.supplier == args.supplier) - if args.blanket_order: - query = query.where(bo.name == args.blanket_order) - if args.transaction_date: - query = query.where(bo.to_date >= args.transaction_date) + if ctx.customer and ctx.doctype == "Sales Order": + query = query.where(bo.customer == ctx.customer) + elif ctx.supplier and ctx.doctype == "Purchase Order": + query = query.where(bo.supplier == ctx.supplier) + if ctx.blanket_order: + query = query.where(bo.name == ctx.blanket_order) + if ctx.transaction_date: + query = query.where(bo.to_date >= ctx.transaction_date) blanket_order_details = query.run(as_dict=True) blanket_order_details = blanket_order_details[0] if blanket_order_details else ""