refactor(minor): SRE functions

This commit is contained in:
s-aga-r 2023-03-27 12:18:40 +05:30
parent 22a9c8ad55
commit 00ac49f81b
3 changed files with 85 additions and 66 deletions

View File

@ -623,13 +623,13 @@ def make_project(source_name, target_doc=None):
@frappe.whitelist() @frappe.whitelist()
def make_delivery_note(source_name, target_doc=None, skip_item_mapping=False): def make_delivery_note(source_name, target_doc=None, skip_item_mapping=False):
from erpnext.stock.doctype.stock_reservation_entry.stock_reservation_entry import ( from erpnext.stock.doctype.stock_reservation_entry.stock_reservation_entry import (
get_stock_reservation_entry_for_voucher, get_stock_reservation_entries_for_voucher,
has_reserved_stock, has_reserved_stock,
) )
def set_missing_values(source, target): def set_missing_values(source, target):
if not target.items and has_reserved_stock("Sales Order", source_name): if not target.items and has_reserved_stock("Sales Order", source_name):
sre_list = get_stock_reservation_entry_for_voucher("Sales Order", source_name) sre_list = get_stock_reservation_entries_for_voucher("Sales Order", source_name)
sre_dict = {d.pop("voucher_detail_no"): d for d in sre_list} sre_dict = {d.pop("voucher_detail_no"): d for d in sre_list}
for item in source.get("items"): for item in source.get("items"):

View File

@ -277,20 +277,20 @@ class DeliveryNote(SellingController):
def update_stock_reservation_entry(self): def update_stock_reservation_entry(self):
if not self.is_return: if not self.is_return:
from erpnext.stock.doctype.stock_reservation_entry.stock_reservation_entry import ( from erpnext.stock.doctype.stock_reservation_entry.stock_reservation_entry import (
update_delivered_qty, update_sre_delivered_qty,
) )
for item in self.get("items"): for item in self.get("items"):
if item.against_sre: if item.against_sre:
update_delivered_qty(item.doctype, item.against_sre) update_sre_delivered_qty(item.doctype, item.against_sre)
def validate_against_sre(self): def validate_against_sre(self):
from erpnext.stock.doctype.stock_reservation_entry.stock_reservation_entry import ( from erpnext.stock.doctype.stock_reservation_entry.stock_reservation_entry import (
get_stock_reservation_entry_for_items, get_stock_reservation_entries_for_items,
has_reserved_stock, has_reserved_stock,
) )
sre_details = get_stock_reservation_entry_for_items(self.items) sre_details = get_stock_reservation_entries_for_items(self.items)
for item in self.items: for item in self.items:
if item.against_sre: if item.against_sre:

View File

