From 4affdd51f6b56c4f889c9f0423f8c79fb6a0b380 Mon Sep 17 00:00:00 2001 From: Nabin Hait Date: Thu, 4 Jun 2026 15:35:04 +0530 Subject: [PATCH] refactor(stock): extract StockLedgerService from StockController Move SLE building and reposting (get_sl_entries, update_inventory_dimensions, get_stock_ledger_details, get_items_and_warehouses, make_sl_entries, repost_future_sle_and_gle) into erpnext/stock/services/stock_ledger.py as a delegating service. All six keep thin controller delegators (each has external callers). The repost helper *functions* stay module-level in stock_controller (imported widely); the service calls them. Also drop import orphaned by this and the prior bundle extraction. Behaviour-preserving: ledger characterization snapshots and the repost item valuation suite stay green. --- erpnext/controllers/stock_controller.py | 221 ++------------------- erpnext/stock/services/stock_ledger.py | 250 ++++++++++++++++++++++++ 2 files changed, 264 insertions(+), 207 deletions(-) create mode 100644 erpnext/stock/services/stock_ledger.py diff --git a/erpnext/controllers/stock_controller.py b/erpnext/controllers/stock_controller.py index 833e28876a5..098babd6738 100644 --- a/erpnext/controllers/stock_controller.py +++ b/erpnext/controllers/stock_controller.py @@ -14,7 +14,7 @@ from erpnext.accounts.general_ledger import ( make_gl_entries, make_reverse_gl_entries, ) -from erpnext.accounts.utils import cancel_exchange_gain_loss_journal, get_fiscal_year +from erpnext.accounts.utils import cancel_exchange_gain_loss_journal from erpnext.controllers.accounts_controller import AccountsController from erpnext.controllers.sales_and_purchase_return import ( available_serial_batch_for_return, @@ -24,15 +24,7 @@ from erpnext.controllers.sales_and_purchase_return import ( from erpnext.setup.doctype.brand.brand import get_brand_defaults from erpnext.setup.doctype.item_group.item_group import get_item_group_defaults from erpnext.stock import get_warehouse_account_map -from erpnext.stock.doctype.batch.batch import get_batch_qty -from erpnext.stock.doctype.inventory_dimension.inventory_dimension import ( - get_evaluated_inventory_dimension, -) from erpnext.stock.doctype.item.item import get_item_defaults -from erpnext.stock.doctype.serial_and_batch_bundle.serial_and_batch_bundle import ( - combine_datetime, - get_type_of_transaction, -) from erpnext.stock.stock_ledger import get_items_to_be_repost @@ -296,59 +288,14 @@ class StockController(AccountsController): return details def get_items_and_warehouses(self) -> tuple[list[str], list[str]]: - """Get list of items and warehouses affected by a transaction""" + from erpnext.stock.services.stock_ledger import StockLedgerService - if not (hasattr(self, "items") or hasattr(self, "packed_items")): - return [], [] - - item_rows = (self.get("items") or []) + (self.get("packed_items") or []) - - items = {d.item_code for d in item_rows if d.item_code} - - warehouses = set() - for d in item_rows: - if d.get("warehouse"): - warehouses.add(d.warehouse) - - if self.doctype == "Stock Entry": - if d.get("s_warehouse"): - warehouses.add(d.s_warehouse) - if d.get("t_warehouse"): - warehouses.add(d.t_warehouse) - - return list(items), list(warehouses) + return StockLedgerService(self).get_items_and_warehouses() def get_stock_ledger_details(self): - stock_ledger = {} + from erpnext.stock.services.stock_ledger import StockLedgerService - table = frappe.qb.DocType("Stock Ledger Entry") - - stock_ledger_entries = ( - frappe.qb.from_(table) - .select( - table.name, - table.warehouse, - table.stock_value_difference, - table.valuation_rate, - table.voucher_detail_no, - table.item_code, - table.posting_date, - table.posting_time, - table.actual_qty, - table.qty_after_transaction, - table.project, - ) - .where( - (table.voucher_type == self.doctype) - & (table.voucher_no == self.name) - & (table.is_cancelled == 0) - ) - ).run(as_dict=True) - - for sle in stock_ledger_entries: - stock_ledger.setdefault(sle.voucher_detail_no, []).append(sle) - - return stock_ledger + return StockLedgerService(self).get_stock_ledger_details() def check_expense_account(self, item): if not item.get("expense_account"): @@ -409,41 +356,9 @@ class StockController(AccountsController): ) def get_sl_entries(self, d, args): - sl_dict = frappe._dict( - { - "item_code": d.get("item_code", None), - "warehouse": d.get("warehouse", None), - "serial_and_batch_bundle": d.get("serial_and_batch_bundle"), - "posting_date": self.posting_date, - "posting_time": self.posting_time, - "fiscal_year": get_fiscal_year(self.posting_date, company=self.company)[0], - "voucher_type": self.doctype, - "voucher_no": self.name, - "voucher_detail_no": d.name, - "actual_qty": (self.docstatus == 1 and 1 or -1) * flt(d.get("stock_qty")), - "stock_uom": frappe.get_cached_value( - "Item", args.get("item_code") or d.get("item_code"), "stock_uom" - ), - "incoming_rate": 0, - "company": self.company, - "project": d.get("project") or self.get("project"), - "is_cancelled": 1 if self.docstatus == 2 else 0, - } - ) + from erpnext.stock.services.stock_ledger import StockLedgerService - sl_dict.update(args) - self.update_inventory_dimensions(d, sl_dict) - - if self.docstatus == 2: - from erpnext.deprecation_dumpster import deprecation_warning - - deprecation_warning("unknown", "v16", "No instructions.") - # To handle denormalized serial no records, will br deprecated in v16 - for field in ["serial_no", "batch_no"]: - if d.get(field): - sl_dict[field] = d.get(field) - - return sl_dict + return StockLedgerService(self).get_sl_entries(d, args) def set_landed_cost_voucher_amount(self): for d in self.get("items"): @@ -531,104 +446,17 @@ class StockController(AccountsController): return item_account_wise_cost def update_inventory_dimensions(self, row, sl_dict) -> None: - # To handle delivery note and sales invoice - if row.get("item_row"): - row = row.get("item_row") + from erpnext.stock.services.stock_ledger import StockLedgerService - dimensions = get_evaluated_inventory_dimension(row, sl_dict, parent_doc=self) - for dimension in dimensions: - if not dimension: - continue - - if ( - self.doctype in ["Purchase Invoice", "Purchase Receipt"] - and row.get("rejected_warehouse") - and sl_dict.get("warehouse") == row.get("rejected_warehouse") - ): - fieldname = f"rejected_{dimension.source_fieldname}" - sl_dict[dimension.target_fieldname] = row.get(fieldname) - continue - - if self.doctype in [ - "Purchase Invoice", - "Purchase Receipt", - "Sales Invoice", - "Delivery Note", - "Stock Entry", - ]: - if ( - ( - sl_dict.actual_qty > 0 - and not self.get("is_return") - or sl_dict.actual_qty < 0 - and self.get("is_return") - ) - and self.doctype in ["Purchase Invoice", "Purchase Receipt", "Stock Entry"] - ) or ( - ( - sl_dict.actual_qty < 0 - and not self.get("is_return") - or sl_dict.actual_qty > 0 - and self.get("is_return") - ) - and self.doctype in ["Sales Invoice", "Delivery Note", "Stock Entry"] - ): - if self.doctype == "Stock Entry": - if row.get("t_warehouse") == sl_dict.warehouse and sl_dict.get("actual_qty") > 0: - fieldname = f"to_{dimension.source_fieldname}" - if dimension.source_fieldname.startswith("to_"): - fieldname = f"{dimension.source_fieldname}" - - sl_dict[dimension.target_fieldname] = row.get(fieldname) - continue - - sl_dict[dimension.target_fieldname] = row.get(dimension.source_fieldname) - else: - fieldname_start_with = "to" - if self.doctype in ["Purchase Invoice", "Purchase Receipt"]: - fieldname_start_with = "from" - - fieldname = f"{fieldname_start_with}_{dimension.source_fieldname}" - sl_dict[dimension.target_fieldname] = row.get(fieldname) - - if not sl_dict.get(dimension.target_fieldname): - sl_dict[dimension.target_fieldname] = row.get(dimension.source_fieldname) - - elif row.get(dimension.source_fieldname): - sl_dict[dimension.target_fieldname] = row.get(dimension.source_fieldname) - - if not sl_dict.get(dimension.target_fieldname) and dimension.fetch_from_parent: - sl_dict[dimension.target_fieldname] = self.get(dimension.fetch_from_parent) - - # Get value based on doctype name - if not sl_dict.get(dimension.target_fieldname): - fieldname = next( - ( - field.fieldname - for field in frappe.get_meta(self.doctype).fields - if field.options == dimension.fetch_from_parent - ), - None, - ) - - if fieldname and self.get(fieldname): - sl_dict[dimension.target_fieldname] = self.get(fieldname) - - if sl_dict[dimension.target_fieldname] and self.docstatus == 1: - row.db_set(dimension.source_fieldname, sl_dict[dimension.target_fieldname]) + return StockLedgerService(self).update_inventory_dimensions(row, sl_dict) def make_sl_entries(self, sl_entries, allow_negative_stock=False, via_landed_cost_voucher=False): - from erpnext.stock.serial_batch_bundle import update_batch_qty - from erpnext.stock.services.serial_batch_bundle import SerialBatchBundleService - from erpnext.stock.stock_ledger import make_sl_entries + from erpnext.stock.services.stock_ledger import StockLedgerService - make_sl_entries(sl_entries, allow_negative_stock, via_landed_cost_voucher) - update_batch_qty( - self.doctype, self.name, self.docstatus, via_landed_cost_voucher=via_landed_cost_voucher + return StockLedgerService(self).make_sl_entries( + sl_entries, allow_negative_stock, via_landed_cost_voucher ) - SerialBatchBundleService(self).validate_reserved_batches() - def make_gl_entries_on_cancel(self, from_repost=False): if not from_repost: cancel_exchange_gain_loss_journal(frappe._dict(doctype=self.doctype, name=self.name)) @@ -1019,30 +847,9 @@ class StockController(AccountsController): return message def repost_future_sle_and_gle(self, force=False, via_landed_cost_voucher=False): - args = frappe._dict( - { - "posting_date": self.posting_date, - "posting_time": self.posting_time, - "voucher_type": self.doctype, - "voucher_no": self.name, - "company": self.company, - "via_landed_cost_voucher": via_landed_cost_voucher, - } - ) + from erpnext.stock.services.stock_ledger import StockLedgerService - if self.docstatus == 2: - force = True - - if force or future_sle_exists(args) or repost_required_for_queue(self): - item_based_reposting = frappe.get_single_value("Stock Reposting Settings", "item_based_reposting") - if item_based_reposting: - create_item_wise_repost_entries( - voucher_type=self.doctype, - voucher_no=self.name, - via_landed_cost_voucher=via_landed_cost_voucher, - ) - else: - create_repost_item_valuation_entry(args) + return StockLedgerService(self).repost_future_sle_and_gle(force, via_landed_cost_voucher) def add_gl_entry( self, diff --git a/erpnext/stock/services/stock_ledger.py b/erpnext/stock/services/stock_ledger.py new file mode 100644 index 00000000000..f41ae7e53ed --- /dev/null +++ b/erpnext/stock/services/stock_ledger.py @@ -0,0 +1,250 @@ +# Copyright (c) 2015, Frappe Technologies Pvt. Ltd. and Contributors +# License: GNU General Public License v3. See license.txt + +"""Stock Ledger Entry building and reposting for stock transactions. + +Extracted from ``StockController``. Builds the SLE dicts for a voucher, writes +them, and triggers future SLE/GL reposting. The repost helper *functions* remain +module-level in ``stock_controller`` (imported widely); this service owns the +instance-level logic. +""" + +import frappe +from frappe.utils import flt + +from erpnext.accounts.utils import get_fiscal_year +from erpnext.stock.doctype.inventory_dimension.inventory_dimension import ( + get_evaluated_inventory_dimension, +) + + +class StockLedgerService: + def __init__(self, doc) -> None: + self.doc = doc + + def get_items_and_warehouses(self) -> tuple[list[str], list[str]]: + """Get list of items and warehouses affected by a transaction""" + + if not (hasattr(self.doc, "items") or hasattr(self.doc, "packed_items")): + return [], [] + + item_rows = (self.doc.get("items") or []) + (self.doc.get("packed_items") or []) + + items = {d.item_code for d in item_rows if d.item_code} + + warehouses = set() + for d in item_rows: + if d.get("warehouse"): + warehouses.add(d.warehouse) + + if self.doc.doctype == "Stock Entry": + if d.get("s_warehouse"): + warehouses.add(d.s_warehouse) + if d.get("t_warehouse"): + warehouses.add(d.t_warehouse) + + return list(items), list(warehouses) + + def get_stock_ledger_details(self): + stock_ledger = {} + + table = frappe.qb.DocType("Stock Ledger Entry") + + stock_ledger_entries = ( + frappe.qb.from_(table) + .select( + table.name, + table.warehouse, + table.stock_value_difference, + table.valuation_rate, + table.voucher_detail_no, + table.item_code, + table.posting_date, + table.posting_time, + table.actual_qty, + table.qty_after_transaction, + table.project, + ) + .where( + (table.voucher_type == self.doc.doctype) + & (table.voucher_no == self.doc.name) + & (table.is_cancelled == 0) + ) + ).run(as_dict=True) + + for sle in stock_ledger_entries: + stock_ledger.setdefault(sle.voucher_detail_no, []).append(sle) + + return stock_ledger + + def get_sl_entries(self, d, args): + sl_dict = frappe._dict( + { + "item_code": d.get("item_code", None), + "warehouse": d.get("warehouse", None), + "serial_and_batch_bundle": d.get("serial_and_batch_bundle"), + "posting_date": self.doc.posting_date, + "posting_time": self.doc.posting_time, + "fiscal_year": get_fiscal_year(self.doc.posting_date, company=self.doc.company)[0], + "voucher_type": self.doc.doctype, + "voucher_no": self.doc.name, + "voucher_detail_no": d.name, + "actual_qty": (self.doc.docstatus == 1 and 1 or -1) * flt(d.get("stock_qty")), + "stock_uom": frappe.get_cached_value( + "Item", args.get("item_code") or d.get("item_code"), "stock_uom" + ), + "incoming_rate": 0, + "company": self.doc.company, + "project": d.get("project") or self.doc.get("project"), + "is_cancelled": 1 if self.doc.docstatus == 2 else 0, + } + ) + + sl_dict.update(args) + self.update_inventory_dimensions(d, sl_dict) + + if self.doc.docstatus == 2: + from erpnext.deprecation_dumpster import deprecation_warning + + deprecation_warning("unknown", "v16", "No instructions.") + # To handle denormalized serial no records, will br deprecated in v16 + for field in ["serial_no", "batch_no"]: + if d.get(field): + sl_dict[field] = d.get(field) + + return sl_dict + + def update_inventory_dimensions(self, row, sl_dict) -> None: + # To handle delivery note and sales invoice + if row.get("item_row"): + row = row.get("item_row") + + dimensions = get_evaluated_inventory_dimension(row, sl_dict, parent_doc=self.doc) + for dimension in dimensions: + if not dimension: + continue + + if ( + self.doc.doctype in ["Purchase Invoice", "Purchase Receipt"] + and row.get("rejected_warehouse") + and sl_dict.get("warehouse") == row.get("rejected_warehouse") + ): + fieldname = f"rejected_{dimension.source_fieldname}" + sl_dict[dimension.target_fieldname] = row.get(fieldname) + continue + + if self.doc.doctype in [ + "Purchase Invoice", + "Purchase Receipt", + "Sales Invoice", + "Delivery Note", + "Stock Entry", + ]: + if ( + ( + sl_dict.actual_qty > 0 + and not self.doc.get("is_return") + or sl_dict.actual_qty < 0 + and self.doc.get("is_return") + ) + and self.doc.doctype in ["Purchase Invoice", "Purchase Receipt", "Stock Entry"] + ) or ( + ( + sl_dict.actual_qty < 0 + and not self.doc.get("is_return") + or sl_dict.actual_qty > 0 + and self.doc.get("is_return") + ) + and self.doc.doctype in ["Sales Invoice", "Delivery Note", "Stock Entry"] + ): + if self.doc.doctype == "Stock Entry": + if row.get("t_warehouse") == sl_dict.warehouse and sl_dict.get("actual_qty") > 0: + fieldname = f"to_{dimension.source_fieldname}" + if dimension.source_fieldname.startswith("to_"): + fieldname = f"{dimension.source_fieldname}" + + sl_dict[dimension.target_fieldname] = row.get(fieldname) + continue + + sl_dict[dimension.target_fieldname] = row.get(dimension.source_fieldname) + else: + fieldname_start_with = "to" + if self.doc.doctype in ["Purchase Invoice", "Purchase Receipt"]: + fieldname_start_with = "from" + + fieldname = f"{fieldname_start_with}_{dimension.source_fieldname}" + sl_dict[dimension.target_fieldname] = row.get(fieldname) + + if not sl_dict.get(dimension.target_fieldname): + sl_dict[dimension.target_fieldname] = row.get(dimension.source_fieldname) + + elif row.get(dimension.source_fieldname): + sl_dict[dimension.target_fieldname] = row.get(dimension.source_fieldname) + + if not sl_dict.get(dimension.target_fieldname) and dimension.fetch_from_parent: + sl_dict[dimension.target_fieldname] = self.doc.get(dimension.fetch_from_parent) + + # Get value based on doctype name + if not sl_dict.get(dimension.target_fieldname): + fieldname = next( + ( + field.fieldname + for field in frappe.get_meta(self.doc.doctype).fields + if field.options == dimension.fetch_from_parent + ), + None, + ) + + if fieldname and self.doc.get(fieldname): + sl_dict[dimension.target_fieldname] = self.doc.get(fieldname) + + if sl_dict[dimension.target_fieldname] and self.doc.docstatus == 1: + row.db_set(dimension.source_fieldname, sl_dict[dimension.target_fieldname]) + + def make_sl_entries(self, sl_entries, allow_negative_stock=False, via_landed_cost_voucher=False): + from erpnext.stock.serial_batch_bundle import update_batch_qty + from erpnext.stock.services.serial_batch_bundle import SerialBatchBundleService + from erpnext.stock.stock_ledger import make_sl_entries + + make_sl_entries(sl_entries, allow_negative_stock, via_landed_cost_voucher) + update_batch_qty( + self.doc.doctype, + self.doc.name, + self.doc.docstatus, + via_landed_cost_voucher=via_landed_cost_voucher, + ) + + SerialBatchBundleService(self.doc).validate_reserved_batches() + + def repost_future_sle_and_gle(self, force=False, via_landed_cost_voucher=False): + from erpnext.controllers.stock_controller import ( + create_item_wise_repost_entries, + create_repost_item_valuation_entry, + future_sle_exists, + repost_required_for_queue, + ) + + args = frappe._dict( + { + "posting_date": self.doc.posting_date, + "posting_time": self.doc.posting_time, + "voucher_type": self.doc.doctype, + "voucher_no": self.doc.name, + "company": self.doc.company, + "via_landed_cost_voucher": via_landed_cost_voucher, + } + ) + + if self.doc.docstatus == 2: + force = True + + if force or future_sle_exists(args) or repost_required_for_queue(self.doc): + item_based_reposting = frappe.get_single_value("Stock Reposting Settings", "item_based_reposting") + if item_based_reposting: + create_item_wise_repost_entries( + voucher_type=self.doc.doctype, + voucher_no=self.doc.name, + via_landed_cost_voucher=via_landed_cost_voucher, + ) + else: + create_repost_item_valuation_entry(args)