refactor: use frappe._dict in importers of ItemDetailsCtx

Extend the boundary rule to callers: non-decorated code that built or
annotated with ItemDetailsCtx now uses frappe._dict directly, and drops
the now-unused import. asset_capitalization keeps ItemDetailsCtx for its
own normalize_ctx_input-decorated functions.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
(cherry picked from commit e6f8f8f7e9)

# Conflicts:
#	erpnext/accounts/doctype/sales_invoice/services/pos.py
#	erpnext/accounts/services/taxes.py
#	erpnext/buying/doctype/purchase_order/test_purchase_order.py
#	erpnext/controllers/accounts_controller.py
#	erpnext/manufacturing/doctype/bom/bom.py
#	erpnext/selling/doctype/sales_order/mapper.py
#	erpnext/stock/doctype/packed_item/packed_item.py
#	erpnext/stock/doctype/stock_entry/stock_entry.py
This commit is contained in:
Mihir Kandoi
2026-07-01 16:30:16 +05:30
committed by Mergify
parent 2c2323a932
commit ed68505825
17 changed files with 2182 additions and 33 deletions

View File

@@ -659,7 +659,6 @@ 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 (
ItemDetailsCtx,
get_pos_profile,
get_pos_profile_item_details_,
)
@@ -732,7 +731,7 @@ class POSInvoice(SalesInvoice):
for item in self.get("items"):
if item.get("item_code"):
profile_details = get_pos_profile_item_details_(
ItemDetailsCtx(item.as_dict()), profile.get("company"), profile
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)):

View File