@ -9,7 +9,7 @@ from erpnext.utilities.transaction_base import TransactionBase
class StockReservationEntry(TransactionBase): class StockReservationEntry(TransactionBase):
def validate(self): def validate(self) -> None:
from erpnext.stock.utils import validate_disabled_warehouse, validate_warehouse_company from erpnext.stock.utils import validate_disabled_warehouse, validate_warehouse_company
self.validate_posting_time() self.validate_posting_time()
@ -17,15 +17,15 @@ class StockReservationEntry(TransactionBase):
validate_disabled_warehouse(self.warehouse) validate_disabled_warehouse(self.warehouse)
validate_warehouse_company(self.warehouse, self.company) validate_warehouse_company(self.warehouse, self.company)
def on_submit(self): def on_submit(self) -> None:
self.update_reserved_qty_in_voucher() self.update_reserved_qty_in_voucher()
self.update_status() self.update_status()
def on_cancel(self): def on_cancel(self) -> None:
self.update_reserved_qty_in_voucher() self.update_reserved_qty_in_voucher()
self.update_status() self.update_status()
def validate_mandatory(self): def validate_mandatory(self) -> None:
mandatory = [ mandatory = [
"item_code", "item_code",
"warehouse", "warehouse",
@ -44,7 +44,7 @@ class StockReservationEntry(TransactionBase):
if not self.get(d): if not self.get(d):
frappe.throw(_("{0} is required").format(self.meta.get_label(d))) frappe.throw(_("{0} is required").format(self.meta.get_label(d)))
def update_status(self, status=None, update_modified=True): def update_status(self, status: str = None, update_modified: bool = True) -> None:
if not status: if not status:
if self.docstatus == 2: if self.docstatus == 2:
status = "Cancelled" status = "Cancelled"
@ -62,41 +62,50 @@ class StockReservationEntry(TransactionBase):
frappe.db.set_value(self.doctype, self.name, "status", status, update_modified=update_modified) frappe.db.set_value(self.doctype, self.name, "status", status, update_modified=update_modified)
def update_reserved_qty_in_voucher(self, update_modified=True): def update_reserved_qty_in_voucher(
sre = frappe.qb.DocType("Stock Reservation Entry") self, reserved_qty_field: str = "stock_reserved_qty", update_modified: bool = True
reserved_qty = ( ) -> None:
frappe.qb.from_(sre) item_doctype = "Sales Order Item" if self.voucher_type == "Sales Order" else None
.select(Sum(sre.reserved_qty))
.where( if item_doctype:
(sre.docstatus == 1) sre = frappe.qb.DocType("Stock Reservation Entry")
& (sre.voucher_type == self.voucher_type) reserved_qty = (
& (sre.voucher_no == self.voucher_no) frappe.qb.from_(sre)
& (sre.voucher_detail_no == self.voucher_detail_no) .select(Sum(sre.reserved_qty))
.where(
(sre.docstatus == 1)
& (sre.voucher_type == self.voucher_type)
& (sre.voucher_no == self.voucher_no)
& (sre.voucher_detail_no == self.voucher_detail_no)
)
).run(as_list=True)[0][0] or 0
frappe.db.set_value(
item_doctype,
self.voucher_detail_no,
reserved_qty_field,
reserved_qty,
update_modified=update_modified,
) )
).run(as_list=True)[0][0] or 0
frappe.db.set_value(
"Sales Order Item",
self.voucher_detail_no,
"stock_reserved_qty",
reserved_qty,
update_modified=update_modified,
)
def get_stock_reservation_entry_for_voucher(voucher_type, voucher_no, voucher_detail_no=None): def get_stock_reservation_entries_for_voucher(
voucher_type: str, voucher_no: str, voucher_detail_no: str = None, fields: list[str] = None
) -> list[dict]:
if not fields or not isinstance(fields, list):
fields = [
"name",
"item_code",
"warehouse",
"voucher_detail_no",
"reserved_qty",
"delivered_qty",
"stock_uom",
]
sre = frappe.qb.DocType("Stock Reservation Entry") sre = frappe.qb.DocType("Stock Reservation Entry")
query = ( query = (
frappe.qb.from_(sre) frappe.qb.from_(sre)
.select(
sre.name,
sre.item_code,
sre.warehouse,
sre.voucher_detail_no,
sre.reserved_qty,
sre.delivered_qty,
sre.stock_uom,
)
.where( .where(
(sre.docstatus == 1) (sre.docstatus == 1)
& (sre.voucher_type == voucher_type) & (sre.voucher_type == voucher_type)
@ -106,20 +115,27 @@ def get_stock_reservation_entry_for_voucher(voucher_type, voucher_no, voucher_de
.orderby(sre.creation) .orderby(sre.creation)
) )
for field in fields:
query = query.select(sre[field])
if voucher_detail_no: if voucher_detail_no:
query = query.where(sre.voucher_detail_no == voucher_detail_no) query = query.where(sre.voucher_detail_no == voucher_detail_no)
return query.run(as_dict=True) return query.run(as_dict=True)
def has_reserved_stock(voucher_type, voucher_no, voucher_detail_no=None): def has_reserved_stock(voucher_type: str, voucher_no: str, voucher_detail_no: str = None) -> bool:
if get_stock_reservation_entry_for_voucher(voucher_type, voucher_no, voucher_detail_no): if get_stock_reservation_entries_for_voucher(
voucher_type, voucher_no, voucher_detail_no, fields=["name"]
):
return True return True
return False return False
def update_delivered_qty(doctype, sre_name, sre_field="against_sre", qty_field="stock_qty"): def update_sre_delivered_qty(
doctype: str, sre_name: str, sre_field: str = "against_sre", qty_field: str = "stock_qty"
) -> None:
table = frappe.qb.DocType(doctype) table = frappe.qb.DocType(doctype)
delivered_qty = ( delivered_qty = (
frappe.qb.from_(table) frappe.qb.from_(table)
@ -133,32 +149,35 @@ def update_delivered_qty(doctype, sre_name, sre_field="against_sre", qty_field="
sre_doc.update_status() sre_doc.update_status()
def get_stock_reservation_entry_for_items(items, sre_field="against_sre"): def get_stock_reservation_entries_for_items(
items: list[dict | object], sre_field: str = "against_sre"
) -> dict[dict]:
sre_details = {} sre_details = {}
sre_list = [item.get(sre_field) for item in items if item.get(sre_field)] if items:
sre_list = [item.get(sre_field) for item in items if item.get(sre_field)]
if sre_list: if sre_list:
sre = frappe.qb.DocType("Stock Reservation Entry") sre = frappe.qb.DocType("Stock Reservation Entry")
sre_data = ( sre_data = (
frappe.qb.from_(sre) frappe.qb.from_(sre)
.select( .select(
sre.name, sre.name,
sre.status, sre.status,
sre.docstatus, sre.docstatus,
sre.item_code, sre.item_code,
sre.warehouse, sre.warehouse,
sre.voucher_type, sre.voucher_type,
sre.voucher_no, sre.voucher_no,
sre.voucher_detail_no, sre.voucher_detail_no,
sre.reserved_qty, sre.reserved_qty,
sre.delivered_qty, sre.delivered_qty,
sre.stock_uom, sre.stock_uom,
) )
.where(sre.name.isin(sre_list)) .where(sre.name.isin(sre_list))
.orderby(sre.creation) .orderby(sre.creation)
).run(as_dict=True) ).run(as_dict=True)
sre_details = {d.name: d for d in sre_data} sre_details = {d.name: d for d in sre_data}
return sre_details return sre_details