diff --git a/erpnext/stock/serial_batch_identity.py b/erpnext/stock/serial_batch_identity.py index 9e4a995e574..9b37ab39f7b 100644 --- a/erpnext/stock/serial_batch_identity.py +++ b/erpnext/stock/serial_batch_identity.py @@ -6,6 +6,8 @@ from frappe.model.naming import make_autoname from frappe.query_builder.functions import Coalesce, Count, Lower, NullIf from frappe.utils import cstr, now +from erpnext.stock.serial_batch_number_lookup import SerialBatchNumberLookup + class SerialBatchIdentity: def __init__(self, doctype): @@ -15,7 +17,7 @@ class SerialBatchIdentity: "Batch": ("item", "batch_id"), }[doctype] - def resolve(self, item_code, numbers, *, create=False, defaults=None): + def resolve(self, item_code, numbers, *, create=False, defaults=None, check_permissions=False): """Return IDs in input order. A physical number is never looked up as a document ID.""" if not isinstance(numbers, list | tuple) or any(not isinstance(number, str) for number in numbers): frappe.throw(_("Physical numbers must be a list of strings")) @@ -25,23 +27,29 @@ class SerialBatchIdentity: if not isinstance(item_code, str) or not item_code or any(not number for number in numbers): frappe.throw(_("Item and physical number are required")) - records = self.get_query(numbers, item_code).run(as_dict=True) - ids = {row[self.number_field]: row.name for row in records} - missing = [] - for number in dict.fromkeys(numbers): - if number in ids: - continue - # Use the database comparison rules, including its collation, for exact lookups. - name = self.exists(number, item_code) if records else None - if not name and create: - missing.append(number) - continue - if not name: - frappe.throw(_("{0} {1} does not exist for Item {2}").format(self.doctype, number, item_code)) - ids[number] = name + if check_permissions: + permission = "select" if frappe.only_has_select_perm(self.doctype) else "read" + frappe.has_permission(self.doctype, permission, throw=True) + lookup = SerialBatchNumberLookup(self, item_code, numbers) + missing = lookup.missing if missing: - ids.update(self.resolve_missing(item_code, missing, defaults)) - return [ids[number] for number in numbers] + if not create: + frappe.throw( + _("{0} {1} does not exist for Item {2}").format(self.doctype, missing[0], item_code) + ) + if check_permissions: + frappe.has_permission(self.doctype, "create", throw=True) + lookup.assign(self.resolve_missing(item_code, missing, defaults)) + names = [lookup.ids[number] for number in numbers] + if check_permissions: + allowed = frappe.get_list( + self.doctype, filters={"name": ("in", names)}, pluck="name", limit_page_length=0 + ) + if set(names) - set(allowed): + frappe.throw( + _("Not permitted to select these serial or batch records"), frappe.PermissionError + ) + return names def resolve_missing(self, item_code, numbers, defaults): savepoint = "serial_batch_resolve_" + frappe.generate_hash(length=10) @@ -51,17 +59,16 @@ class SerialBatchIdentity: except Exception as error: frappe.db.rollback(save_point=savepoint) frappe.db.release_savepoint(savepoint) - if not isinstance(error, frappe.DuplicateEntryError): + if not isinstance(error, frappe.DuplicateEntryError | frappe.UniqueValidationError): raise else: frappe.db.release_savepoint(savepoint) return ids - # Retry aliases using the database's comparison rules, including MariaDB collation. - return { - number: self.exists(number, item_code) or self.create_many(item_code, [number], defaults)[number] - for number in numbers - } + lookup = SerialBatchNumberLookup(self, item_code, numbers) + if lookup.missing: + lookup.assign(self.create_many(item_code, lookup.missing, defaults)) + return lookup.ids def get_query(self, numbers, item_code=None, *, fields=None, filters=None, ignore_permissions=True): table = frappe.qb.DocType(self.doctype) @@ -86,10 +93,8 @@ class SerialBatchIdentity: def create_many(self, item_code, numbers, defaults=None): if self.doctype == "Batch": - return { - number: self.exists(number, item_code) or self.create_batch(item_code, number, defaults) - for number in numbers - } + # New batches still need their expiry and valuation hooks. + return {number: self.create_batch(item_code, number, defaults) for number in numbers} # Inactive serials can be prepared before their first receipt assigns a company. item = frappe.get_cached_value( @@ -129,6 +134,7 @@ class SerialBatchIdentity: doc.update(defaults or {}) doc.item = item_code doc.batch_id = number + doc.flags.serial_batch_number_checked = True doc.insert(ignore_permissions=True) return doc.name @@ -147,7 +153,11 @@ class SerialBatchIdentity: def validate(self, doc): number = cstr(doc.get(self.number_field)).strip() doc.set(self.number_field, number) - if number and self.exists(number, doc.get(self.item_field), exclude=doc.name): + if ( + number + and not (doc.is_new() and doc.flags.serial_batch_number_checked) + and self.exists(number, doc.get(self.item_field), exclude=doc.name) + ): frappe.throw( _("{0} {1} already exists for Item {2}").format( self.doctype, number, doc.get(self.item_field) @@ -241,18 +251,9 @@ def resolve_serial_batch_numbers( ("Batch", batch_numbers, "batch_nos"), ): values = frappe.parse_json(values) or [] - if values: - permission = "create" if create else "select" if frappe.only_has_select_perm(doctype) else "read" - frappe.has_permission(doctype, permission, throw=True) - result[key] = SerialBatchIdentity(doctype).resolve(item_code, values, create=create) - if values and not create: - allowed = frappe.get_list( - doctype, filters={"name": ("in", result[key])}, pluck="name", limit_page_length=0 - ) - if set(result[key]) - set(allowed): - frappe.throw( - _("Not permitted to select these serial or batch records"), frappe.PermissionError - ) + result[key] = SerialBatchIdentity(doctype).resolve( + item_code, values, create=create, check_permissions=True + ) return result @@ -314,8 +315,6 @@ def resolve_number_entries(item_code, entries, *, create=False): ("serial_no", "Serial No", "serial_number"), ): rows = [row for row in entries if row.get(number_field) and not row.get(field)] - identity = SerialBatchIdentity(doctype) - missing = create and any(not identity.exists(row[number_field], item_code) for row in rows) ids = ( resolve_serial_batch_numbers( item_code, @@ -324,7 +323,7 @@ def resolve_number_entries(item_code, entries, *, create=False): row[number_field] for row in rows ] }, - create=missing, + create=create, )["serial_nos" if doctype == "Serial No" else "batch_nos"] if rows else [] diff --git a/erpnext/stock/serial_batch_input.py b/erpnext/stock/serial_batch_input.py index 73283828de2..8520742777e 100644 --- a/erpnext/stock/serial_batch_input.py +++ b/erpnext/stock/serial_batch_input.py @@ -2,7 +2,7 @@ import frappe from frappe import _ from erpnext.stock.serial_batch_fields import NUMBER_INPUT_DOCTYPES -from erpnext.stock.serial_batch_identity import SerialBatchIdentity, resolve_serial_batch_numbers +from erpnext.stock.serial_batch_identity import SerialBatchIdentity def resolve_transaction_numbers(doc, method=None): @@ -48,22 +48,13 @@ class TransactionNumberInput: if not self.item_code: frappe.throw(_("Item is required")) frappe.has_permission("Item", "read", doc=self.item_code, throw=True) - existing = { - record[identity.number_field] - for record in identity.get_query(numbers, self.item_code).run(as_dict=True) - } - missing = any( - number not in existing and not identity.exists(number, self.item_code) for number in numbers + names = identity.resolve( + self.item_code, + numbers, + create=self.can_create(field), + defaults={"company": self.doc.get("company")}, + check_permissions=True, ) - if missing and self.can_create(field): - frappe.has_permission(doctype, "create", throw=True) - identity.resolve( - self.item_code, numbers, create=True, defaults={"company": self.doc.get("company")} - ) - key = "batch_numbers" if field == "batch_no" else "serial_numbers" - names = resolve_serial_batch_numbers(self.item_code, **{key: numbers})[ - "batch_nos" if field == "batch_no" else "serial_nos" - ] if doctype == "Serial No" and len(set(names)) != len(names): frappe.throw(_("A serial number cannot appear twice in the same row")) return names diff --git a/erpnext/stock/serial_batch_number_lookup.py b/erpnext/stock/serial_batch_number_lookup.py new file mode 100644 index 00000000000..b869ef42548 --- /dev/null +++ b/erpnext/stock/serial_batch_number_lookup.py @@ -0,0 +1,61 @@ +import frappe +from frappe.query_builder.terms import ParameterizedValueWrapper +from pypika.analytics import Min + + +class SerialBatchNumberLookup: + """Match inputs and group aliases using the physical column's database collation.""" + + def __init__(self, identity, item_code, numbers): + self.identity = identity + self.item_code = item_code + self.numbers = list(dict.fromkeys(numbers)) + self.ids = {} + self.aliases = {} + self.load() + + @property + def missing(self): + return [ + number for number in self.numbers if number not in self.ids and self.aliases[number] == number + ] + + def load(self): + table = frappe.qb.DocType(self.identity.doctype) + inputs = self.get_inputs(table) + key = self.identity.number_key + rows = ( + frappe.qb.from_(inputs) + .left_join(table) + .on( + (table[self.identity.item_field] == self.item_code) + & (key(table[self.identity.number_field]) == key(inputs.number)) + ) + .select(inputs.ordinal, table.name, Min(inputs.ordinal).over(key(inputs.number))) + ).run() + for index, name, first_index in rows: + number = self.numbers[index] + self.aliases[number] = self.numbers[first_index] + if name: + self.ids[number] = name + + def get_inputs(self, table): + # An empty column select preserves MariaDB's physical-number collation in the union. + inputs = ( + frappe.qb.from_(table) + .select( + table[self.identity.number_field].as_("number"), ParameterizedValueWrapper(-1).as_("ordinal") + ) + .where(table.name.isnull()) + ) + for index, number in enumerate(self.numbers): + inputs = inputs.union_all( + frappe.qb.select( + ParameterizedValueWrapper(number).as_("number"), + ParameterizedValueWrapper(index).as_("ordinal"), + ) + ) + return inputs.as_("numbers") + + def assign(self, names): + self.ids.update({number: names[alias] for number, alias in self.aliases.items() if alias in names}) diff --git a/erpnext/stock/tests/test_serial_batch_number_lookup.py b/erpnext/stock/tests/test_serial_batch_number_lookup.py new file mode 100644 index 00000000000..8003874b308 --- /dev/null +++ b/erpnext/stock/tests/test_serial_batch_number_lookup.py @@ -0,0 +1,121 @@ +from contextlib import contextmanager +from unittest.mock import patch + +import frappe +from frappe.utils import add_days, getdate, today + +from erpnext.stock.doctype.item.test_item import make_item +from erpnext.stock.doctype.purchase_receipt.test_purchase_receipt import make_purchase_receipt +from erpnext.stock.serial_batch_identity import SerialBatchIdentity, resolve_number_entries +from erpnext.stock.serial_batch_input import TransactionNumberInput +from erpnext.stock.serial_batch_number_lookup import SerialBatchNumberLookup +from erpnext.tests.utils import ERPNextTestSuite + + +class TestSerialBatchNumberLookup(ERPNextTestSuite): + def make_item_and_identity(self, doctype): + item = make_item(properties={"has_serial_no": 1, "has_batch_no": int(doctype == "Batch")}) + identity = SerialBatchIdentity(doctype) + identity.resolve(item.name, ["Existing"], create=True) + return item, identity + + def test_existing_number_aliases_use_one_query(self): + for doctype in ("Serial No", "Batch"): + item, identity = self.make_item_and_identity(doctype) + numbers = [f"Number-{index:03d}" for index in range(50)] + ids = identity.resolve(item.name, numbers, create=True) + requested = [number.swapcase() for number in reversed(numbers)] + [numbers[0].upper()] + with self.assert_select_query_count(1): + resolved = identity.resolve(item.name, requested, create=True) + self.assertEqual(resolved, [*reversed(ids), ids[0]]) + + def test_new_serial_aliases_share_one_lookup_and_bulk_insert(self): + item, identity = self.make_item_and_identity("Serial No") + numbers = [f"New-{index:03d}" for index in range(100)] + with patch.object(frappe.db, "bulk_insert", wraps=frappe.db.bulk_insert) as insert: + with self.assert_select_query_count(1): + ids = identity.resolve( + item.name, numbers + [number.upper() for number in numbers], create=True + ) + insert.assert_called_once() + self.assertEqual(ids[:100], ids[100:]) + self.assertEqual(len(set(ids)), 100) + self.assertEqual(identity.labels([ids[0]]), {ids[0]: numbers[0]}) + + def test_duplicate_retry_remains_batched(self): + item, identity = self.make_item_and_identity("Serial No") + numbers = [f"Retry-{index:03d}" for index in range(50)] + create_many = identity.create_many + attempts = [] + + def create(item_code, requested, defaults): + attempts.append(requested) + if len(attempts) == 1: + raise frappe.DuplicateEntryError + return create_many(item_code, requested, defaults) + + with patch.object(identity, "create_many", side_effect=create): + with self.assert_select_query_count(2): + ids = identity.resolve( + item.name, numbers + [number.upper() for number in numbers], create=True + ) + self.assertEqual(attempts, [numbers, numbers]) + self.assertEqual(ids[:50], ids[50:]) + + def test_batch_creation_keeps_lifecycle_validation(self): + item = make_item(properties={"has_batch_no": 1, "has_expiry_date": 1, "shelf_life_in_days": 30}) + identity = SerialBatchIdentity("Batch") + with patch.object(SerialBatchIdentity, "exists", side_effect=AssertionError("Per-number lookup")): + names = identity.resolve( + item.name, + ["Batch-One", "BATCH-ONE", "Batch-Two"], + create=True, + defaults={"manufacturing_date": today()}, + ) + self.assertEqual(names[0], names[1]) + self.assertNotEqual(names[0], names[2]) + batch = frappe.get_doc("Batch", names[0]) + self.assertEqual(getdate(batch.expiry_date), getdate(add_days(today(), 30))) + self.assertEqual(batch.use_batchwise_valuation, 1) + + def test_prechecked_batch_still_has_database_uniqueness(self): + item, identity = self.make_item_and_identity("Batch") + frappe.db.savepoint("prechecked_batch") + try: + with self.assertRaises((frappe.DuplicateEntryError, frappe.UniqueValidationError)): + identity.create_batch(item.name, "EXISTING") + finally: + frappe.db.rollback(save_point="prechecked_batch") + + def test_transaction_and_bundle_resolution_reuse_the_lookup(self): + item, identity = self.make_item_and_identity("Serial No") + numbers = [f"Selected-{index:03d}" for index in range(50)] + names = identity.resolve(item.name, numbers, create=True) + receipt = make_purchase_receipt(item_code=item.name, qty=50, do_not_save=True) + row = receipt.items[0] + row.serial_number = "\n".join(number.lower() for number in numbers) + load = SerialBatchNumberLookup.load + with patch.object(SerialBatchNumberLookup, "load", autospec=True, side_effect=load) as lookup: + TransactionNumberInput(receipt, row).resolve() + lookup.assert_called_once() + self.assertEqual(row.serial_no.splitlines(), names) + entries = [{"serial_number": number.lower()} for number in numbers] + with patch.object(SerialBatchNumberLookup, "load", autospec=True, side_effect=load) as lookup: + resolve_number_entries(item.name, entries, create=True) + lookup.assert_called_once() + self.assertEqual([entry["serial_no"] for entry in entries], names) + + def test_bound_numbers_do_not_become_sql(self): + item, identity = self.make_item_and_identity("Serial No") + numbers = ["Serial'One", "100%_Matched", 'A"B'] + names = identity.resolve(item.name, numbers, create=True) + self.assertEqual(identity.resolve(item.name, numbers), names) + self.assertEqual(identity.labels(names), dict(zip(names, numbers, strict=True))) + + @contextmanager + def assert_select_query_count(self, count): + with patch.object(frappe.db, "sql", wraps=frappe.db.sql) as sql: + yield + queries = [str(call.args[0]) for call in sql.call_args_list] + selects = [query for query in queries if query.lstrip().lower().startswith("select")] + self.assertEqual(len(selects), count, "\n".join(selects))