@@ -0,0 +1,417 @@
# Copyright (c) 2015, Frappe Technologies Pvt. Ltd. and Contributors
# License: GNU General Public License v3. See license.txt
"""POS helpers for Sales Invoice."""
import frappe
from frappe import _
from frappe.utils import cint, flt, get_link_to_form
class PartialPaymentValidationError(frappe.ValidationError):
pass
class POSService:
def __init__(self, doc) -> None:
self.doc = doc
def set_pos_fields(self, for_validate: bool = False) -> frappe.Document | dict | None:
"""Populate POS-profile fields on the invoice; return the profile, {} or None."""
doc = self.doc
if cint(doc.is_pos) != 1:
return None
self._set_default_change_amount_account()
if not self._ensure_pos_profile():
return None
pos = frappe.get_doc("POS Profile", doc.pos_profile) if doc.pos_profile else {}
if pos:
self._apply_pos_profile(pos, for_validate)
return pos
def _set_default_change_amount_account(self) -> None:
doc = self.doc
if not doc.account_for_change_amount:
doc.account_for_change_amount = frappe.get_cached_value(
"Company", doc.company, "default_cash_account"
)
def _ensure_pos_profile(self) -> bool:
"""Auto-pick a POS Profile for the company; return False if none could be found."""
doc = self.doc
if doc.pos_profile or doc.flags.ignore_pos_profile:
return True
from erpnext.stock.get_item_details import get_pos_profile
pos_profile = get_pos_profile(doc.company) or {}
if not pos_profile:
return False
doc.pos_profile = pos_profile.get("name")
return True
def _apply_pos_profile(self, pos, for_validate: bool) -> None:
doc = self.doc
if not for_validate:
self._apply_editable_pos_defaults(pos)
if pos.get("account_for_change_amount"):
doc.account_for_change_amount = pos.get("account_for_change_amount")
self._copy_pos_profile_fields(pos, for_validate)
if pos.get("company_address"):
doc.company_address = pos.get("company_address")
self._set_selling_price_list(pos)
if not for_validate:
self._set_update_stock_from_profile(pos)
self._apply_pos_item_defaults(pos, for_validate)
self._set_terms_and_taxes(pos)
def _apply_editable_pos_defaults(self, pos) -> None:
"""Profile defaults the user may override; only applied outside validation."""
doc = self.doc
update_multi_mode_option(doc, pos)
doc.tax_category = pos.get("tax_category")
if not doc.customer:
doc.customer = pos.customer
doc.ignore_pricing_rule = pos.ignore_pricing_rule
def _copy_pos_profile_fields(self, pos, for_validate: bool) -> None:
doc = self.doc
for fieldname in (
"currency",
"letter_head",
"tc_name",
"company",
"select_print_heading",
"write_off_account",
"taxes_and_charges",
"write_off_cost_center",
"apply_discount_on",
"cost_center",
):
if (not for_validate) or (for_validate and not doc.get(fieldname)):
doc.set(fieldname, pos.get(fieldname))
def _set_selling_price_list(self, pos) -> None:
doc = self.doc
if doc.customer:
customer_price_list, customer_group = frappe.get_value(
"Customer", doc.customer, ["default_price_list", "customer_group"]
)
customer_group_price_list = frappe.get_value(
"Customer Group", customer_group, "default_price_list"
)
selling_price_list = (
customer_price_list or customer_group_price_list or pos.get("selling_price_list")
)
else:
selling_price_list = pos.get("selling_price_list")
if selling_price_list:
doc.set("selling_price_list", selling_price_list)
def _set_update_stock_from_profile(self, pos) -> None:
doc = self.doc
dn_flag = any(d.get("dn_detail") for d in doc.get("items"))
doc.update_stock = 0 if dn_flag else cint(pos.get("update_stock"))
def _apply_pos_item_defaults(self, pos, for_validate: bool) -> None:
from erpnext.stock.get_item_details import get_pos_profile_item_details_
for item in self.doc.get("items"):
if not item.get("item_code"):
continue
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)):
item.set(fname, val)
def _set_terms_and_taxes(self, pos) -> None:
doc = self.doc
if doc.tc_name and not doc.terms:
doc.terms = frappe.db.get_value("Terms and Conditions", doc.tc_name, "terms")
if doc.taxes_and_charges and not len(doc.get("taxes")):
from erpnext.accounts.services.taxes import TaxService
TaxService(doc).set_taxes()
def update_paid_amount(self) -> None:
doc = self.doc
paid_amount = 0.0
base_paid_amount = 0.0
if not cint(doc.is_pos) and doc.is_return:
doc.set("payments", [])
doc.paid_amount = paid_amount
doc.base_paid_amount = base_paid_amount
return
for data in doc.payments:
data.base_amount = flt(data.amount * doc.conversion_rate, doc.precision("base_paid_amount"))
paid_amount += data.amount
base_paid_amount += data.base_amount
doc.paid_amount = paid_amount
doc.base_paid_amount = base_paid_amount
def set_account_for_mode_of_payment(self) -> None:
for payment in self.doc.payments:
payment.account = get_bank_cash_account(payment.mode_of_payment, self.doc.company).get("account")
def reset_mode_of_payments(self) -> None:
doc = self.doc
if doc.pos_profile:
pos_profile = frappe.get_cached_doc("POS Profile", doc.pos_profile)
update_multi_mode_option(doc, pos_profile)
doc.paid_amount = 0
def validate_pos_return(self) -> None:
"""Ensure POS return payments are not less than the (negative) invoice total."""
doc = self.doc
if doc.is_consolidated:
return
if doc.is_pos and doc.is_return:
total_amount_in_payments = sum(payment.amount for payment in doc.payments)
invoice_total = doc.rounded_total or doc.grand_total
if total_amount_in_payments < invoice_total:
frappe.throw(_("Total payments amount can't be greater than {0}").format(-invoice_total))
def validate_pos_paid_amount(self) -> None:
doc = self.doc
if len(doc.payments) == 0 and doc.is_pos and flt(doc.grand_total) > 0:
frappe.throw(_("At least one mode of payment is required for POS invoice."))
def validate_pos(self) -> None:
"""On a POS return, paid amount plus write-off cannot exceed the grand total."""
doc = self.doc
if doc.is_return:
invoice_total = doc.rounded_total or doc.grand_total
if abs(flt(doc.paid_amount)) + abs(flt(doc.write_off_amount)) - abs(flt(invoice_total)) > 1.0 / (
10.0 ** (doc.precision("grand_total") + 1.0)
):
frappe.throw(_("Paid amount + Write Off Amount can not be greater than Grand Total"))
def validate_created_using_pos(self) -> None:
doc = self.doc
if doc.is_created_using_pos and not doc.pos_profile:
frappe.throw(_("POS Profile is mandatory to mark this invoice as POS Transaction."))
doc.invoice_type_in_pos = frappe.db.get_single_value("POS Settings", "invoice_type")
if doc.invoice_type_in_pos == "POS Invoice" and not doc.is_return:
frappe.throw(_("Transactions using Sales Invoice in POS are disabled."))
self.validate_pos_opening_entry()
def validate_full_payment(self) -> None:
"""Block partial payment on a submitted POS invoice unless the profile allows it."""
doc = self.doc
allow_partial_payment = frappe.db.get_value("POS Profile", doc.pos_profile, "allow_partial_payment")
invoice_total = flt(doc.rounded_total) or flt(doc.grand_total)
if (
doc.docstatus == 1
and not doc.is_return
and not allow_partial_payment
and doc.paid_amount < invoice_total
):
frappe.throw(
msg=_("Partial Payment in POS Transactions are not allowed."),
exc=PartialPaymentValidationError,
)
def validate_pos_opening_entry(self) -> None:
"""Require exactly one current, open POS Opening Entry for the profile."""
doc = self.doc
opening_entries = frappe.get_all(
"POS Opening Entry",
fields=["name", "period_start_date"],
filters={"pos_profile": doc.pos_profile, "status": "Open"},
order_by="period_start_date desc",
)
if not opening_entries:
frappe.throw(
title=_("POS Opening Entry Missing"),
msg=_("No open POS Opening Entry found for POS Profile {0}.").format(
frappe.bold(doc.pos_profile)
),
)
if len(opening_entries) > 1:
frappe.throw(
title=_("Multiple POS Opening Entry"),
msg=_(
"POS Profile - {0} has multiple open POS Opening Entries. Please close or cancel the existing entries before proceeding."
).format(doc.pos_profile),
)
if frappe.utils.get_date_str(opening_entries[0].get("period_start_date")) != frappe.utils.today():
frappe.throw(
title=_("Outdated POS Opening Entry"),
msg=_(
"POS Opening Entry - {0} is outdated. Please close the POS and create a new POS Opening Entry."
).format(opening_entries[0].get("name")),
)
def check_if_consolidated_invoice(self) -> None:
doc = self.doc
if doc.doctype == "Sales Invoice" and doc.is_consolidated:
invoice_or_credit_note = "consolidated_credit_note" if doc.is_return else "consolidated_invoice"
pos_closing_entry = frappe.get_all(
"POS Invoice Merge Log",
filters={invoice_or_credit_note: doc.name},
pluck="pos_closing_entry",
)
if pos_closing_entry and pos_closing_entry[0]:
msg = _("To cancel a {0} you need to cancel the POS Closing Entry {1}.").format(
frappe.bold(_("Consolidated Sales Invoice")),
get_link_to_form("POS Closing Entry", pos_closing_entry[0]),
)
frappe.throw(msg, title=_("Not Allowed"))
def check_if_created_using_pos_and_pos_closing_entry_generated(self) -> None:
doc = self.doc
if doc.doctype == "Sales Invoice" and doc.is_created_using_pos and doc.pos_closing_entry:
pos_closing_entry_docstatus = frappe.db.get_value(
"POS Closing Entry", doc.pos_closing_entry, "docstatus"
)
if pos_closing_entry_docstatus == 1:
frappe.throw(
msg=_(
"To cancel this Sales Invoice you need to cancel the POS Closing Entry {0}."
).format(get_link_to_form("POS Closing Entry", doc.pos_closing_entry)),
title=_("Not Allowed"),
)
def cancel_pos_invoice_credit_note_generated_during_sales_invoice_mode(self) -> None:
pos_invoices = frappe.get_all(
"POS Invoice", filters={"consolidated_invoice": self.doc.name}, pluck="name"
)
for pos_invoice in pos_invoices:
frappe.get_doc("POS Invoice", pos_invoice).cancel()
def clear_unallocated_mode_of_payments(self) -> None:
doc = self.doc
doc.set("payments", doc.get("payments", {"amount": ["not in", [0, None, ""]]}))
frappe.db.delete("Sales Invoice Payment", filters={"parent": doc.name, "amount": 0})
def allow_write_off_only_on_pos(self) -> None:
if not self.doc.is_pos and self.doc.write_off_account:
self.doc.write_off_account = None
def verify_payment_amount_is_positive(self) -> None:
for entry in self.doc.payments:
if entry.amount < 0:
frappe.throw(_("Row #{0} (Payment Table): Amount must be positive").format(entry.idx))
def verify_payment_amount_is_negative(self) -> None:
for entry in self.doc.payments:
if entry.amount > 0:
frappe.throw(_("Row #{0} (Payment Table): Amount must be negative").format(entry.idx))
def get_bank_cash_account(mode_of_payment: str, company: str) -> dict:
account = frappe.db.get_value(
"Mode of Payment Account",
{"parent": mode_of_payment, "company": company},
"default_account",
)
if not account:
frappe.throw(
_("Please set default Cash or Bank account in Mode of Payment {0}").format(
get_link_to_form("Mode of Payment", mode_of_payment)
),
title=_("Missing Account"),
)
return {"account": account}
def update_multi_mode_option(doc, pos_profile) -> None:
def append_payment(payment_mode):
payment = doc.append("payments", {})
payment.default = payment_mode.default
payment.mode_of_payment = payment_mode.mop
payment.account = payment_mode.default_account
payment.type = payment_mode.type
mop_refetched = bool(doc.payments) and not doc.is_created_using_pos
doc.set("payments", [])
invalid_modes = []
mode_of_payments = [d.mode_of_payment for d in pos_profile.get("payments")]
mode_of_payments_info = get_mode_of_payments_info(mode_of_payments, doc.company)
for row in pos_profile.get("payments"):
payment_mode = mode_of_payments_info.get(row.mode_of_payment)
if not payment_mode:
invalid_modes.append(get_link_to_form("Mode of Payment", row.mode_of_payment))
continue
payment_mode.default = row.default
append_payment(payment_mode)
if invalid_modes:
if invalid_modes == 1:
msg = _("Please set default Cash or Bank account in Mode of Payment {0}")
else:
msg = _("Please set default Cash or Bank account in Mode of Payments {0}")
frappe.throw(msg.format(", ".join(invalid_modes)), title=_("Missing Account"))
if mop_refetched:
frappe.toast(
_("Payment methods refreshed. Please review before proceeding."),
indicator="orange",
)
def get_all_mode_of_payments(doc) -> list:
"""All enabled modes of payment with their default accounts for the doc's company."""
query, mopa, mop = _enabled_mode_of_payment_query(doc.company)
return query.select(mopa.default_account, mopa.parent, mop.type.as_("type")).run(as_dict=1)
def get_mode_of_payments_info(mode_of_payments: list, company: str) -> dict:
"""Map each of the named modes of payment to its account info for the company."""
query, mopa, mop = _enabled_mode_of_payment_query(company)
data = (
query.select(mopa.default_account, mopa.parent.as_("mop"), mop.type.as_("type"))
.where(mop.name.isin(mode_of_payments))
# group by all selected columns so postgres accepts it (one row per mode of payment)
.groupby(mopa.default_account, mopa.parent, mop.type)
.run(as_dict=1)
)
return {row.get("mop"): row for row in data}
def get_mode_of_payment_info(mode_of_payment: str, company: str) -> list:
"""Account info for a single mode of payment in the company."""
query, mopa, mop = _enabled_mode_of_payment_query(company)
return (
query.select(mopa.default_account, mopa.parent, mop.type.as_("type"))
.where(mop.name == mode_of_payment)
.run(as_dict=1)
)
def _enabled_mode_of_payment_query(company: str):
"""Base query joining enabled modes of payment to their accounts for a company."""
mopa = frappe.qb.DocType("Mode of Payment Account")
mop = frappe.qb.DocType("Mode of Payment")
query = (
frappe.qb.from_(mopa)
.join(mop)
.on(mopa.parent == mop.name)
.where(mopa.company == company)
.where(mop.enabled == 1)
)
return query, mopa, mop

