refactor(accounts): extract tax helpers into accounts/services/taxes.py

Move validate_conversion_rate, validate_taxes_and_charges, validate_account_head,
validate_cost_center, validate_inclusive_tax, set_balance_in_account_currency,
set_child_tax_template_and_map, add_taxes_from_tax_template, merge_taxes,
get_tax_rate, get_default_taxes_and_charges, and get_taxes_and_charges out of
accounts_controller into a dedicated accounts/services/taxes.py module.

Re-export all symbols from accounts_controller for backward compatibility.
This commit is contained in:
Nabin Hait
2026-05-27 16:28:02 +05:30
parent 58c90ad651
commit 29261c5fc2
2 changed files with 301 additions and 275 deletions

View File

@@ -0,0 +1,287 @@
# Copyright (c) 2015, Frappe Technologies Pvt. Ltd. and Contributors
# License: GNU General Public License v3. See license.txt
"""Tax template and validation helpers shared across buying and selling controllers."""
import json
import frappe
from frappe import _, throw
from frappe.utils import cint, flt
from erpnext.stock.get_item_details import (
NOT_APPLICABLE_TAX,
ItemDetailsCtx,
_get_item_tax_template,
_get_item_tax_template_from_item_group,
get_item_tax_map,
)
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 = ItemDetailsCtx(
{
"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

@@ -68,14 +68,10 @@ 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 (
NOT_APPLICABLE_TAX,
ItemDetailsCtx,
_get_item_tax_template,
_get_item_tax_template_from_item_group,
get_bin_details,
get_conversion_factor,
get_item_details,
get_item_tax_map,
get_item_warehouse_,
)
from erpnext.utilities.regional import temporary_flag
@@ -2877,184 +2873,26 @@ class AccountsController(TransactionBase):
self.calculate_taxes_and_totals()
@frappe.whitelist()
def get_tax_rate(account_head: str):
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
):
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):
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, conversion_rate, conversion_rate_label, company):
"""common validation for currency and price list currency"""
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):
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):
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):
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":
# inclusive tax cannot be of 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
):
# referred row should also be inclusive
_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]]
):
# all rows about the referred tax should be inclusive
_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=None, conversion_rate=None, company_currency=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
# set debit/credit in account currency if not provided
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)
)
from erpnext.accounts.services.advances import (
get_advance_journal_entries,
get_advance_payment_entries,
get_advance_payment_entries_for_regional,
get_common_query,
)
from erpnext.accounts.services.taxes import (
add_taxes_from_tax_template,
get_default_taxes_and_charges,
get_tax_rate,
get_taxes_and_charges,
merge_taxes,
set_balance_in_account_currency,
set_child_tax_template_and_map,
validate_account_head,
validate_conversion_rate,
validate_cost_center,
validate_inclusive_tax,
validate_taxes_and_charges,
)
def update_invoice_status():
@@ -3221,62 +3059,6 @@ def get_supplier_block_status(party_name):
return info
def set_child_tax_template_and_map(item, child_item, parent_doc):
ctx = ItemDetailsCtx(
{
"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=True):
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 []
# add new row for tax head only if missing
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 set_order_defaults(parent_doctype, parent_doctype_name, child_doctype, child_docname, trans_item):
"""
Returns a Sales/Purchase Order Item child item containing the default values
@@ -3830,49 +3612,6 @@ def check_if_child_table_updated(child_table_before_update, child_table_after_up
return False
def merge_taxes(source_doc, target_doc):
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
@erpnext.allow_regional
def validate_regional(doc):
pass