Shiloh/brotherton-erpnext
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

refactor: Use QB for serial fetching query

Browse Source
This commit is contained in:
Ankush Menat 2022-03-19 19:10:57 +05:30 committed by Ankush Menat
parent a585dff6fd
commit 4b695915f4
2 changed files with 39 additions and 30 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

55
erpnext/stock/doctype/serial_no/serial_no.py
View File

@ -7,6 +7,7 @@ import json
import frappe
from frappe import ValidationError, _
from frappe.model.naming import make_autoname
from frappe.query_builder.functions import Coalesce
from frappe.utils import (
add_days,
cint,
@ -626,37 +627,37 @@ def get_pos_reserved_serial_nos(filters):
def fetch_serial_numbers(filters, qty, do_not_include=None):
if do_not_include is None:
do_not_include = []
batch_join_selection = ""
batch_no_condition = ""
batch_nos = filters.get("batch_no")
expiry_date = filters.get("expiry_date")
serial_no = frappe.qb.DocType("Serial No")
query = (
frappe.qb
.from_(serial_no)
.select(serial_no.name)
.where(
(serial_no.item_code == filters["item_code"])
& (serial_no.warehouse == filters["warehouse"])
& (Coalesce(serial_no.sales_invoice, "") == "")
& (Coalesce(serial_no.delivery_document_no, "") == "")
)
.orderby(serial_no.creation)
.limit(qty or 1)
)
if do_not_include:
query = query.where(serial_no.name.notin(do_not_include))
if batch_nos:
batch_no_condition = """and sr.batch_no in ({}) """.format(', '.join("'%s'" % d for d in batch_nos))
query = query.where(serial_no.batch_no.isin(batch_nos))
if expiry_date:
batch_join_selection = "LEFT JOIN `tabBatch` batch on sr.batch_no = batch.name "
expiry_date_cond = "AND ifnull(batch.expiry_date, '2500-12-31') >= %(expiry_date)s "
batch_no_condition += expiry_date_cond
excluded_sr_nos = ", ".join(["" + frappe.db.escape(sr) + "" for sr in do_not_include]) or "''"
serial_numbers = frappe.db.sql("""
SELECT sr.name FROM `tabSerial No` sr {batch_join_selection}
WHERE
sr.name not in ({excluded_sr_nos}) AND
sr.item_code = %(item_code)s AND
sr.warehouse = %(warehouse)s AND
ifnull(sr.sales_invoice,'') = '' AND
ifnull(sr.delivery_document_no, '') = ''
{batch_no_condition}
ORDER BY
sr.creation
LIMIT
{qty}
""".format(
excluded_sr_nos=excluded_sr_nos,
qty=qty or 1,
batch_join_selection=batch_join_selection,
batch_no_condition=batch_no_condition
), filters, as_dict=1)
batch = frappe.qb.DocType("Batch")
query = (query
.left_join(batch).on(serial_no.batch_no == batch.name)
.where(Coalesce(batch.expiry_date, "4000-12-31") >= expiry_date)
)
serial_numbers = query.run(as_dict=True)
return serial_numbers

14
erpnext/stock/doctype/serial_no/test_serial_no.py
View File

@ -256,14 +256,17 @@ class TestSerialNo(FrappeTestCase):
in1.reload()
in2.reload()
batch1 = in1.items[0].batch_no
batch2 = in2.items[0].batch_no
batch_wise_serials = {
in1.items[0].batch_no: get_serial_nos(in1.items[0].serial_no),
in2.items[0].batch_no: get_serial_nos(in2.items[0].serial_no)
batch1 : get_serial_nos(in1.items[0].serial_no),
batch2: get_serial_nos(in2.items[0].serial_no)
}
# Test FIFO
first_fetch = auto_fetch_serial_number(5, item_code, warehouse)
self.assertEqual(first_fetch, batch_wise_serials[in1.items[0].batch_no])
self.assertEqual(first_fetch, batch_wise_serials[batch1])
# partial FIFO
partial_fetch = auto_fetch_serial_number(2, item_code, warehouse)
@ -286,3 +289,8 @@ class TestSerialNo(FrappeTestCase):
all_serials = [sr for sr_list in batch_wise_serials.values() for sr in sr_list]
fetched_serials = auto_fetch_serial_number(10, item_code, warehouse, batch_nos=list(batch_wise_serials.keys()))
self.assertEqual(sorted(all_serials), fetched_serials)
# expiry date
frappe.db.set_value("Batch", batch1, "expiry_date", "1980-01-01")
non_expired_serials = auto_fetch_serial_number(5, item_code, warehouse, posting_date="2021-01-01", batch_nos=batch1)
self.assertEqual(non_expired_serials, [])