View File

@@ -0,0 +1,446 @@
# Copyright (c) 2015, Frappe Technologies Pvt. Ltd. and Contributors
# License: GNU General Public License v3. See license.txt
"""Tax helpers: TaxService class for doc-mutating operations, free functions for stateless utilities."""
import json
import frappe
from frappe import _, throw
from frappe.utils import cint, flt, parse_json
import erpnext
from erpnext.stock.get_item_details import (
NOT_APPLICABLE_TAX,
_get_item_tax_template,
_get_item_tax_template_from_item_group,
get_item_tax_map,
)
class TaxService:
def __init__(self, doc):
self.doc = doc
def set_taxes(self) -> None:
doc = self.doc
if not doc.meta.get_field("taxes"):
return
tax_master_doctype = doc.meta.get_field("taxes_and_charges").options
if (doc.is_new() or self.is_pos_profile_changed()) and not doc.get("taxes"):
if doc.company and not doc.get("taxes_and_charges"):
doc.taxes_and_charges = frappe.db.get_value(
tax_master_doctype, {"is_default": 1, "company": doc.company}
)
self.append_taxes_from_master(tax_master_doctype)
def is_pos_profile_changed(self) -> bool:
doc = self.doc
if (
doc.doctype == "Sales Invoice"
and doc.is_pos
and doc.pos_profile != frappe.db.get_value("Sales Invoice", doc.name, "pos_profile")
):
return True
def set_taxes_and_charges(self) -> None:
doc = self.doc
if doc.doctype == "Material Request":
return
if doc.get("taxes") or doc.get("is_pos"):
return
if frappe.get_single_value(
"Accounts Settings", "add_taxes_from_taxes_and_charges_template"
) and hasattr(doc, "taxes_and_charges"):
if tax_master_doctype := doc.meta.get_field("taxes_and_charges").options:
self.append_taxes_from_master(tax_master_doctype)
if frappe.get_single_value("Accounts Settings", "add_taxes_from_item_tax_template"):
self.append_taxes_from_item_tax_template()
def append_taxes_from_master(self, tax_master_doctype=None) -> None:
doc = self.doc
if doc.get("taxes_and_charges"):
if not tax_master_doctype:
tax_master_doctype = doc.meta.get_field("taxes_and_charges").options
doc.extend("taxes", get_taxes_and_charges(tax_master_doctype, doc.get("taxes_and_charges")))
def append_taxes_from_item_tax_template(self) -> None:
doc = self.doc
if not frappe.get_single_value("Accounts Settings", "add_taxes_from_item_tax_template"):
return
for row in doc.items:
item_tax_rate = row.get("item_tax_rate")
if not item_tax_rate:
continue
if isinstance(item_tax_rate, str):
item_tax_rate = parse_json(item_tax_rate)
for account_head, _rate in item_tax_rate.items():
if not self.get_tax_row(account_head):
doc.append(
"taxes",
{
"charge_type": "On Net Total",
"account_head": account_head,
"rate": 0,
"description": account_head,
"set_by_item_tax_template": 1,
"category": "Total",
"add_deduct_tax": "Add",
},
)
def get_tax_row(self, account_head):
for row in self.doc.taxes:
if row.account_head == account_head:
return row
def set_other_charges(self) -> None:
self.doc.set("taxes", [])
self.set_taxes()
def validate_enabled_taxes_and_charges(self) -> None:
doc = self.doc
taxes_and_charges_doctype = doc.meta.get_options("taxes_and_charges")
if doc.taxes_and_charges and frappe.get_cached_value(
taxes_and_charges_doctype, doc.taxes_and_charges, "disabled"
):
frappe.throw(_("{0} '{1}' is disabled").format(taxes_and_charges_doctype, doc.taxes_and_charges))
def validate_tax_account_company(self) -> None:
doc = self.doc
for d in doc.get("taxes"):
if d.account_head:
tax_account_company = frappe.get_cached_value("Account", d.account_head, "company")
if tax_account_company != doc.company:
frappe.throw(
_("Row #{0}: Account {1} does not belong to company {2}").format(
d.idx, d.account_head, doc.company
)
)
def get_tax_map(self) -> dict:
tax_map = {}
for tax in self.doc.get("taxes"):
tax_map.setdefault(tax.account_head, 0.0)
tax_map[tax.account_head] += tax.tax_amount
return tax_map
def get_amount_and_base_amount(self, item, enable_discount_accounting):
doc = self.doc
amount = item.net_amount
base_amount = item.base_net_amount
if (
enable_discount_accounting
and doc.get("discount_amount")
and doc.get("additional_discount_account")
):
if not hasattr(doc, "__has_distributed_discount_set"):
doc.__has_distributed_discount_set = any(
i.distributed_discount_amount for i in doc.get("items")
)
if not doc.__has_distributed_discount_set:
return item.amount, item.base_amount
amount += item.distributed_discount_amount
base_amount += flt(
item.distributed_discount_amount * doc.get("conversion_rate"),
item.precision("distributed_discount_amount"),
)
return amount, base_amount
def get_tax_amounts(self, tax, enable_discount_accounting):
doc = self.doc
amount = tax.tax_amount_after_discount_amount
base_amount = tax.base_tax_amount_after_discount_amount
if (
enable_discount_accounting
and doc.get("discount_amount")
and doc.get("additional_discount_account")
and doc.get("apply_discount_on") == "Grand Total"
):
amount = tax.tax_amount
base_amount = tax.base_tax_amount
return amount, base_amount
@frappe.whitelist()
def get_tax_rate(account_head: str) -> dict:
return frappe.get_cached_value("Account", account_head, ["tax_rate", "account_name"], as_dict=True)
@frappe.whitelist()
def get_default_taxes_and_charges(
master_doctype: str, tax_template: str | None = None, company: str | None = None
) -> dict | None:
if not company:
return {}
if tax_template and company:
tax_template_company = frappe.get_cached_value(master_doctype, tax_template, "company")
if tax_template_company == company:
return
default_tax = frappe.db.get_value(master_doctype, {"is_default": 1, "company": company})
return {
"taxes_and_charges": default_tax,
"taxes": get_taxes_and_charges(master_doctype, default_tax),
}
@frappe.whitelist()
def get_taxes_and_charges(master_doctype: str, master_name: str | None = None) -> list | None:
if not master_name:
return
from frappe.model import child_table_fields, default_fields
tax_master = frappe.get_doc(master_doctype, master_name)
taxes_and_charges = []
for _i, tax in enumerate(tax_master.get("taxes")):
tax = tax.as_dict()
for fieldname in default_fields + child_table_fields:
if fieldname in tax:
del tax[fieldname]
taxes_and_charges.append(tax)
return taxes_and_charges
def validate_conversion_rate(
currency: str, conversion_rate: float, conversion_rate_label: str, company: str
) -> None:
"""Throw a validation error if conversion_rate is falsy."""
company_currency = frappe.get_cached_value("Company", company, "default_currency")
if not conversion_rate:
throw(
_("{0} is mandatory. Maybe Currency Exchange record is not created for {1} to {2}.").format(
conversion_rate_label, currency, company_currency
)
)
def validate_taxes_and_charges(tax) -> None:
if tax.charge_type in ["Actual", "On Net Total", "On Paid Amount"] and tax.row_id:
frappe.throw(
_("Can refer row only if the charge type is 'On Previous Row Amount' or 'Previous Row Total'")
)
elif tax.charge_type in ["On Previous Row Amount", "On Previous Row Total"]:
if cint(tax.idx) == 1:
frappe.throw(
_(
"Cannot select charge type as 'On Previous Row Amount' or 'On Previous Row Total' for first row"
)
)
elif not tax.row_id:
frappe.throw(
_("Please specify a valid Row ID for row {0} in table {1}").format(tax.idx, _(tax.doctype))
)
elif tax.row_id and cint(tax.row_id) >= cint(tax.idx):
frappe.throw(
_("Cannot refer row number greater than or equal to current row number for this Charge type")
)
if tax.charge_type == "Actual":
tax.rate = None
def validate_account_head(idx: int, account: str, company: str, context: str | None = None) -> None:
"""Throw a ValidationError if the account belongs to a different company or is a group account."""
if company != frappe.get_cached_value("Account", account, "company"):
frappe.throw(
_("Row {0}: The {3} Account {1} does not belong to the company {2}").format(
idx, frappe.bold(account), frappe.bold(company), context or ""
),
title=_("Invalid Account"),
)
if frappe.get_cached_value("Account", account, "is_group"):
frappe.throw(
_(
"You selected the account group {1} as {2} Account in row {0}. Please select a single account."
).format(idx, frappe.bold(account), context or ""),
title=_("Invalid Account"),
)
def validate_cost_center(tax, doc) -> None:
if not tax.cost_center:
return
company = frappe.get_cached_value("Cost Center", tax.cost_center, "company")
if company != doc.company:
frappe.throw(
_("Row {0}: Cost Center {1} does not belong to Company {2}").format(
tax.idx, frappe.bold(tax.cost_center), frappe.bold(doc.company)
),
title=_("Invalid Cost Center"),
)
def validate_inclusive_tax(tax, doc) -> None:
def _on_previous_row_error(row_range):
throw(
_("To include tax in row {0} in Item rate, taxes in rows {1} must also be included").format(
tax.idx, row_range
)
)
if cint(getattr(tax, "included_in_print_rate", None)):
if tax.charge_type == "Actual":
throw(
_("Charge of type 'Actual' in row {0} cannot be included in Item Rate or Paid Amount").format(
tax.idx
)
)
elif tax.charge_type == "On Previous Row Amount" and not cint(
doc.get("taxes")[cint(tax.row_id) - 1].included_in_print_rate
):
_on_previous_row_error(tax.row_id)
elif tax.charge_type == "On Previous Row Total" and not all(
[cint(t.included_in_print_rate) for t in doc.get("taxes")[: cint(tax.row_id) - 1]]
):
_on_previous_row_error("1 - %d" % (tax.row_id,))
elif tax.get("category") == "Valuation":
frappe.throw(_("Valuation type charges can not be marked as Inclusive"))
def set_balance_in_account_currency(
gl_dict,
account_currency: str | None = None,
conversion_rate: float | None = None,
company_currency: str | None = None,
) -> None:
if (not conversion_rate) and (account_currency != company_currency):
frappe.throw(
_("Account: {0} with currency: {1} can not be selected").format(gl_dict.account, account_currency)
)
gl_dict["account_currency"] = account_currency
if flt(gl_dict.debit) and not flt(gl_dict.debit_in_account_currency):
gl_dict.debit_in_account_currency = (
gl_dict.debit if account_currency == company_currency else flt(gl_dict.debit / conversion_rate, 2)
)
if flt(gl_dict.credit) and not flt(gl_dict.credit_in_account_currency):
gl_dict.credit_in_account_currency = (
gl_dict.credit
if account_currency == company_currency
else flt(gl_dict.credit / conversion_rate, 2)
)
def set_child_tax_template_and_map(item, child_item, parent_doc) -> None:
ctx = frappe._dict(
{
"item_code": item.item_code,
"posting_date": parent_doc.transaction_date,
"tax_category": parent_doc.get("tax_category"),
"company": parent_doc.get("company"),
"base_net_rate": item.get("base_net_rate"),
}
)
item_tax_template = _get_item_tax_template(ctx, item.taxes)
if not item_tax_template:
item_tax_template = _get_item_tax_template_from_item_group(ctx, item.item_group)
child_item.item_tax_template = item_tax_template
child_item.item_tax_rate = get_item_tax_map(
doc=parent_doc,
tax_template=child_item.item_tax_template,
as_json=True,
)
def add_taxes_from_tax_template(child_item, parent_doc, db_insert: bool = True) -> None:
add_taxes_from_item_tax_template = frappe.get_single_value(
"Accounts Settings", "add_taxes_from_item_tax_template"
)
if child_item.get("item_tax_rate") and add_taxes_from_item_tax_template:
tax_map = json.loads(child_item.get("item_tax_rate"))
for tax_type, tax_rate in tax_map.items():
if tax_rate == NOT_APPLICABLE_TAX:
continue
tax_rate = flt(tax_rate)
taxes = parent_doc.get("taxes") or []
found = any(tax.account_head == tax_type for tax in taxes)
if not found:
tax_row = parent_doc.append("taxes", {})
tax_row.update(
{
"description": str(tax_type).split(" - ")[0],
"charge_type": "On Net Total",
"account_head": tax_type,
"rate": tax_rate,
"set_by_item_tax_template": 1,
}
)
if parent_doc.doctype == "Purchase Order":
tax_row.update({"category": "Total", "add_deduct_tax": "Add"})
if db_insert:
tax_row.db_insert()
def merge_taxes(source_doc, target_doc) -> None:
tax_map = {}
for tax in source_doc.get("taxes") or []:
found = False
for t in target_doc.get("taxes") or []:
if t.account_head == tax.account_head and t.cost_center == tax.cost_center:
t.tax_amount = flt(t.tax_amount) + flt(tax.tax_amount_after_discount_amount)
t.base_tax_amount = flt(t.base_tax_amount) + flt(tax.base_tax_amount_after_discount_amount)
tax_map[tax.name] = t
found = True
if not found:
tax.charge_type = "Actual"
tax.included_in_print_rate = 0
tax.dont_recompute_tax = 1
tax.row_id = None
tax.idx = None
tax.tax_amount = tax.tax_amount_after_discount_amount
tax.base_tax_amount = tax.base_tax_amount_after_discount_amount
tax_map[tax.name] = target_doc.append("taxes", tax)
item_map = {d._old_name: d for d in target_doc.get("items") if d.get("_old_name")}
item_tax_details = target_doc.get("_item_wise_tax_details") or []
for row in source_doc.get("item_wise_tax_details"):
item = item_map.get(row.item_row)
tax = tax_map.get(row.tax_row)
if not (item and tax):
continue
item_tax_details.append(
frappe._dict(
item=item,
tax=tax,
amount=row.amount,
rate=row.rate,
taxable_amount=row.taxable_amount,
)
)
target_doc._item_wise_tax_details = item_tax_details

