nsinnovations/erpnext
2
0
Fork 0
mirror of https://github.com/frappe/erpnext.git synced 2026-05-29 09:54:47 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor: convert sql queries to qb queries

Browse Source
This commit is contained in:
Sugesh393
2024-12-09 18:32:45 +05:30
parent b9f9fc0afe
commit 7614f166d8
1 changed files with 137 additions and 108 deletions
Download Patch File Download Diff File Expand all files Collapse all files

245
erpnext/accounts/report/customer_ledger_summary/customer_ledger_summary.py
View File

@@ -4,6 +4,7 @@
import frappe import frappe
from frappe import _, qb, scrub from frappe import _, qb, scrub
from frappe.query_builder.functions import IfNull
from frappe.utils import getdate, nowdate from frappe.utils import getdate, nowdate
from erpnext.accounts.doctype.accounting_dimension.accounting_dimension import ( from erpnext.accounts.doctype.accounting_dimension.accounting_dimension import (
@@ -252,102 +253,114 @@ class PartyLedgerSummaryReport:
return out return out
def get_gl_entries(self): def get_gl_entries(self):
conditions = self.prepare_conditions() gle = qb.DocType("GL Entry")
join = join_field = "" query = (
if self.filters.party_type == "Customer": qb.from_(gle)
join_field = ", p.customer_name as party_name" .select(
join = "left join `tabCustomer` p on gle.party = p.name" gle.posting_date,
elif self.filters.party_type == "Supplier": gle.party,
join_field = ", p.supplier_name as party_name" gle.voucher_type,
join = "left join `tabSupplier` p on gle.party = p.name" gle.voucher_no,
gle.against_voucher_type,
self.gl_entries = frappe.db.sql( gle.against_voucher,
f""" gle.debit,
select gle.credit,
gle.posting_date, gle.party, gle.voucher_type, gle.voucher_no, gle.against_voucher_type, gle.is_opening,
gle.against_voucher, gle.debit, gle.credit, gle.is_opening {join_field} )
from `tabGL Entry` gle .where(
{join} (gle.docstatus < 2)
where & (gle.is_cancelled == 0)
gle.docstatus < 2 and gle.is_cancelled = 0 and gle.party_type=%(party_type)s and ifnull(gle.party, '') != '' & (gle.party_type == self.filters.party_type)
and gle.posting_date <= %(to_date)s {conditions} & (IfNull(gle.party, "") != "")
order by gle.posting_date & (gle.posting_date <= self.filters.to_date)
""", )
self.filters, .orderby(gle.posting_date)
as_dict=True,
) )
def prepare_conditions(self): if self.filters.party_type == "Customer":
conditions = [""] ctr = qb.DocType("Customer")
query = query.select(ctr.customer_name.as_("party_name")).left_join(ctr).on(ctr.name == gle.party)
elif self.filters.party_type == "Supplier":
spr = qb.DocType("Supplier")
query = query.select(spr.supplier_name.as_("party_name")).left_join(spr).on(spr.name == gle.party)
query = self.prepare_conditions(query)
self.gl_entries = query.run(as_dict=True)
def prepare_conditions(self, query):
gle = qb.DocType("GL Entry")
if self.filters.company: if self.filters.company:
conditions.append("gle.company=%(company)s") query = query.where(gle.company == self.filters.company)
if self.filters.finance_book: if self.filters.finance_book:
conditions.append("ifnull(finance_book,'') in (%(finance_book)s, '')") query = query.where(IfNull(gle.finance_book, "") == self.filters.finance_book)
if self.filters.get("party"): if self.filters.party:
conditions.append("party=%(party)s") query = query.where(gle.party == self.filters.party)
if self.filters.party_type == "Customer": if self.filters.party_type == "Customer":
if self.filters.get("customer_group"): ctr = qb.DocType("Customer")
lft, rgt = frappe.get_cached_value( if self.filters.customer_group:
"Customer Group", self.filters["customer_group"], ["lft", "rgt"] query = query.where(
(gle.party).isin(
qb.from_(ctr)
.select(ctr.name)
.where(ctr.customer_group == self.filters.customer_group)
)
) )
conditions.append( if self.filters.territory:
f"""party in (select name from tabCustomer query = query.where(
where exists(select name from `tabCustomer Group` where lft >= {lft} and rgt <= {rgt} (gle.party).isin(
and name=tabCustomer.customer_group))""" qb.from_(ctr).select(ctr.name).where(ctr.territory == self.filters.territory)
)
) )
if self.filters.get("territory"): if self.filters.payment_terms_template:
lft, rgt = frappe.db.get_value("Territory", self.filters.get("territory"), ["lft", "rgt"]) query = query.where(
(gle.party).isin(
conditions.append( qb.from_(ctr)
f"""party in (select name from tabCustomer .select(ctr.name)
where exists(select name from `tabTerritory` where lft >= {lft} and rgt <= {rgt} .where(ctr.payment_terms == self.filters.payment_terms_template)
and name=tabCustomer.territory))""" )
) )
if self.filters.get("payment_terms_template"): if self.filters.sales_partner:
conditions.append( query = query.where(
"party in (select name from tabCustomer where payment_terms=%(payment_terms_template)s)" (gle.party).isin(
qb.from_(ctr)
.select(ctr.name)
.where(ctr.default_sales_partner == self.filters.sales_partner)
)
) )
if self.filters.get("sales_partner"): if self.filters.sales_person:
conditions.append( sl_team = qb.DocType("Sales Team")
"party in (select name from tabCustomer where default_sales_partner=%(sales_partner)s)" query = query.where(
) (gle.party).isin(
qb.from_(sl_team)
if self.filters.get("sales_person"): .select(sl_team.parent)
lft, rgt = frappe.db.get_value( .where(sl_team.sales_person == self.filters.sales_person)
"Sales Person", self.filters.get("sales_person"), ["lft", "rgt"] )
)
conditions.append(
"""exists(select name from `tabSales Team` steam where
steam.sales_person in (select name from `tabSales Person` where lft >= {} and rgt <= {})
and ((steam.parent = voucher_no and steam.parenttype = voucher_type)
or (steam.parent = against_voucher and steam.parenttype = against_voucher_type)
or (steam.parent = party and steam.parenttype = 'Customer')))""".format(lft, rgt)
) )
if self.filters.party_type == "Supplier": if self.filters.party_type == "Supplier":
if self.filters.get("supplier_group"): if self.filters.supplier_group:
conditions.append( spr = qb.DocType("Supplier")
"""party in (select name from tabSupplier query = query.where(
where supplier_group=%(supplier_group)s)""" (gle.party).isin(
qb.from_(spr)
.select(spr.name)
.where(spr.supplier_group == self.filters.supplier_group)
)
) )
if self.filters.get("cost_center"): if self.filters.cost_center:
self.filters.cost_center = frappe.parse_json(self.filters.get("cost_center"))
self.filters.cost_center = get_cost_centers_with_children(self.filters.cost_center) self.filters.cost_center = get_cost_centers_with_children(self.filters.cost_center)
conditions.append("gle.cost_center in %(cost_center)s") query = query.where((gle.cost_center).isin(self.filters.cost_center))
if self.filters.get("project"): if self.filters.project:
self.filters.project = frappe.parse_json(self.filters.get("project")) query = query.where((gle.project).isin(self.filters.project))
conditions.append("gle.project in %(project)s")
accounting_dimensions = get_accounting_dimensions(as_list=False) accounting_dimensions = get_accounting_dimensions(as_list=False)
@@ -358,11 +371,15 @@ class PartyLedgerSummaryReport:
self.filters[dimension.fieldname] = get_dimension_with_children( self.filters[dimension.fieldname] = get_dimension_with_children(
dimension.document_type, self.filters.get(dimension.fieldname) dimension.document_type, self.filters.get(dimension.fieldname)
) )
conditions.append(f"{dimension.fieldname} in %({dimension.fieldname})s") query = query.where(
(gle[dimension.fieldname]).isin(self.filters.get(dimension.fieldname))
)
else: else:
conditions.append(f"{dimension.fieldname} in %({dimension.fieldname})s") query = query.where(
(gle[dimension.fieldname]).isin(self.filters.get(dimension.fieldname))
)
return " and ".join(conditions) return query
def get_return_invoices(self): def get_return_invoices(self):
doctype = "Sales Invoice" if self.filters.party_type == "Customer" else "Purchase Invoice" doctype = "Sales Invoice" if self.filters.party_type == "Customer" else "Purchase Invoice"
@@ -379,53 +396,45 @@ class PartyLedgerSummaryReport:
] ]
def get_party_adjustment_amounts(self): def get_party_adjustment_amounts(self):
conditions = self.prepare_conditions()
account_type = "Expense Account" if self.filters.party_type == "Customer" else "Income Account" account_type = "Expense Account" if self.filters.party_type == "Customer" else "Income Account"
income_or_expense_accounts = frappe.db.get_all( self.income_or_expense_accounts = frappe.db.get_all(
"Account", filters={"account_type": account_type, "company": self.filters.company}, pluck="name" "Account", filters={"account_type": account_type, "company": self.filters.company}, pluck="name"
) )
invoice_dr_or_cr = "debit" if self.filters.party_type == "Customer" else "credit" invoice_dr_or_cr = "debit" if self.filters.party_type == "Customer" else "credit"
reverse_dr_or_cr = "credit" if self.filters.party_type == "Customer" else "debit" reverse_dr_or_cr = "credit" if self.filters.party_type == "Customer" else "debit"
round_off_account = frappe.get_cached_value("Company", self.filters.company, "round_off_account") round_off_account = frappe.get_cached_value("Company", self.filters.company, "round_off_account")
gl = qb.DocType("GL Entry") if not self.income_or_expense_accounts:
if not income_or_expense_accounts:
# prevent empty 'in' condition # prevent empty 'in' condition
income_or_expense_accounts.append("") self.income_or_expense_accounts.append("")
else: else:
# escape '%' in account name # escape '%' in account name
# ignoring frappe.db.escape as it replaces single quotes with double quotes # ignoring frappe.db.escape as it replaces single quotes with double quotes
income_or_expense_accounts = [x.replace("%", "%%") for x in income_or_expense_accounts] self.income_or_expense_accounts = [x.replace("%", "%%") for x in self.income_or_expense_accounts]
accounts_query = ( gl = qb.DocType("GL Entry")
accounts_query = self.get_base_accounts_query()
accounts_query_voucher_no = accounts_query.select(gl.voucher_no)
accounts_query_voucher_type = accounts_query.select(gl.voucher_type)
subquery = self.get_base_subquery()
subquery_voucher_no = subquery.select(gl.voucher_no)
subquery_voucher_type = subquery.select(gl.voucher_type)
gl_entries = (
qb.from_(gl) qb.from_(gl)
.select(gl.voucher_type, gl.voucher_no) .select(
.where( gl.posting_date, gl.account, gl.party, gl.voucher_type, gl.voucher_no, gl.debit, gl.credit
(gl.account.isin(income_or_expense_accounts))
& (gl.posting_date.gte(self.filters.from_date))
& (gl.posting_date.lte(self.filters.to_date))
) )
) .where(
(gl.docstatus < 2)
gl_entries = frappe.db.sql( & (gl.is_cancelled == 0)
f""" & (gl.voucher_no.isin(accounts_query_voucher_no))
select & (gl.voucher_type.isin(accounts_query_voucher_type))
posting_date, account, party, voucher_type, voucher_no, debit, credit & (gl.voucher_no.isin(subquery_voucher_no))
from & (gl.voucher_type.isin(subquery_voucher_type))
`tabGL Entry` )
where ).run(as_dict=True)
docstatus < 2 and is_cancelled = 0
and (voucher_type, voucher_no) in (
{accounts_query}
) and (voucher_type, voucher_no) in (
select voucher_type, voucher_no from `tabGL Entry` gle
where gle.party_type=%(party_type)s and ifnull(party, '') != ''
and gle.posting_date between %(from_date)s and %(to_date)s and gle.docstatus < 2 {conditions}
)
""",
self.filters,
as_dict=True,
)
self.party_adjustment_details = {} self.party_adjustment_details = {}
self.party_adjustment_accounts = set() self.party_adjustment_accounts = set()
@@ -467,6 +476,26 @@ class PartyLedgerSummaryReport:
self.party_adjustment_details[party].setdefault(account, 0) self.party_adjustment_details[party].setdefault(account, 0)
self.party_adjustment_details[party][account] += amount self.party_adjustment_details[party][account] += amount
def get_base_accounts_query(self):
gl = qb.DocType("GL Entry")
query = qb.from_(gl).where(
(gl.account.isin(self.income_or_expense_accounts))
& (gl.posting_date.gte(self.filters.from_date))
& (gl.posting_date.lte(self.filters.to_date))
)
return query
def get_base_subquery(self):
gl = qb.DocType("GL Entry")
query = qb.from_(gl).where(
(gl.docstatus < 2)
& (gl.party_type == self.filters.party_type)
& (IfNull(gl.party, "") != "")
& (gl.posting_date.between(self.filters.from_date, self.filters.to_date))
)
query = self.prepare_conditions(query)
return query
def execute(filters=None): def execute(filters=None):
args = { args = {