nsinnovations/erpnext
2
0
Fork 0
mirror of https://github.com/frappe/erpnext.git synced 2026-08-16 07:58:38 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(postgres): port point_of_sale get_items to the query builder (partial backport #56153) (#57527)

Browse Source 
Co-authored-by: Mihir Kandoi <kandoimihir@gmail.com>
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
(cherry picked from commit fe6534b888)

# Conflicts:
#	erpnext/selling/page/point_of_sale/point_of_sale.py
This commit is contained in:
Diptanil Saha
2026-07-28 12:17:20 +05:30
committed by Mergify
parent 43a96c3109
commit 2eaf42bf84
2 changed files with 224 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files

146
erpnext/selling/page/point_of_sale/point_of_sale.py
View File

@@ -5,8 +5,13 @@
import json
import frappe
<<<<<<< HEAD
from frappe.query_builder import DocType, Order
from frappe.utils import cint
=======
from frappe.query_builder import Criterion, DocType, Order
from frappe.utils import cint, get_datetime
>>>>>>> fe6534b888 (refactor(postgres): port point_of_sale get_items to the query builder (partial backport #56153) (#57527))
from frappe.utils.nestedset import get_root_of
from erpnext.accounts.doctype.pos_invoice.pos_invoice import get_item_group, get_stock_availability
@@ -146,50 +151,55 @@ def get_items(start, page_length, price_list, item_group, pos_profile, search_te
if not frappe.db.exists("Item Group", item_group):
item_group = get_root_of("Item Group")
condition = get_conditions(search_term)
condition += get_item_group_condition(pos_profile)
lft, rgt = frappe.db.get_value("Item Group", item_group, ["lft", "rgt"])
bin_join_selection, bin_join_condition = "", ""
if hide_unavailable_items:
bin_join_selection = "LEFT JOIN `tabBin` bin ON bin.item_code = item.name"
bin_join_condition = "AND (item.is_stock_item = 0 OR (item.is_stock_item = 1 AND bin.warehouse = %(warehouse)s AND bin.actual_qty > 0))"
item = frappe.qb.DocType("Item")
item_group_dt = frappe.qb.DocType("Item Group")
items_data = frappe.db.sql(
"""
SELECT
item.name AS item_code,
item_group_subquery = (
frappe.qb.from_(item_group_dt)
.select(item_group_dt.name)
.where((item_group_dt.lft >= lft) & (item_group_dt.rgt <= rgt))
)
query = (
frappe.qb.from_(item)
.select(
item.name.as_("item_code"),
item.item_name,
item.description,
item.stock_uom,
item.image AS item_image,
item.image.as_("item_image"),
item.is_stock_item,
item.sales_uom
FROM
`tabItem` item {bin_join_selection}
WHERE
item.disabled = 0
AND item.has_variants = 0
AND item.is_sales_item = 1
AND item.is_fixed_asset = 0
AND item.item_group in (SELECT name FROM `tabItem Group` WHERE lft >= {lft} AND rgt <= {rgt})
AND {condition}
{bin_join_condition}
ORDER BY
item.name asc
LIMIT
{page_length} offset {start}""".format(
start=cint(start),
page_length=cint(page_length),
lft=cint(lft),
rgt=cint(rgt),
condition=condition,
bin_join_selection=bin_join_selection,
bin_join_condition=bin_join_condition,
),
{"warehouse": warehouse},
as_dict=1,
item.sales_uom,
)
.where(
(item.disabled == 0)
& (item.has_variants == 0)
& (item.is_sales_item == 1)
& (item.is_fixed_asset == 0)
& (item.item_group.isin(item_group_subquery))
& get_conditions(search_term, item)
)
)
item_group_condition = get_item_group_condition(pos_profile, item)
if item_group_condition is not None:
query = query.where(item_group_condition)
if hide_unavailable_items:
bin_dt = frappe.qb.DocType("Bin")
query = (
query.left_join(bin_dt)
.on(bin_dt.item_code == item.name)
.where(
(item.is_stock_item == 0)
| ((item.is_stock_item == 1) & (bin_dt.warehouse == warehouse) & (bin_dt.actual_qty > 0))
)
)
items_data = (
query.orderby(item.name, order=Order.asc).limit(cint(page_length)).offset(cint(start)).run(as_dict=1)
)
# return (empty) list if there are no results
@@ -260,54 +270,72 @@ def search_for_serial_or_batch_or_barcode_number(search_value: str) -> dict[str,
return scan_barcode(search_value)
def get_conditions(search_term):
condition = "("
condition += """item.name like {search_term}
or item.item_name like {search_term}""".format(search_term=frappe.db.escape("%" + search_term + "%"))
condition += add_search_fields_condition(search_term)
condition += ")"
def get_conditions(search_term, item=None):
if item is None:
item = frappe.qb.DocType("Item")
return condition
pattern = f"%{search_term}%"
conditions = [item.name.like(pattern), item.item_name.like(pattern)]
conditions += add_search_fields_condition(search_term, item)
return Criterion.any(conditions)
def add_search_fields_condition(search_term):
condition = ""
def add_search_fields_condition(search_term, item=None):
if item is None:
item = frappe.qb.DocType("Item")
pattern = f"%{search_term}%"
conditions = []
search_fields = frappe.get_all("POS Search Fields", fields=["fieldname"])
<<<<<<< HEAD
if search_fields:
for field in search_fields:
condition += " or item.`{}` like {}".format(
field["fieldname"], frappe.db.escape("%" + search_term + "%")
)
return condition
=======
for field in search_fields:
if not field.get("fieldname"):
continue
conditions.append(item[field["fieldname"]].like(pattern))
return conditions
>>>>>>> fe6534b888 (refactor(postgres): port point_of_sale get_items to the query builder (partial backport #56153) (#57527))
def get_item_group_condition(pos_profile):
cond = "and 1=1"
def get_item_group_condition(pos_profile, item=None):
if item is None:
item = frappe.qb.DocType("Item")
item_groups = get_item_groups(pos_profile)
if item_groups:
cond = "and item.item_group in (%s)" % (", ".join(["%s"] * len(item_groups)))
return item.item_group.isin(item_groups)
return cond % tuple(item_groups)
return None
@frappe.whitelist()
@frappe.validate_and_sanitize_search_inputs
def item_group_query(doctype, txt, searchfield, start, page_len, filters):
item_groups = []
cond = "1=1"
pos_profile = filters.get("pos_profile")
item_filters = [["name", "like", f"%{txt}%"]]
if pos_profile:
item_groups = get_item_groups(pos_profile)
if item_groups:
cond = "name in (%s)" % (", ".join(["%s"] * len(item_groups)))
cond = cond % tuple(item_groups)
item_filters.append(["name", "in", item_groups])
return frappe.db.sql(
f""" select distinct name from `tabItem Group`
where {cond} and (name like %(txt)s) limit {page_len} offset {start}""",
{"txt": "%%%s%%" % txt},
return frappe.get_all(
"Item Group",
filters=item_filters,
fields=["name"],
distinct=True,
order_by="", # original raw SQL had no ORDER BY; suppress the injected default (creation desc on MariaDB)
limit_start=start,
limit_page_length=page_len,
as_list=True,
)

137
erpnext/selling/page/point_of_sale/test_point_of_sale.py Normal file
View File

@@ -0,0 +1,137 @@
# Copyright (c) 2024, Frappe Technologies Pvt. Ltd. and Contributors
# License: GNU General Public License v3. See license.txt
import frappe
from frappe.utils import random_string
from erpnext.accounts.doctype.pos_profile.test_pos_profile import make_pos_profile
from erpnext.selling.page.point_of_sale.point_of_sale import get_items
from erpnext.stock.doctype.stock_entry.stock_entry_utils import make_stock_entry
from erpnext.tests.utils import ERPNextTestSuite
class TestPointOfSaleGetItems(ERPNextTestSuite):
"""Covers the raw-SQL -> frappe.qb conversion of point_of_sale.get_items."""
def setUp(self):
super().setUp()
# Reuse the bootstrap leaf item group; an item assigned directly to it
# falls inside its own (lft, rgt) subtree, which is what get_items filters on.
self.item_group = "_Test Item Group"
# A non-stock sales item keeps get_stock_availability cheap (no Bin needed)
# and keeps the item out of the hide_unavailable_items branch.
self.item_code = "_Test POS Item " + random_string(10)
item = frappe.get_doc(
{
"doctype": "Item",
"item_code": self.item_code,
"item_name": self.item_code,
"item_group": self.item_group,
"stock_uom": "_Test UOM",
"is_stock_item": 0,
"is_sales_item": 1,
"is_fixed_asset": 0,
"has_variants": 0,
"disabled": 0,
}
)
item.insert()
self.item = item
# make_pos_profile builds "_Test POS Profile" (hide_unavailable_items unset,
# no item_groups restriction). Rolled back by tearDown.
self.pos_profile = make_pos_profile().name
def _get_item_codes(self, search_term):
result = get_items(
start=0,
page_length=100,
price_list="Standard Selling",
item_group=self.item_group,
pos_profile=self.pos_profile,
search_term=search_term,
)
# get_items returns {"items": [...]} when the qb query yields rows,
# and a bare (empty) list when nothing matches.
items = result["items"] if isinstance(result, dict) else result
return [row.get("item_code") for row in items]
def _make_stock_item(self):
# Fresh stock item in the filtered item group so it passes the
# item_group.isin(subquery) clause and reaches the Bin left-join.
item_code = "_Test POS Stock Item " + random_string(10)
frappe.get_doc(
{
"doctype": "Item",
"item_code": item_code,
"item_name": item_code,
"item_group": self.item_group,
"stock_uom": "_Test UOM",
"is_stock_item": 1,
"is_sales_item": 1,
"is_fixed_asset": 0,
"has_variants": 0,
"disabled": 0,
}
).insert()
return item_code
def test_matching_search_term_returns_item(self):
# search_term matches Item.name / Item.item_name via the LIKE OR-condition;
# scan_barcode finds nothing for this value, so the converted qb query runs.
item_codes = self._get_item_codes(self.item_code)
self.assertIn(self.item_code, item_codes)
def test_non_matching_search_term_excludes_item(self):
non_matching = "zzz_no_such_item_" + random_string(10)
item_codes = self._get_item_codes(non_matching)
self.assertNotIn(self.item_code, item_codes)
def test_partial_search_term_matches_on_item_name(self):
# A substring of the item code must still match (LIKE %term%),
# proving the OR/LIKE clause survived the SQL->qb conversion.
partial = self.item_code.split(" ")[-1]
item_codes = self._get_item_codes(partial)
self.assertIn(self.item_code, item_codes)
def test_disabled_item_is_excluded(self):
# disabled == 0 is part of the converted WHERE clause; flipping it
# must drop the item even when the search term matches.
frappe.db.set_value("Item", self.item_code, "disabled", 1)
item_codes = self._get_item_codes(self.item_code)
self.assertNotIn(self.item_code, item_codes)
def test_non_sales_item_is_excluded(self):
# is_sales_item == 1 is part of the converted WHERE clause.
frappe.db.set_value("Item", self.item_code, "is_sales_item", 0)
item_codes = self._get_item_codes(self.item_code)
self.assertNotIn(self.item_code, item_codes)
def test_hide_unavailable_items_filters_on_bin_actual_qty(self):
# Covers the hide_unavailable_items branch: the Bin left-join only keeps a
# stock item when bin.warehouse == profile warehouse AND bin.actual_qty > 0.
# A second stock item with no Bin row (no stock) must be hidden.
warehouse = frappe.db.get_value("POS Profile", self.pos_profile, "warehouse")
frappe.db.set_value("POS Profile", self.pos_profile, "hide_unavailable_items", 1)
in_stock_item = self._make_stock_item()
out_of_stock_item = self._make_stock_item()
# Material Receipt gives in_stock_item actual_qty > 0 in the profile warehouse;
# out_of_stock_item gets no Bin row at all.
make_stock_entry(item_code=in_stock_item, target=warehouse, qty=5, basic_rate=100)
# Sanity-check the precondition the branch keys off of.
self.assertGreater(
frappe.db.get_value("Bin", {"item_code": in_stock_item, "warehouse": warehouse}, "actual_qty")
or 0,
0,
)
self.assertFalse(frappe.db.exists("Bin", {"item_code": out_of_stock_item}))
in_stock_codes = self._get_item_codes(in_stock_item)
self.assertIn(in_stock_item, in_stock_codes)
out_of_stock_codes = self._get_item_codes(out_of_stock_item)
self.assertNotIn(out_of_stock_item, out_of_stock_codes)