View File

@@ -657,7 +657,7 @@ def get_target_item_details(item_code: str | None = None, company: str | None =
item_group_defaults = get_item_group_defaults(item.name, company)
brand_defaults = get_brand_defaults(item.name, company)
out.cost_center = get_default_cost_center(
ItemDetailsCtx({"item_code": item.name, "company": company}),
frappe._dict({"item_code": item.name, "company": company}),
item_defaults,
item_group_defaults,
brand_defaults,

View File

@@ -1564,6 +1564,91 @@ class TestPurchaseOrder(ERPNextTestSuite):
pi2 = make_pi_from_po(po.name)
self.assertEqual(len(pi2.items), 2)
<<<<<<< HEAD
=======
def test_get_item_details_propagates_drop_ship_flag_to_po(self):
"""`get_item_details` should propagate the Item master's
`delivered_by_supplier` flag to Purchase Orders, not only to Sales
Orders/Invoices, so that POs can be created as drop-ship directly
(via the standard item lookup the form uses) without going through
the Sales Order → Purchase Order mapping pipeline.
"""
from erpnext.stock.get_item_details import get_item_details
item = make_item("_Test Drop Ship From Master", {"is_stock_item": 1, "delivered_by_supplier": 1})
ctx = frappe._dict(
{
"item_code": item.item_code,
"doctype": "Purchase Order",
"company": "_Test Company",
"supplier": "_Test Supplier",
"transaction_date": nowdate(),
"currency": "INR",
"conversion_rate": 1.0,
"buying_price_list": "Standard Buying",
"price_list_currency": "INR",
"plc_conversion_rate": 1.0,
"qty": 1,
}
)
details = get_item_details(ctx, frappe.new_doc("Purchase Order"))
self.assertEqual(details.get("delivered_by_supplier"), 1)
def test_drop_ship_po_allows_non_company_shipping_address_without_so(self):
"""A PO with a drop-ship item should save with a non-company shipping
address even when there is no linked Sales Order.
Regression test for https://github.com/frappe/erpnext/issues/51629.
"""
from erpnext.crm.doctype.prospect.test_prospect import make_address
item = make_item("_Test Drop Ship Direct PO", {"is_stock_item": 1, "delivered_by_supplier": 1})
customer_shipping = make_address(
address_title="Drop Ship Direct PO", address_type="Shipping", address_line1="1"
)
customer_shipping.append("links", {"link_doctype": "Customer", "link_name": "_Test Customer"})
customer_shipping.save()
po = create_purchase_order(item=item.item_code, qty=1, do_not_save=True)
# In the UI, `get_item_details` propagates the master flag to the row when
# the item is added; here we simulate that step explicitly.
po.items[0].delivered_by_supplier = 1
po.items[0].warehouse = ""
po.shipping_address = customer_shipping.name
po.save()
self.assertEqual(po.items[0].delivered_by_supplier, 1)
self.assertFalse(po.items[0].warehouse)
self.assertEqual(po.shipping_address, customer_shipping.name)
def test_drop_ship_flag_overridable_per_po_line(self):
"""The drop-ship default from the Item master should be overridable
on individual PO lines (e.g. ordering a normally drop-shipped item
into the own warehouse for samples or stock).
"""
item = make_item("_Test Drop Ship Override", {"is_stock_item": 1, "delivered_by_supplier": 1})
po = create_purchase_order(item=item.item_code, qty=1, do_not_save=True)
po.items[0].delivered_by_supplier = 0
po.save()
self.assertEqual(po.items[0].delivered_by_supplier, 0)
self.assertEqual(po.items[0].warehouse, "_Test Warehouse - _TC")
def test_remove_unlinked_item_from_mixed_po_does_not_crash(self):
"""In a PO that mixes SO-linked and freely-added items, removing an
item that has no `sales_order_item` via Update Items must not crash
on the missing reference.
"""
po = create_purchase_order(do_not_submit=True)
# Force the SO codepath without needing a real linked Sales Order:
po.items[0].sales_order = "DUMMY-SO"
po.update_ordered_qty_in_so_for_removed_items([frappe._dict({"sales_order_item": None, "qty": 1})])
>>>>>>> e6f8f8f7e9 (refactor: use frappe._dict in importers of ItemDetailsCtx)
def create_po_for_sc_testing():
from erpnext.controllers.tests.test_subcontracting_controller import (

View File

@@ -67,12 +67,15 @@ from erpnext.setup.utils import get_exchange_rate
from erpnext.stock.doctype.item.item import get_uom_conv_factor
from erpnext.stock.doctype.packed_item.packed_item import make_packing_list
from erpnext.stock.get_item_details import (
<<<<<<< HEAD
NOT_APPLICABLE_TAX,
ItemDetailsCtx,
_get_item_tax_template,
_get_item_tax_template_from_item_group,
get_bin_details,
get_conversion_factor,
=======
>>>>>>> e6f8f8f7e9 (refactor: use frappe._dict in importers of ItemDetailsCtx)
get_item_details,
get_item_tax_map,
get_item_warehouse_,
@@ -1087,7 +1090,7 @@ class AccountsController(TransactionBase):
for item in self.get("items"):
if item.get("item_code"):
ctx: ItemDetailsCtx = ItemDetailsCtx(parent_dict.copy())
ctx: frappe._dict = frappe._dict(parent_dict.copy())
ctx.update(item.as_dict())
ctx.update(

View File

@@ -16,7 +16,7 @@ from pypika import Order
import erpnext
from erpnext.accounts.utils import build_qb_match_conditions
from erpnext.stock.get_item_details import ItemDetailsCtx, _get_item_tax_template
from erpnext.stock.get_item_details import _get_item_tax_template
from erpnext.stock.utils import get_combine_datetime
@@ -949,7 +949,7 @@ def get_tax_template(doctype, txt, searchfield, start, page_len, filters):
valid_from = filters.get("valid_from")
valid_from = valid_from[1] if isinstance(valid_from, list) else valid_from
ctx = ItemDetailsCtx(
ctx = frappe._dict(
{
"item_code": filters.get("item_code"),
"posting_date": valid_from,

View File

@@ -22,7 +22,6 @@ from erpnext.controllers.accounts_controller import (
from erpnext.deprecation_dumpster import deprecated
from erpnext.stock.get_item_details import (
NOT_APPLICABLE_TAX,
ItemDetailsCtx,
_get_item_tax_template,
get_item_tax_map,
)
@@ -100,7 +99,7 @@ class calculate_taxes_and_totals:
for item in self.doc.items:
if item.item_code and item.get("item_tax_template"):
item_doc = frappe.get_cached_doc("Item", item.item_code)
ctx = ItemDetailsCtx(
ctx = frappe._dict(
{
"net_rate": item.net_rate or item.rate,
"base_net_rate": item.base_net_rate or item.base_rate,

View File

@@ -18,7 +18,7 @@ from frappe.website.website_generator import WebsiteGenerator
import erpnext
from erpnext.setup.utils import get_exchange_rate
from erpnext.stock.doctype.item.item import get_item_details
from erpnext.stock.get_item_details import ItemDetailsCtx, get_conversion_factor, get_price_list_rate
from erpnext.stock.get_item_details import get_conversion_factor, get_price_list_rate
form_grid_templates = {"items": "templates/form_grid/item_grid.html"}
@@ -1335,6 +1335,35 @@ def get_bom_item_rate(args, bom_doc):
return flt(rate)
<<<<<<< HEAD
=======
def _get_price_list_item_rate(args, bom_doc):
if not bom_doc.buying_price_list:
frappe.throw(_("Please select Price List"))
ctx = frappe._dict(
{
"doctype": "BOM",
"price_list": bom_doc.buying_price_list,
"qty": args.get("qty") or 1,
"uom": args.get("uom") or args.get("stock_uom"),
"stock_uom": args.get("stock_uom"),
"transaction_type": "buying",
"company": bom_doc.company,
"currency": bom_doc.currency,
"conversion_rate": 1, # Passed conversion rate as 1 purposefully, as conversion rate is applied at the end of the function
"conversion_factor": args.get("conversion_factor") or 1,
"plc_conversion_rate": 1,
"ignore_party": True,
"ignore_conversion_rate": True,
}
)
item_doc = frappe.get_cached_doc("Item", args.get("item_code"))
price_list_data = get_price_list_rate(ctx, item_doc)
return price_list_data.price_list_rate
>>>>>>> e6f8f8f7e9 (refactor: use frappe._dict in importers of ItemDetailsCtx)
def get_valuation_rate(data):
"""
1) Get average valuation rate from all warehouses

View File

@@ -290,7 +290,7 @@ class TestQuotation(ERPNextTestSuite):
def test_gross_profit(self):
from erpnext.stock.doctype.item.test_item import make_item
from erpnext.stock.doctype.stock_entry.stock_entry_utils import make_stock_entry
from erpnext.stock.get_item_details import ItemDetailsCtx, insert_item_price
from erpnext.stock.get_item_details import insert_item_price
item_doc = make_item("_Test Item for Gross Profit", {"is_stock_item": 1})
item_code = item_doc.name
@@ -299,7 +299,7 @@ class TestQuotation(ERPNextTestSuite):
selling_price_list = frappe.get_all("Price List", filters={"selling": 1}, limit=1)[0].name
frappe.db.set_single_value("Stock Settings", "auto_insert_price_list_rate_if_missing", 1)
insert_item_price(
ItemDetailsCtx(
frappe._dict(
{
"item_code": item_code,
"price_list": selling_price_list,

File diff suppressed because it is too large Load Diff

View File

@@ -19,7 +19,7 @@ from erpnext.stock.doctype.serial_and_batch_bundle.test_serial_and_batch_bundle
get_batch_from_bundle,
)
from erpnext.stock.doctype.stock_entry.stock_entry_utils import make_stock_entry
from erpnext.stock.get_item_details import ItemDetailsCtx, get_item_details
from erpnext.stock.get_item_details import get_item_details
from erpnext.stock.serial_batch_bundle import SerialBatchCreation
from erpnext.tests.utils import ERPNextTestSuite
@@ -549,7 +549,7 @@ class TestBatch(ERPNextTestSuite):
company = "_Test Company with perpetual inventory"
currency = frappe.get_cached_value("Company", company, "default_currency")
ctx = ItemDetailsCtx(
ctx = frappe._dict(
{
"item_code": "_Test Batch Price Item",
"company": company,

View File

@@ -25,7 +25,7 @@ from erpnext.stock.doctype.item.item import (
validate_is_stock_item,
)
from erpnext.stock.doctype.stock_entry.stock_entry_utils import make_stock_entry
from erpnext.stock.get_item_details import ItemDetailsCtx, get_item_details
from erpnext.stock.get_item_details import get_item_details
from erpnext.tests.utils import ERPNextTestSuite
@@ -146,7 +146,7 @@ class TestItem(ERPNextTestSuite):
currency = frappe.get_cached_value("Company", company, "default_currency")
details = get_item_details(
ItemDetailsCtx(
frappe._dict(
{
"item_code": "_Test Item",
"company": company,
@@ -176,7 +176,7 @@ class TestItem(ERPNextTestSuite):
create_fixed_asset_item()
details = get_item_details(
ItemDetailsCtx(
frappe._dict(
{
"item_code": "Macbook Pro",
"company": "_Test Company",
@@ -189,7 +189,7 @@ class TestItem(ERPNextTestSuite):
frappe.db.set_value("Asset Category", "Computers", "enable_cwip_accounting", "1")
details = get_item_details(
ItemDetailsCtx(
frappe._dict(
{
"item_code": "Macbook Pro",
"company": "_Test Company",
@@ -279,7 +279,7 @@ class TestItem(ERPNextTestSuite):
for data in expected_item_tax_template:
details = get_item_details(
ItemDetailsCtx(
frappe._dict(
{
"item_code": data["item_code"],
"tax_category": data["tax_category"],
@@ -331,7 +331,7 @@ class TestItem(ERPNextTestSuite):
"cost_center": "_Test Cost Center 2 - _TC", # from item group
}
sales_item_details = get_item_details(
ItemDetailsCtx(
frappe._dict(
{
"item_code": "Test Item With Defaults",
"company": "_Test Company",
@@ -356,7 +356,7 @@ class TestItem(ERPNextTestSuite):
"cost_center": "_Test Write Off Cost Center - _TC", # from item
}
purchase_item_details = get_item_details(
ItemDetailsCtx(
frappe._dict(
{
"item_code": "Test Item With Defaults",
"company": "_Test Company",

View File

@@ -6,7 +6,7 @@ import frappe
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 ItemDetailsCtx, get_price_list_rate_for
from erpnext.stock.get_item_details import get_price_list_rate_for
from erpnext.tests.utils import ERPNextTestSuite
@@ -69,7 +69,7 @@ class TestItemPrice(ERPNextTestSuite):
# Check correct price at this quantity
doc = frappe.copy_doc(self.globalTestRecords["Item Price"][2])
ctx = ItemDetailsCtx(
ctx = frappe._dict(
{
"price_list": doc.price_list,
"customer": doc.customer,
@@ -85,7 +85,7 @@ class TestItemPrice(ERPNextTestSuite):
def test_price_with_no_qty(self):
# Check correct price when no quantity
doc = frappe.copy_doc(self.globalTestRecords["Item Price"][2])
ctx = ItemDetailsCtx(
ctx = frappe._dict(
{
"price_list": doc.price_list,
"customer": doc.customer,
@@ -101,7 +101,7 @@ class TestItemPrice(ERPNextTestSuite):
# Check correct price at first date
doc = frappe.copy_doc(self.globalTestRecords["Item Price"][2])
ctx = ItemDetailsCtx(
ctx = frappe._dict(
{
"price_list": doc.price_list,
"customer": "_Test Customer",
@@ -118,7 +118,7 @@ class TestItemPrice(ERPNextTestSuite):
# Check correct price at invalid date
doc = frappe.copy_doc(self.globalTestRecords["Item Price"][3])
ctx = ItemDetailsCtx(
ctx = frappe._dict(
{
"price_list": doc.price_list,
"qty": 7,
@@ -134,7 +134,7 @@ class TestItemPrice(ERPNextTestSuite):
# Check correct price when outside of the date
doc = frappe.copy_doc(self.globalTestRecords["Item Price"][4])
ctx = ItemDetailsCtx(
ctx = frappe._dict(
{
"price_list": doc.price_list,
"customer": "_Test Customer",
@@ -151,7 +151,7 @@ class TestItemPrice(ERPNextTestSuite):
# Check lowest price when no date provided
doc = frappe.copy_doc(self.globalTestRecords["Item Price"][1])
ctx = ItemDetailsCtx(
ctx = frappe._dict(
{
"price_list": doc.price_list,
"uom": "_Test UOM",
@@ -183,7 +183,7 @@ class TestItemPrice(ERPNextTestSuite):
doc.price_list_rate = 21
doc.insert()
ctx = ItemDetailsCtx(
ctx = frappe._dict(
{
"price_list": doc.price_list,
"uom": "_Test UOM",

View File

@@ -11,7 +11,7 @@ import frappe.defaults
from frappe.model.document import Document
from frappe.utils import flt
from erpnext.stock.get_item_details import ItemDetailsCtx, get_item_details, get_price_list_rate
from erpnext.stock.get_item_details import get_item_details, get_price_list_rate
class PackedItem(Document):
@@ -286,7 +286,7 @@ def update_packed_item_price_data(pi_row, item_data, doc):
return
item_doc = frappe.get_cached_doc("Item", pi_row.item_code)
ctx = ItemDetailsCtx(pi_row.as_dict().copy())
ctx = frappe._dict(pi_row.as_dict().copy())
ctx.update(
{
"company": doc.get("company"),
@@ -378,7 +378,33 @@ def on_doctype_update():
def get_items_from_product_bundle(row):
row, items = ItemDetailsCtx(json.loads(row)), []
<<<<<<< HEAD
bundled_items = get_product_bundle_items(row["item_code"])
=======
``row.product_bundle`` selects a specific version by document name (the buying
dialog passes this); ``row.item_code`` is the legacy contract, resolving the
parent item's active version.
"""
from erpnext.selling.doctype.product_bundle.product_bundle import get_active_product_bundle
row, items = frappe._dict(frappe.parse_json(row)), []
if bundle_name := row.get("product_bundle"):
frappe.has_permission("Product Bundle", "read", bundle_name, throw=True)
bundle = frappe.db.get_value("Product Bundle", bundle_name, ["docstatus", "disabled"], as_dict=True)
if not bundle or bundle.docstatus != 1:
frappe.throw(_("Product Bundle {0} is not submitted").format(frappe.bold(bundle_name)))
if bundle.disabled:
frappe.throw(
_("Product Bundle {0} is disabled and cannot be used in transactions.").format(
frappe.bold(bundle_name)
)
)
elif bundle_name := get_active_product_bundle(row.get("item_code")):
frappe.has_permission("Product Bundle", "read", bundle_name, throw=True)
bundled_items = get_product_bundle_items_by_name(bundle_name) if bundle_name else []
>>>>>>> e6f8f8f7e9 (refactor: use frappe._dict in importers of ItemDetailsCtx)
for item in bundled_items:
row.update(
{

View File

@@ -2385,7 +2385,7 @@ class TestPurchaseReceipt(ERPNextTestSuite):
self.assertTrue(return_pi.docstatus == 1)
def test_disable_last_purchase_rate(self):
from erpnext.stock.get_item_details import ItemDetailsCtx, get_item_details
from erpnext.stock.get_item_details import get_item_details
item = make_item(
"_Test Disable Last Purchase Rate",
@@ -2400,7 +2400,7 @@ class TestPurchaseReceipt(ERPNextTestSuite):
item_code=item.name,
)
ctx = ItemDetailsCtx(pr.items[0].as_dict())
ctx = frappe._dict(pr.items[0].as_dict())
ctx.update(
{
"supplier": pr.supplier,

View File

@@ -42,7 +42,6 @@ from erpnext.stock.doctype.stock_reconciliation.stock_reconciliation import (
OpeningEntryAccountError,
)
from erpnext.stock.get_item_details import (
ItemDetailsCtx,
get_barcode_data,
get_bin_details,
get_conversion_factor,
@@ -2577,8 +2576,34 @@ class StockEntry(StockController, SubcontractingInwardController):
return reserved_work_orders
@frappe.whitelist()
<<<<<<< HEAD
def get_item_details(self, args: ItemDetailsCtx = None, for_update=False):
item = frappe.qb.DocType("Item")
=======
def get_item_details(self, args: frappe._dict | None = None, for_update: bool = False):
item = self._fetch_item_data(args)
item_group_defaults = get_item_group_defaults(item.name, self.company)
brand_defaults = get_brand_defaults(item.name, self.company)
ret = self._build_item_ret(args, item, item_group_defaults, brand_defaults, for_update)
self._apply_account_defaults(ret)
args["posting_date"] = self.posting_date
args["posting_time"] = self.posting_time
ret.update(get_warehouse_details(args) if args.get("warehouse") else {})
if self.purpose == "Send to Subcontractor":
self._resolve_subcontract_item(args, ret)
barcode_data = get_barcode_data(item_code=item.name)
if barcode_data and len(barcode_data.get(item.name)) == 1:
ret["barcode"] = barcode_data.get(item.name)[0]
return ret
def _fetch_item_data(self, args):
item_dt = frappe.qb.DocType("Item")
>>>>>>> e6f8f8f7e9 (refactor: use frappe._dict in importers of ItemDetailsCtx)
item_default = frappe.qb.DocType("Item Default")
query = (