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.
This commit is contained in:
Nabin Hait
2026-06-04 15:35:04 +05:30
parent a26d8d448c
commit 4affdd51f6
2 changed files with 264 additions and 207 deletions

View File

@@ -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,

View File

@@ -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)