refactor(stock): adding type hint for stock module

This commit is contained in:
Nishka Gosalia
2026-02-20 16:37:50 +05:30
parent 9541cb226d
commit 13995a64b8
34 changed files with 294 additions and 169 deletions

View File

@@ -9,7 +9,12 @@ from erpnext.stock.doctype.stock_reservation_entry.stock_reservation_entry impor
@frappe.whitelist() @frappe.whitelist()
def get_data( def get_data(
item_code=None, warehouse=None, item_group=None, start=0, sort_by="actual_qty", sort_order="desc" item_code: str | None = None,
warehouse: str | None = None,
item_group: str | None = None,
start: int = 0,
sort_by: str = "actual_qty",
sort_order: str = "desc",
): ):
"""Return data to render the item dashboard""" """Return data to render the item dashboard"""
filters = [] filters = []

View File

@@ -7,13 +7,13 @@ from erpnext.stock.utils import get_stock_balance
@frappe.whitelist() @frappe.whitelist()
def get_data( def get_data(
item_code=None, item_code: str | None = None,
warehouse=None, warehouse: str | None = None,
parent_warehouse=None, parent_warehouse: str | None = None,
company=None, company: str | None = None,
start=0, start: int = 0,
sort_by="stock_capacity", sort_by: str = "stock_capacity",
sort_order="desc", sort_order: str = "desc",
): ):
"""Return data to render the warehouse capacity dashboard.""" """Return data to render the warehouse capacity dashboard."""
filters = get_filters(item_code, warehouse, parent_warehouse, company) filters = get_filters(item_code, warehouse, parent_warehouse, company)

View File

@@ -2,6 +2,8 @@
# License: GNU General Public License v3. See license.txt # License: GNU General Public License v3. See license.txt
from typing import Any
import frappe import frappe
from frappe import _ from frappe import _
from frappe.query_builder.functions import Sum from frappe.query_builder.functions import Sum
@@ -11,15 +13,15 @@ from frappe.utils.dashboard import cache_source
@frappe.whitelist() @frappe.whitelist()
@cache_source @cache_source
def get( def get(
chart_name=None, chart_name: str | None = None,
chart=None, chart: Any = None,
no_cache=None, no_cache: Any = None,
filters=None, filters: dict | str | None = None,
from_date=None, from_date: Any = None,
to_date=None, to_date: Any = None,
timespan=None, timespan: Any = None,
time_interval=None, time_interval: Any = None,
heatmap_year=None, heatmap_year: Any = None,
): ):
if filters and isinstance(filters, str): if filters and isinstance(filters, str):
filters = frappe.parse_json(filters) filters = frappe.parse_json(filters)

View File

@@ -2,6 +2,8 @@
# License: GNU General Public License v3. See license.txt # License: GNU General Public License v3. See license.txt
from typing import Any
import frappe import frappe
from frappe import _ from frappe import _
from frappe.utils.dashboard import cache_source from frappe.utils.dashboard import cache_source
@@ -10,15 +12,15 @@ from frappe.utils.dashboard import cache_source
@frappe.whitelist() @frappe.whitelist()
@cache_source @cache_source
def get( def get(
chart_name=None, chart_name: str | None = None,
chart=None, chart: Any | None = None,
no_cache=None, no_cache: Any | None = None,
filters=None, filters: dict | str | None = None,
from_date=None, from_date: Any | None = None,
to_date=None, to_date: Any | None = None,
timespan=None, timespan: Any | None = None,
time_interval=None, time_interval: Any | None = None,
heatmap_year=None, heatmap_year: Any | None = None,
): ):
labels, datapoints = [], [] labels, datapoints = [], []
filters = frappe.parse_json(filters) filters = frappe.parse_json(filters)

View File

@@ -2,6 +2,7 @@
# License: GNU General Public License v3. See license.txt # License: GNU General Public License v3. See license.txt
import datetime
from collections import OrderedDict, defaultdict from collections import OrderedDict, defaultdict
import frappe import frappe
@@ -10,7 +11,7 @@ from frappe.model.document import Document
from frappe.model.naming import make_autoname, revert_series_if_last from frappe.model.naming import make_autoname, revert_series_if_last
from frappe.query_builder.functions import CurDate, Sum from frappe.query_builder.functions import CurDate, Sum
from frappe.utils import cint, flt, get_link_to_form from frappe.utils import cint, flt, get_link_to_form
from frappe.utils.data import add_days from frappe.utils.data import DateTimeLikeObject, add_days
from frappe.utils.jinja import render_template from frappe.utils.jinja import render_template
@@ -235,18 +236,18 @@ class Batch(Document):
@frappe.whitelist() @frappe.whitelist()
def get_batch_qty( def get_batch_qty(
batch_no=None, batch_no: str | None = None,
warehouse=None, warehouse: str | None = None,
item_code=None, item_code: str | None = None,
creation=None, creation: DateTimeLikeObject | None = None,
posting_datetime=None, posting_datetime: DateTimeLikeObject | None = None,
posting_date=None, posting_date: DateTimeLikeObject | None = None,
posting_time=None, posting_time: datetime.timedelta | None = None,
ignore_voucher_nos=None, ignore_voucher_nos: list | None = None,
for_stock_levels=False, for_stock_levels: bool = False,
consider_negative_batches=False, consider_negative_batches: bool = False,
do_not_check_future_batches=False, do_not_check_future_batches: bool = False,
ignore_reserved_stock=False, ignore_reserved_stock: bool = False,
): ):
"""Returns batch actual qty if warehouse is passed, """Returns batch actual qty if warehouse is passed,
or returns dict of qty by warehouse if warehouse is None or returns dict of qty by warehouse if warehouse is None
@@ -295,7 +296,7 @@ def get_batch_qty(
@frappe.whitelist() @frappe.whitelist()
def get_batches_by_oldest(item_code, warehouse): def get_batches_by_oldest(item_code: str, warehouse: str):
"""Returns the oldest batch and qty for the given item_code and warehouse""" """Returns the oldest batch and qty for the given item_code and warehouse"""
batches = get_batch_qty(item_code=item_code, warehouse=warehouse) batches = get_batch_qty(item_code=item_code, warehouse=warehouse)
batches_dates = [[batch, frappe.get_value("Batch", batch.batch_no, "expiry_date")] for batch in batches] batches_dates = [[batch, frappe.get_value("Batch", batch.batch_no, "expiry_date")] for batch in batches]
@@ -447,7 +448,7 @@ def make_batch(kwargs):
@frappe.whitelist() @frappe.whitelist()
def get_pos_reserved_batch_qty(filters): def get_pos_reserved_batch_qty(filters: dict | str):
import json import json
if isinstance(filters, str): if isinstance(filters, str):

View File

@@ -8,6 +8,7 @@ import frappe
from frappe import _ from frappe import _
from frappe.contacts.doctype.address.address import get_company_address from frappe.contacts.doctype.address.address import get_company_address
from frappe.desk.notifications import clear_doctype_notifications from frappe.desk.notifications import clear_doctype_notifications
from frappe.model.document import Document
from frappe.model.mapper import get_mapped_doc from frappe.model.mapper import get_mapped_doc
from frappe.model.utils import get_fetch_values from frappe.model.utils import get_fetch_values
from frappe.query_builder import DocType from frappe.query_builder import DocType
@@ -846,7 +847,9 @@ def get_returned_qty_map(delivery_note):
@frappe.whitelist() @frappe.whitelist()
def make_sales_invoice(source_name, target_doc=None, args=None): def make_sales_invoice(
source_name: str, target_doc: str | Document | None = None, args: dict | str | None = None
):
if args is None: if args is None:
args = {} args = {}
if isinstance(args, str): if isinstance(args, str):
@@ -975,7 +978,9 @@ def make_sales_invoice(source_name, target_doc=None, args=None):
@frappe.whitelist() @frappe.whitelist()
def make_delivery_trip(source_name, target_doc=None, kwargs=None): def make_delivery_trip(
source_name: str, target_doc: str | Document | None = None, kwargs: dict | None = None
):
if not target_doc: if not target_doc:
target_doc = frappe.new_doc("Delivery Trip") target_doc = frappe.new_doc("Delivery Trip")
doclist = get_mapped_doc( doclist = get_mapped_doc(
@@ -1001,7 +1006,9 @@ def make_delivery_trip(source_name, target_doc=None, kwargs=None):
@frappe.whitelist() @frappe.whitelist()
def make_installation_note(source_name, target_doc=None, kwargs=None): def make_installation_note(
source_name: str, target_doc: str | Document | None = None, kwargs: dict | None = None
):
def update_item(obj, target, source_parent): def update_item(obj, target, source_parent):
target.qty = flt(obj.qty) - flt(obj.installed_qty) target.qty = flt(obj.qty) - flt(obj.installed_qty)
target.serial_no = obj.serial_no target.serial_no = obj.serial_no
@@ -1029,7 +1036,7 @@ def make_installation_note(source_name, target_doc=None, kwargs=None):
@frappe.whitelist() @frappe.whitelist()
def make_packing_slip(source_name, target_doc=None): def make_packing_slip(source_name: str, target_doc: str | Document | None = None):
def set_missing_values(source, target): def set_missing_values(source, target):
target.run_method("set_missing_values") target.run_method("set_missing_values")
@@ -1084,7 +1091,7 @@ def make_packing_slip(source_name, target_doc=None):
@frappe.whitelist() @frappe.whitelist()
def make_shipment(source_name, target_doc=None): def make_shipment(source_name: str, target_doc: str | Document | None = None):
def postprocess(source, target): def postprocess(source, target):
user = frappe.db.get_value( user = frappe.db.get_value(
"User", frappe.session.user, ["email", "full_name", "phone", "mobile_no"], as_dict=1 "User", frappe.session.user, ["email", "full_name", "phone", "mobile_no"], as_dict=1
@@ -1159,20 +1166,20 @@ def make_shipment(source_name, target_doc=None):
@frappe.whitelist() @frappe.whitelist()
def make_sales_return(source_name, target_doc=None): def make_sales_return(source_name: str, target_doc: str | Document | None = None):
from erpnext.controllers.sales_and_purchase_return import make_return_doc from erpnext.controllers.sales_and_purchase_return import make_return_doc
return make_return_doc("Delivery Note", source_name, target_doc) return make_return_doc("Delivery Note", source_name, target_doc)
@frappe.whitelist() @frappe.whitelist()
def update_delivery_note_status(docname, status): def update_delivery_note_status(docname: str, status: str):
dn = frappe.get_lazy_doc("Delivery Note", docname) dn = frappe.get_lazy_doc("Delivery Note", docname)
dn.update_status(status) dn.update_status(status)
@frappe.whitelist() @frappe.whitelist()
def make_inter_company_purchase_receipt(source_name, target_doc=None): def make_inter_company_purchase_receipt(source_name: str, target_doc: str | Document | None = None):
return make_inter_company_transaction("Delivery Note", source_name, target_doc) return make_inter_company_transaction("Delivery Note", source_name, target_doc)

View File

@@ -150,7 +150,7 @@ class DeliveryTrip(Document):
frappe.msgprint(_("Delivery Notes {0} updated").format(", ".join(delivery_notes_updated))) frappe.msgprint(_("Delivery Notes {0} updated").format(", ".join(delivery_notes_updated)))
@frappe.whitelist() @frappe.whitelist()
def process_route(self, optimize): def process_route(self, optimize: bool):
""" """
Estimate the arrival times for each stop in the Delivery Trip. Estimate the arrival times for each stop in the Delivery Trip.
If `optimize` is True, the stops will be re-arranged, based If `optimize` is True, the stops will be re-arranged, based
@@ -305,7 +305,7 @@ class DeliveryTrip(Document):
@frappe.whitelist() @frappe.whitelist()
def get_contact_and_address(name): def get_contact_and_address(name: str):
out = frappe._dict() out = frappe._dict()
get_default_contact(out, name) get_default_contact(out, name)
@@ -367,7 +367,7 @@ def get_default_address(out, name):
@frappe.whitelist() @frappe.whitelist()
def get_contact_display(contact): def get_contact_display(contact: str):
contact_info = frappe.db.get_value( contact_info = frappe.db.get_value(
"Contact", contact, ["first_name", "last_name", "phone", "mobile_no"], as_dict=1 "Contact", contact, ["first_name", "last_name", "phone", "mobile_no"], as_dict=1
) )
@@ -403,7 +403,7 @@ def sanitize_address(address):
@frappe.whitelist() @frappe.whitelist()
def notify_customers(delivery_trip): def notify_customers(delivery_trip: str):
delivery_trip = frappe.get_doc("Delivery Trip", delivery_trip) delivery_trip = frappe.get_doc("Delivery Trip", delivery_trip)
context = delivery_trip.as_dict() context = delivery_trip.as_dict()
@@ -468,7 +468,7 @@ def get_attachments(delivery_stop):
@frappe.whitelist() @frappe.whitelist()
def get_driver_email(driver): def get_driver_email(driver: str):
employee = frappe.db.get_value("Driver", driver, "employee") employee = frappe.db.get_value("Driver", driver, "employee")
email = frappe.db.get_value("Employee", employee, "prefered_email") email = frappe.db.get_value("Employee", employee, "prefered_email")
return {"email": email} return {"email": email}

View File

@@ -1,6 +1,8 @@
# Copyright (c) 2022, Frappe Technologies Pvt. Ltd. and contributors # Copyright (c) 2022, Frappe Technologies Pvt. Ltd. and contributors
# For license information, please see license.txt # For license information, please see license.txt
from typing import Any
import frappe import frappe
from frappe import _, bold, scrub from frappe import _, bold, scrub
from frappe.custom.doctype.custom_field.custom_field import create_custom_fields from frappe.custom.doctype.custom_field.custom_field import create_custom_fields
@@ -306,7 +308,12 @@ def field_exists(doctype, fieldname) -> str or None:
@frappe.whitelist() @frappe.whitelist()
def get_inventory_documents( def get_inventory_documents(
doctype=None, txt=None, searchfield=None, start=None, page_len=None, filters=None doctype: Any | None = None,
txt: str | None = None,
searchfield: str | None = None,
start: int | None = None,
page_len: int | None = None,
filters: dict | None = None,
): ):
and_filters = [["DocField", "parent", "not in", ["Batch", "Serial No", "Item Price"]]] and_filters = [["DocField", "parent", "not in", ["Batch", "Serial No", "Item Price"]]]
or_filters = [ or_filters = [
@@ -395,13 +402,13 @@ def get_inventory_dimensions():
@frappe.whitelist() @frappe.whitelist()
def delete_dimension(dimension): def delete_dimension(dimension: str):
doc = frappe.get_doc("Inventory Dimension", dimension) doc = frappe.get_doc("Inventory Dimension", dimension)
doc.delete() doc.delete()
@frappe.whitelist() @frappe.whitelist()
def get_parent_fields(child_doctype, dimension_name): def get_parent_fields(child_doctype: str, dimension_name: str):
parent_doctypes = frappe.get_all("DocField", fields=["parent"], filters={"options": child_doctype}) parent_doctypes = frappe.get_all("DocField", fields=["parent"], filters={"options": child_doctype})
fields = [] fields = []

View File

@@ -1341,7 +1341,7 @@ def set_item_default(item_code, company, fieldname, value):
@frappe.whitelist() @frappe.whitelist()
def get_item_details(item_code, company=None): def get_item_details(item_code: str, company: str | None = None):
out = frappe._dict() out = frappe._dict()
if company: if company:
out = get_item_defaults(item_code, company) or frappe._dict() out = get_item_defaults(item_code, company) or frappe._dict()
@@ -1353,7 +1353,7 @@ def get_item_details(item_code, company=None):
@frappe.whitelist() @frappe.whitelist()
def get_uom_conv_factor(uom, stock_uom): def get_uom_conv_factor(uom: str | None, stock_uom: str | None):
"""Get UOM conversion factor from uom to stock_uom """Get UOM conversion factor from uom to stock_uom
e.g. uom = "Kg", stock_uom = "Gram" then returns 1000.0 e.g. uom = "Kg", stock_uom = "Gram" then returns 1000.0
""" """
@@ -1399,7 +1399,7 @@ def get_uom_conv_factor(uom, stock_uom):
@frappe.whitelist() @frappe.whitelist()
def get_item_attribute(parent, attribute_value=""): def get_item_attribute(parent: str, attribute_value: str = ""):
"""Used for providing auto-completions in child table.""" """Used for providing auto-completions in child table."""
if not frappe.has_permission("Item"): if not frappe.has_permission("Item"):
frappe.throw(_("No Permission")) frappe.throw(_("No Permission"))

View File

@@ -2,6 +2,8 @@
# For license information, please see license.txt # For license information, please see license.txt
from typing import Any
import frappe import frappe
from frappe import _ from frappe import _
from frappe.model.document import Document from frappe.model.document import Document
@@ -83,7 +85,7 @@ class ItemAlternative(Document):
@frappe.whitelist() @frappe.whitelist()
@frappe.validate_and_sanitize_search_inputs @frappe.validate_and_sanitize_search_inputs
def get_alternative_items(doctype, txt, searchfield, start, page_len, filters): def get_alternative_items(doctype: Any, txt: str, searchfield: Any, start: int, page_len: int, filters: dict):
return frappe.db.sql( return frappe.db.sql(
f""" (select alternative_item_code from `tabItem Alternative` f""" (select alternative_item_code from `tabItem Alternative`
where item_code = %(item_code)s and alternative_item_code like %(txt)s) where item_code = %(item_code)s and alternative_item_code like %(txt)s)

View File

@@ -86,7 +86,7 @@ class ItemManufacturer(Document):
@frappe.whitelist() @frappe.whitelist()
def get_item_manufacturer_part_no(item_code, manufacturer): def get_item_manufacturer_part_no(item_code: str, manufacturer: str):
return frappe.db.get_value( return frappe.db.get_value(
"Item Manufacturer", "Item Manufacturer",
{"item_code": item_code, "manufacturer": manufacturer}, {"item_code": item_code, "manufacturer": manufacturer},

View File

@@ -2,6 +2,8 @@
# For license information, please see license.txt # For license information, please see license.txt
from typing import Any
import frappe import frappe
from frappe import _ from frappe import _
from frappe.model.document import Document from frappe.model.document import Document
@@ -221,7 +223,7 @@ class LandedCostVoucher(Document):
) )
@frappe.whitelist() @frappe.whitelist()
def get_receipt_document_details(self, receipt_document_type, receipt_document): def get_receipt_document_details(self, receipt_document_type: str, receipt_document: str):
if receipt_document_type in [ if receipt_document_type in [
"Purchase Invoice", "Purchase Invoice",
"Purchase Receipt", "Purchase Receipt",
@@ -356,7 +358,7 @@ class LandedCostVoucher(Document):
) )
@frappe.whitelist() @frappe.whitelist()
def get_vendor_invoice_amount(self, vendor_invoice): def get_vendor_invoice_amount(self, vendor_invoice: str):
filters = frappe._dict( filters = frappe._dict(
{ {
"name": vendor_invoice, "name": vendor_invoice,
@@ -427,7 +429,9 @@ def get_pr_items(purchase_receipt):
@frappe.whitelist() @frappe.whitelist()
@frappe.validate_and_sanitize_search_inputs @frappe.validate_and_sanitize_search_inputs
def get_vendor_invoices(doctype, txt, searchfield, start, page_len, filters): def get_vendor_invoices(
doctype: str, txt: str | None, searchfield: Any, start: int, page_len: int, filters: dict
):
if not frappe.has_permission("Purchase Invoice", "read"): if not frappe.has_permission("Purchase Invoice", "read"):
return [] return []

View File

@@ -6,10 +6,12 @@
import json import json
from typing import Any
import frappe import frappe
import frappe.defaults import frappe.defaults
from frappe import _, msgprint from frappe import _, msgprint
from frappe.model.document import Document
from frappe.model.mapper import get_mapped_doc from frappe.model.mapper import get_mapped_doc
from frappe.query_builder import Order from frappe.query_builder import Order
from frappe.query_builder.functions import Sum from frappe.query_builder.functions import Sum
@@ -479,14 +481,16 @@ def get_list_context(context=None):
@frappe.whitelist() @frappe.whitelist()
def update_status(name, status): def update_status(name: str, status: str):
material_request = frappe.get_doc("Material Request", name) material_request = frappe.get_doc("Material Request", name)
material_request.check_permission("write") material_request.check_permission("write")
material_request.update_status(status) material_request.update_status(status)
@frappe.whitelist() @frappe.whitelist()
def make_purchase_order(source_name, target_doc=None, args=None): def make_purchase_order(
source_name: str, target_doc: str | Document | None = None, args: dict | str | None = None
):
if args is None: if args is None:
args = {} args = {}
if isinstance(args, str): if isinstance(args, str):
@@ -563,7 +567,7 @@ def make_purchase_order(source_name, target_doc=None, args=None):
@frappe.whitelist() @frappe.whitelist()
def make_request_for_quotation(source_name, target_doc=None): def make_request_for_quotation(source_name: str, target_doc: str | Document | None = None):
doclist = get_mapped_doc( doclist = get_mapped_doc(
"Material Request", "Material Request",
source_name, source_name,
@@ -588,7 +592,9 @@ def make_request_for_quotation(source_name, target_doc=None):
@frappe.whitelist() @frappe.whitelist()
def make_purchase_order_based_on_supplier(source_name, target_doc=None, args=None): def make_purchase_order_based_on_supplier(
source_name: str, target_doc: str | Document | None = None, args: dict | None = None
):
mr = source_name mr = source_name
supplier_items = get_items_based_on_default_supplier(args.get("supplier")) supplier_items = get_items_based_on_default_supplier(args.get("supplier"))
@@ -631,7 +637,7 @@ def make_purchase_order_based_on_supplier(source_name, target_doc=None, args=Non
@frappe.whitelist() @frappe.whitelist()
def get_items_based_on_default_supplier(supplier): def get_items_based_on_default_supplier(supplier: str):
supplier_items = [ supplier_items = [
d.parent d.parent
for d in frappe.db.get_all( for d in frappe.db.get_all(
@@ -644,7 +650,9 @@ def get_items_based_on_default_supplier(supplier):
@frappe.whitelist() @frappe.whitelist()
@frappe.validate_and_sanitize_search_inputs @frappe.validate_and_sanitize_search_inputs
def get_material_requests_based_on_supplier(doctype, txt, searchfield, start, page_len, filters): def get_material_requests_based_on_supplier(
doctype: Any, txt: str, searchfield: Any, start: int, page_len: int, filters: dict
):
supplier = filters.get("supplier") supplier = filters.get("supplier")
supplier_items = get_items_based_on_default_supplier(supplier) supplier_items = get_items_based_on_default_supplier(supplier)
@@ -688,7 +696,9 @@ def get_material_requests_based_on_supplier(doctype, txt, searchfield, start, pa
@frappe.whitelist() @frappe.whitelist()
@frappe.validate_and_sanitize_search_inputs @frappe.validate_and_sanitize_search_inputs
def get_default_supplier_query(doctype, txt, searchfield, start, page_len, filters): def get_default_supplier_query(
doctype: Any, txt: str, searchfield: str, start: int, page_len: int, filters: dict
):
doc = frappe.get_doc("Material Request", filters.get("doc")) doc = frappe.get_doc("Material Request", filters.get("doc"))
item_list = [] item_list = []
for d in doc.items: for d in doc.items:
@@ -718,7 +728,7 @@ def get_default_supplier_query(doctype, txt, searchfield, start, page_len, filte
@frappe.whitelist() @frappe.whitelist()
def make_supplier_quotation(source_name, target_doc=None): def make_supplier_quotation(source_name: str, target_doc: str | Document | None = None):
def postprocess(source, target_doc): def postprocess(source, target_doc):
set_missing_values(source, target_doc) set_missing_values(source, target_doc)
@@ -748,7 +758,7 @@ def make_supplier_quotation(source_name, target_doc=None):
@frappe.whitelist() @frappe.whitelist()
def make_stock_entry(source_name, target_doc=None): def make_stock_entry(source_name: str, target_doc: str | Document | None = None):
def update_item(obj, target, source_parent): def update_item(obj, target, source_parent):
qty = ( qty = (
flt(flt(obj.stock_qty) - flt(obj.ordered_qty)) / target.conversion_factor flt(flt(obj.stock_qty) - flt(obj.ordered_qty)) / target.conversion_factor
@@ -841,7 +851,7 @@ def make_stock_entry(source_name, target_doc=None):
@frappe.whitelist() @frappe.whitelist()
def raise_work_orders(material_request, company): def raise_work_orders(material_request: str, company: str):
mr = frappe.get_doc("Material Request", material_request) mr = frappe.get_doc("Material Request", material_request)
errors = [] errors = []
work_orders = [] work_orders = []
@@ -912,7 +922,7 @@ def raise_work_orders(material_request, company):
@frappe.whitelist() @frappe.whitelist()
def create_pick_list(source_name, target_doc=None): def create_pick_list(source_name: str, target_doc: str | Document | None = None):
def update_item(obj, target, source_parent): def update_item(obj, target, source_parent):
qty = flt((obj.stock_qty - obj.picked_qty) / target.conversion_factor, obj.precision("qty")) qty = flt((obj.stock_qty - obj.picked_qty) / target.conversion_factor, obj.precision("qty"))
target.qty = qty target.qty = qty
@@ -951,7 +961,7 @@ def create_pick_list(source_name, target_doc=None):
@frappe.whitelist() @frappe.whitelist()
def make_in_transit_stock_entry(source_name, in_transit_warehouse): def make_in_transit_stock_entry(source_name: str, in_transit_warehouse: str):
ste_doc = make_stock_entry(source_name) ste_doc = make_stock_entry(source_name)
ste_doc.add_to_transit = 1 ste_doc.add_to_transit = 1
ste_doc.to_warehouse = in_transit_warehouse ste_doc.to_warehouse = in_transit_warehouse

View File

@@ -366,7 +366,7 @@ def on_doctype_update():
@frappe.whitelist() @frappe.whitelist()
def get_items_from_product_bundle(row): def get_items_from_product_bundle(row: str):
row, items = ItemDetailsCtx(json.loads(row)), [] row, items = ItemDetailsCtx(json.loads(row)), []
bundled_items = get_product_bundle_items(row["item_code"]) bundled_items = get_product_bundle_items(row["item_code"])

View File

@@ -206,7 +206,7 @@ class PackingSlip(StatusUpdater):
@frappe.whitelist() @frappe.whitelist()
@frappe.validate_and_sanitize_search_inputs @frappe.validate_and_sanitize_search_inputs
def item_details(doctype, txt, searchfield, start, page_len, filters): def item_details(doctype: str, txt: str, searchfield: str, start: int, page_len: int, filters: dict):
from erpnext.controllers.queries import get_match_cond from erpnext.controllers.queries import get_match_cond
return frappe.db.sql( return frappe.db.sql(

View File

@@ -4,9 +4,11 @@
import json import json
from collections import OrderedDict, defaultdict from collections import OrderedDict, defaultdict
from itertools import groupby from itertools import groupby
from typing import Any
import frappe import frappe
from frappe import _, bold from frappe import _, bold
from frappe.model.document import Document
from frappe.model.mapper import map_child_doc from frappe.model.mapper import map_child_doc
from frappe.query_builder import Case from frappe.query_builder import Case
from frappe.query_builder.custom import GROUP_CONCAT from frappe.query_builder.custom import GROUP_CONCAT
@@ -428,7 +430,7 @@ class PickList(TransactionBase):
frappe.get_doc("Sales Order", sales_order, for_update=True).update_picking_status() frappe.get_doc("Sales Order", sales_order, for_update=True).update_picking_status()
@frappe.whitelist() @frappe.whitelist()
def create_stock_reservation_entries(self, notify=True) -> None: def create_stock_reservation_entries(self, notify: bool = True) -> None:
"""Creates Stock Reservation Entries for Sales Order Items against Pick List.""" """Creates Stock Reservation Entries for Sales Order Items against Pick List."""
so_items_details_map = {} so_items_details_map = {}
@@ -455,7 +457,7 @@ class PickList(TransactionBase):
) )
@frappe.whitelist() @frappe.whitelist()
def cancel_stock_reservation_entries(self, notify=True) -> None: def cancel_stock_reservation_entries(self, notify: bool = True) -> None:
"""Cancel Stock Reservation Entries for Sales Order Items created against Pick List.""" """Cancel Stock Reservation Entries for Sales Order Items created against Pick List."""
from erpnext.stock.doctype.stock_reservation_entry.stock_reservation_entry import ( from erpnext.stock.doctype.stock_reservation_entry.stock_reservation_entry import (
@@ -480,7 +482,7 @@ class PickList(TransactionBase):
) )
@frappe.whitelist() @frappe.whitelist()
def set_item_locations(self, save=False): def set_item_locations(self, save: bool = False):
self.validate_for_qty() self.validate_for_qty()
items = self.aggregate_item_qty() items = self.aggregate_item_qty()
picked_items_details = self.get_picked_items_details(items) picked_items_details = self.get_picked_items_details(items)
@@ -1209,7 +1211,7 @@ def get_available_item_locations_for_other_item(
@frappe.whitelist() @frappe.whitelist()
def create_delivery_note(source_name, target_doc=None): def create_delivery_note(source_name: str, target_doc: str | Document | None = None):
pick_list = frappe.get_doc("Pick List", source_name) pick_list = frappe.get_doc("Pick List", source_name)
validate_item_locations(pick_list) validate_item_locations(pick_list)
sales_dict = dict() sales_dict = dict()
@@ -1280,7 +1282,9 @@ def create_dn_wo_so(pick_list, delivery_note=None):
@frappe.whitelist() @frappe.whitelist()
def create_dn_for_pick_lists(source_name, target_doc=None, kwargs=None): def create_dn_for_pick_lists(
source_name: str, target_doc: str | Document | None = None, kwargs: dict | str | None = None
):
"""Get Items from Multiple Pick Lists and create a Delivery Note for filtered customer""" """Get Items from Multiple Pick Lists and create a Delivery Note for filtered customer"""
if kwargs is None: if kwargs is None:
kwargs = {} kwargs = {}
@@ -1428,7 +1432,7 @@ def add_product_bundles_to_delivery_note(
@frappe.whitelist() @frappe.whitelist()
def create_stock_entry(pick_list): def create_stock_entry(pick_list: str):
pick_list = frappe.get_doc(json.loads(pick_list)) pick_list = frappe.get_doc(json.loads(pick_list))
validate_item_locations(pick_list) validate_item_locations(pick_list)
@@ -1454,7 +1458,15 @@ def create_stock_entry(pick_list):
@frappe.whitelist() @frappe.whitelist()
def get_pending_work_orders(doctype, txt, searchfield, start, page_length, filters, as_dict): def get_pending_work_orders(
doctype: Any,
txt: str,
searchfield: str,
start: int,
page_length: int,
filters: dict,
as_dict: bool = False,
):
wo = frappe.qb.DocType("Work Order") wo = frappe.qb.DocType("Work Order")
return ( return (
frappe.qb.from_(wo) frappe.qb.from_(wo)
@@ -1474,7 +1486,9 @@ def get_pending_work_orders(doctype, txt, searchfield, start, page_length, filte
@frappe.whitelist() @frappe.whitelist()
def get_item_details(item_code, uom=None, warehouse=None, company=None): def get_item_details(
item_code: str, uom: str | None = None, warehouse: str | None = None, company: str | None = None
):
details = frappe.db.get_value("Item", item_code, "stock_uom", as_dict=1) details = frappe.db.get_value("Item", item_code, "stock_uom", as_dict=1)
details.uom = uom or details.stock_uom details.uom = uom or details.stock_uom
if uom: if uom:
@@ -1610,7 +1624,7 @@ def get_rejected_warehouses():
@frappe.whitelist() @frappe.whitelist()
def get_pick_list_query(doctype, txt, searchfield, start, page_len, filters): def get_pick_list_query(doctype: Any, txt: str, searchfield: Any, start: int, page_len: int, filters: dict):
frappe.has_permission("Pick List", throw=True) frappe.has_permission("Pick List", throw=True)
if not filters.get("company"): if not filters.get("company"):

View File

@@ -7,6 +7,7 @@ import json
import frappe import frappe
from frappe import _, throw from frappe import _, throw
from frappe.desk.notifications import clear_doctype_notifications from frappe.desk.notifications import clear_doctype_notifications
from frappe.model.document import Document
from frappe.model.mapper import get_mapped_doc from frappe.model.mapper import get_mapped_doc
from frappe.query_builder.functions import Abs, CombineDatetime, Sum from frappe.query_builder.functions import Abs, CombineDatetime, Sum
from frappe.utils import cint, flt, get_datetime, getdate, nowdate from frappe.utils import cint, flt, get_datetime, getdate, nowdate
@@ -246,7 +247,8 @@ class PurchaseReceipt(BuyingController):
from erpnext.stock.doctype.putaway_rule.putaway_rule import apply_putaway_rule from erpnext.stock.doctype.putaway_rule.putaway_rule import apply_putaway_rule
if self.get("items") and self.apply_putaway_rule and not self.get("is_return"): if self.get("items") and self.apply_putaway_rule and not self.get("is_return"):
apply_putaway_rule(self.doctype, self.get("items"), self.company) if items := apply_putaway_rule(self.doctype, self.get("items"), self.company):
self.items = items
def validate(self): def validate(self):
self.validate_posting_time() self.validate_posting_time()
@@ -1401,7 +1403,9 @@ def get_item_wise_returned_qty(pr_doc):
@frappe.whitelist() @frappe.whitelist()
def make_purchase_invoice(source_name, target_doc=None, args=None): def make_purchase_invoice(
source_name: str | None, target_doc: str | Document | None = None, args: dict | str | None = None
):
if args is None: if args is None:
args = {} args = {}
if isinstance(args, str): if isinstance(args, str):
@@ -1554,14 +1558,14 @@ def get_returned_qty_map(purchase_receipt):
@frappe.whitelist() @frappe.whitelist()
def make_purchase_return_against_rejected_warehouse(source_name): def make_purchase_return_against_rejected_warehouse(source_name: str):
from erpnext.controllers.sales_and_purchase_return import make_return_doc from erpnext.controllers.sales_and_purchase_return import make_return_doc
return make_return_doc("Purchase Receipt", source_name, return_against_rejected_qty=True) return make_return_doc("Purchase Receipt", source_name, return_against_rejected_qty=True)
@frappe.whitelist() @frappe.whitelist()
def make_purchase_return(source_name, target_doc=None): def make_purchase_return(source_name: str, target_doc: str | Document | None = None):
from erpnext.controllers.sales_and_purchase_return import make_return_doc from erpnext.controllers.sales_and_purchase_return import make_return_doc
return make_return_doc("Purchase Receipt", source_name, target_doc) return make_return_doc("Purchase Receipt", source_name, target_doc)
@@ -1574,7 +1578,7 @@ def update_purchase_receipt_status(docname, status):
@frappe.whitelist() @frappe.whitelist()
def make_stock_entry(source_name, target_doc=None): def make_stock_entry(source_name: str, target_doc: str | Document | None = None):
def set_missing_values(source, target): def set_missing_values(source, target):
target.stock_entry_type = "Material Transfer" target.stock_entry_type = "Material Transfer"
target.purpose = "Material Transfer" target.purpose = "Material Transfer"
@@ -1634,7 +1638,7 @@ def make_stock_entry(source_name, target_doc=None):
@frappe.whitelist() @frappe.whitelist()
def make_inter_company_delivery_note(source_name, target_doc=None): def make_inter_company_delivery_note(source_name: str, target_doc: str | Document | None = None):
return make_inter_company_transaction("Purchase Receipt", source_name, target_doc) return make_inter_company_transaction("Purchase Receipt", source_name, target_doc)
@@ -1644,7 +1648,7 @@ def update_regional_gl_entries(gl_list, doc):
@frappe.whitelist() @frappe.whitelist()
def make_lcv(doctype, docname): def make_lcv(doctype: str, docname: str):
landed_cost_voucher = frappe.new_doc("Landed Cost Voucher") landed_cost_voucher = frappe.new_doc("Landed Cost Voucher")
details = frappe.db.get_value(doctype, docname, ["supplier", "company", "base_grand_total"], as_dict=1) details = frappe.db.get_value(doctype, docname, ["supplier", "company", "base_grand_total"], as_dict=1)

View File

@@ -90,7 +90,7 @@ class PutawayRule(Document):
@frappe.whitelist() @frappe.whitelist()
def get_available_putaway_capacity(rule): def get_available_putaway_capacity(rule: str):
stock_capacity, item_code, warehouse = frappe.db.get_value( stock_capacity, item_code, warehouse = frappe.db.get_value(
"Putaway Rule", rule, ["stock_capacity", "item_code", "warehouse"] "Putaway Rule", rule, ["stock_capacity", "item_code", "warehouse"]
) )
@@ -100,7 +100,9 @@ def get_available_putaway_capacity(rule):
@frappe.whitelist() @frappe.whitelist()
def apply_putaway_rule(doctype, items, company, sync=None, purpose=None): def apply_putaway_rule(
doctype: str, items: list | str, company: str, sync: str | bool | None = None, purpose: str | None = None
):
"""Applies Putaway Rule on line items. """Applies Putaway Rule on line items.
items: List of Purchase Receipt/Stock Entry Items items: List of Purchase Receipt/Stock Entry Items
@@ -193,8 +195,8 @@ def apply_putaway_rule(doctype, items, company, sync=None, purpose=None):
show_unassigned_items_message(items_not_accomodated) show_unassigned_items_message(items_not_accomodated)
if updated_table and _items_changed(items, updated_table, doctype): if updated_table and _items_changed(items, updated_table, doctype):
items[:] = updated_table
frappe.msgprint(_("Applied putaway rules."), alert=True) frappe.msgprint(_("Applied putaway rules."), alert=True)
return updated_table
if sync and json.loads(sync): # sync with client side if sync and json.loads(sync): # sync with client side
return items return items

View File

@@ -2,6 +2,8 @@
# License: GNU General Public License v3. See license.txt # License: GNU General Public License v3. See license.txt
from typing import Any
import frappe import frappe
from frappe import _ from frappe import _
from frappe.model.document import Document from frappe.model.document import Document
@@ -364,7 +366,7 @@ class QualityInspection(Document):
@frappe.whitelist() @frappe.whitelist()
@frappe.validate_and_sanitize_search_inputs @frappe.validate_and_sanitize_search_inputs
def item_query(doctype, txt, searchfield, start, page_len, filters): def item_query(doctype: Any, txt: str | None, searchfield: Any, start: int, page_len: int, filters: dict):
from frappe.desk.reportview import get_match_cond from frappe.desk.reportview import get_match_cond
from_doctype = cstr(filters.get("from")) from_doctype = cstr(filters.get("from"))
@@ -420,7 +422,9 @@ def item_query(doctype, txt, searchfield, start, page_len, filters):
@frappe.whitelist() @frappe.whitelist()
@frappe.validate_and_sanitize_search_inputs @frappe.validate_and_sanitize_search_inputs
def quality_inspection_query(doctype, txt, searchfield, start, page_len, filters): def quality_inspection_query(
doctype: Any, txt: str | None, searchfield: Any, start: int, page_len: int, filters: dict
):
return frappe.get_all( return frappe.get_all(
"Quality Inspection", "Quality Inspection",
limit_start=start, limit_start=start,
@@ -437,7 +441,7 @@ def quality_inspection_query(doctype, txt, searchfield, start, page_len, filters
@frappe.whitelist() @frappe.whitelist()
def make_quality_inspection(source_name, target_doc=None): def make_quality_inspection(source_name: str, target_doc: Document | str | None = None):
def postprocess(source, doc): def postprocess(source, doc):
doc.inspected_by = frappe.session.user doc.inspected_by = frappe.session.user
doc.get_quality_inspection_template() doc.get_quality_inspection_template()

View File

@@ -5,6 +5,7 @@
import frappe import frappe
from frappe import _ from frappe import _
from frappe.model.document import Document from frappe.model.document import Document
from frappe.utils.data import DateTimeLikeObject
from erpnext.stock.utils import get_stock_balance, get_stock_value_on from erpnext.stock.utils import get_stock_balance, get_stock_value_on
@@ -32,7 +33,7 @@ class QuickStockBalance(Document):
@frappe.whitelist() @frappe.whitelist()
def get_stock_item_details(warehouse, date, item=None, barcode=None): def get_stock_item_details(warehouse: str, date: str, item: str | None = None, barcode: str | None = None):
out = {} out = {}
if barcode: if barcode:
out["item"] = frappe.db.get_value("Item Barcode", filters={"barcode": barcode}, fieldname=["parent"]) out["item"] = frappe.db.get_value("Item Barcode", filters={"barcode": barcode}, fieldname=["parent"])

View File

@@ -332,7 +332,7 @@ class RepostItemValuation(Document):
@frappe.whitelist() @frappe.whitelist()
def bulk_restart_reposting(names): def bulk_restart_reposting(names: str):
names = json.loads(names) names = json.loads(names)
for name in names: for name in names:
doc = frappe.get_doc("Repost Item Valuation", name) doc = frappe.get_doc("Repost Item Valuation", name)

View File

@@ -5,6 +5,7 @@ import collections
import csv import csv
import json import json
from collections import Counter, defaultdict from collections import Counter, defaultdict
from typing import Any
import frappe import frappe
import frappe.query_builder import frappe.query_builder
@@ -27,6 +28,7 @@ from frappe.utils import (
) )
from frappe.utils.csvutils import build_csv_response from frappe.utils.csvutils import build_csv_response
from erpnext.stock.doctype.purchase_receipt_item.purchase_receipt_item import PurchaseReceiptItem
from erpnext.stock.serial_batch_bundle import ( from erpnext.stock.serial_batch_bundle import (
BatchNoValuation, BatchNoValuation,
SerialNoValuation, SerialNoValuation,
@@ -1631,7 +1633,7 @@ class SerialandBatchBundle(Document):
self.delink_reference_from_batch() self.delink_reference_from_batch()
@frappe.whitelist() @frappe.whitelist()
def add_serial_batch(self, data): def add_serial_batch(self, data: str | dict):
serial_nos, batch_nos = [], [] serial_nos, batch_nos = [], []
if isinstance(data, str): if isinstance(data, str):
data = parse_json(data) data = parse_json(data)
@@ -1658,7 +1660,7 @@ class SerialandBatchBundle(Document):
@frappe.whitelist() @frappe.whitelist()
def download_blank_csv_template(content): def download_blank_csv_template(content: str | list):
csv_data = [] csv_data = []
if isinstance(content, str): if isinstance(content, str):
content = parse_json(content) content = parse_json(content)
@@ -1672,7 +1674,7 @@ def download_blank_csv_template(content):
@frappe.whitelist() @frappe.whitelist()
def upload_csv_file(item_code, file_path): def upload_csv_file(item_code: str, file_path: str):
serial_nos, batch_nos = [], [] serial_nos, batch_nos = [], []
serial_nos, batch_nos = get_serial_batch_from_csv(item_code, file_path) serial_nos, batch_nos = get_serial_batch_from_csv(item_code, file_path)
@@ -1770,7 +1772,7 @@ def get_serial_batch_from_data(item_code, kwargs):
@frappe.whitelist() @frappe.whitelist()
def create_serial_nos(item_code, serial_nos): def create_serial_nos(item_code: str, serial_nos: list | str):
serial_nos = get_serial_batch_from_data( serial_nos = get_serial_batch_from_data(
item_code, item_code,
{ {
@@ -1887,7 +1889,9 @@ def make_batch_nos(item_code, batch_nos):
@frappe.whitelist() @frappe.whitelist()
@frappe.validate_and_sanitize_search_inputs @frappe.validate_and_sanitize_search_inputs
def item_query(doctype, txt, searchfield, start, page_len, filters, as_dict=False): def item_query(
doctype: Any, txt: str, searchfield: str, start: int, page_len: int, filters: Any, as_dict: bool = False
):
item_filters = {"disabled": 0} item_filters = {"disabled": 0}
if txt: if txt:
item_filters["name"] = ("like", f"%{txt}%") item_filters["name"] = ("like", f"%{txt}%")
@@ -1902,7 +1906,13 @@ def item_query(doctype, txt, searchfield, start, page_len, filters, as_dict=Fals
@frappe.whitelist() @frappe.whitelist()
def get_serial_batch_ledgers(item_code=None, docstatus=None, voucher_no=None, name=None, child_row=None): def get_serial_batch_ledgers(
item_code: str | None = None,
docstatus: str | list | int | None = None,
voucher_no: str | None = None,
name: str | list | None = None,
child_row: dict | str | None = None,
):
filters = get_filters_for_bundle( filters = get_filters_for_bundle(
item_code=item_code, docstatus=docstatus, voucher_no=voucher_no, name=name, child_row=child_row item_code=item_code, docstatus=docstatus, voucher_no=voucher_no, name=name, child_row=child_row
) )
@@ -1978,7 +1988,13 @@ def get_reference_serial_and_batch_bundle(child_row):
@frappe.whitelist() @frappe.whitelist()
def add_serial_batch_ledgers(entries, child_row, doc, warehouse, do_not_save=False) -> object: def add_serial_batch_ledgers(
entries: list | str,
child_row: PurchaseReceiptItem | dict | str,
doc: Document,
warehouse: str | None = None,
do_not_save: bool = False,
):
if isinstance(child_row, str): if isinstance(child_row, str):
child_row = frappe._dict(parse_json(child_row)) child_row = frappe._dict(parse_json(child_row))
@@ -2131,7 +2147,7 @@ def update_serial_batch_no_ledgers(bundle, entries, child_row, parent_doc, wareh
@frappe.whitelist() @frappe.whitelist()
def update_serial_or_batch(bundle_id, serial_no=None, batch_no=None): def update_serial_or_batch(bundle_id: str, serial_no: str | None = None, batch_no: str | None = None):
if batch_no and not serial_no: if batch_no and not serial_no:
if qty := frappe.db.get_value( if qty := frappe.db.get_value(
"Serial and Batch Entry", {"parent": bundle_id, "batch_no": batch_no}, "qty" "Serial and Batch Entry", {"parent": bundle_id, "batch_no": batch_no}, "qty"
@@ -3303,12 +3319,14 @@ def get_stock_ledgers_batches(kwargs):
@frappe.whitelist() @frappe.whitelist()
def get_batch_no_from_serial_no(serial_no): def get_batch_no_from_serial_no(serial_no: str):
return frappe.get_cached_value("Serial No", serial_no, "batch_no") return frappe.get_cached_value("Serial No", serial_no, "batch_no")
@frappe.whitelist() @frappe.whitelist()
def is_serial_batch_no_exists(item_code, type_of_transaction, serial_no=None, batch_no=None): def is_serial_batch_no_exists(
item_code: str, type_of_transaction: str, serial_no: str | None = None, batch_no: str | None = None
):
if serial_no and not frappe.db.exists("Serial No", serial_no): if serial_no and not frappe.db.exists("Serial No", serial_no):
if type_of_transaction != "Inward": if type_of_transaction != "Inward":
frappe.throw(_("Serial No {0} does not exists").format(serial_no)) frappe.throw(_("Serial No {0} does not exists").format(serial_no))
@@ -3337,7 +3355,7 @@ def make_batch_no(batch_no, item_code):
@frappe.whitelist() @frappe.whitelist()
def is_duplicate_serial_no(bundle_id, serial_no): def is_duplicate_serial_no(bundle_id: str, serial_no: str):
return frappe.db.exists("Serial and Batch Entry", {"parent": bundle_id, "serial_no": serial_no}) return frappe.db.exists("Serial and Batch Entry", {"parent": bundle_id, "serial_no": serial_no})

View File

@@ -221,7 +221,7 @@ def auto_fetch_serial_number(
@frappe.whitelist() @frappe.whitelist()
def get_pos_reserved_serial_nos(filters): def get_pos_reserved_serial_nos(filters: str | dict):
if isinstance(filters, str): if isinstance(filters, str):
filters = json.loads(filters) filters = json.loads(filters)

View File

@@ -114,19 +114,19 @@ class Shipment(Document):
@frappe.whitelist() @frappe.whitelist()
def get_address_name(ref_doctype, docname): def get_address_name(ref_doctype: str, docname: str):
# Return address name # Return address name
return get_party_shipping_address(ref_doctype, docname) return get_party_shipping_address(ref_doctype, docname)
@frappe.whitelist() @frappe.whitelist()
def get_contact_name(ref_doctype, docname): def get_contact_name(ref_doctype: str, docname: str):
# Return address name # Return address name
return get_default_contact(ref_doctype, docname) return get_default_contact(ref_doctype, docname)
@frappe.whitelist() @frappe.whitelist()
def get_company_contact(user): def get_company_contact(user: str):
contact = frappe.db.get_value( contact = frappe.db.get_value(
"User", "User",
user, user,

View File

@@ -7,6 +7,7 @@ from collections import defaultdict
import frappe import frappe
from frappe import _, bold from frappe import _, bold
from frappe.model.document import Document
from frappe.model.mapper import get_mapped_doc from frappe.model.mapper import get_mapped_doc
from frappe.query_builder import DocType from frappe.query_builder import DocType
from frappe.query_builder.functions import Sum from frappe.query_builder.functions import Sum
@@ -215,7 +216,10 @@ class StockEntry(StockController, SubcontractingInwardController):
apply_rule = self.apply_putaway_rule and (self.purpose in ["Material Transfer", "Material Receipt"]) apply_rule = self.apply_putaway_rule and (self.purpose in ["Material Transfer", "Material Receipt"])
if self.get("items") and apply_rule: if self.get("items") and apply_rule:
apply_putaway_rule(self.doctype, self.get("items"), self.company, purpose=self.purpose) if items := apply_putaway_rule(
self.doctype, self.get("items"), self.company, purpose=self.purpose
):
self.items = items
if self.project: if self.project:
for item in self.items: for item in self.items:
@@ -2094,7 +2098,7 @@ class StockEntry(StockController, SubcontractingInwardController):
return False return False
@frappe.whitelist() @frappe.whitelist()
def get_item_details(self, args: ItemDetailsCtx = None, for_update=False): def get_item_details(self, args: ItemDetailsCtx | None = None, for_update: bool = False):
item = frappe.qb.DocType("Item") item = frappe.qb.DocType("Item")
item_default = frappe.qb.DocType("Item Default") item_default = frappe.qb.DocType("Item Default")
@@ -3515,7 +3519,7 @@ class StockEntry(StockController, SubcontractingInwardController):
@frappe.whitelist() @frappe.whitelist()
def move_sample_to_retention_warehouse(company, items): def move_sample_to_retention_warehouse(company: str, items: str | list):
from erpnext.stock.serial_batch_bundle import ( from erpnext.stock.serial_batch_bundle import (
SerialBatchCreation, SerialBatchCreation,
get_batch_nos, get_batch_nos,
@@ -3596,7 +3600,7 @@ def move_sample_to_retention_warehouse(company, items):
@frappe.whitelist() @frappe.whitelist()
def make_stock_in_entry(source_name, target_doc=None): def make_stock_in_entry(source_name: str, target_doc: str | Document | None = None):
def set_missing_values(source, target): def set_missing_values(source, target):
target.stock_entry_type = "Material Transfer" target.stock_entry_type = "Material Transfer"
target.set_missing_values() target.set_missing_values()
@@ -3647,7 +3651,7 @@ def make_stock_in_entry(source_name, target_doc=None):
@frappe.whitelist() @frappe.whitelist()
def get_work_order_details(work_order, company): def get_work_order_details(work_order: str, company: str):
work_order = frappe.get_doc("Work Order", work_order) work_order = frappe.get_doc("Work Order", work_order)
pending_qty_to_produce = flt(work_order.qty) - flt(work_order.produced_qty) pending_qty_to_produce = flt(work_order.qty) - flt(work_order.produced_qty)
@@ -3766,7 +3770,7 @@ def get_valuation_rate_for_finished_good_entry(work_order):
@frappe.whitelist() @frappe.whitelist()
def get_uom_details(item_code, uom, qty): def get_uom_details(item_code: str, uom: str, qty: float | None):
"""Returns dict `{"conversion_factor": [value], "transfer_qty": qty * [value]}` """Returns dict `{"conversion_factor": [value], "transfer_qty": qty * [value]}`
:param args: dict with `item_code`, `uom` and `qty`""" :param args: dict with `item_code`, `uom` and `qty`"""
conversion_factor = get_conversion_factor(item_code, uom).get("conversion_factor") conversion_factor = get_conversion_factor(item_code, uom).get("conversion_factor")
@@ -3825,7 +3829,7 @@ def get_expired_batches():
@frappe.whitelist() @frappe.whitelist()
def get_warehouse_details(args): def get_warehouse_details(args: str | dict):
if isinstance(args, str): if isinstance(args, str):
args = json.loads(args) args = json.loads(args)
@@ -3847,7 +3851,7 @@ def get_warehouse_details(args):
@frappe.whitelist() @frappe.whitelist()
def validate_sample_quantity(item_code, sample_quantity, qty, batch_no=None): def validate_sample_quantity(item_code: str, sample_quantity: int, qty: float, batch_no: str | None = None):
if cint(qty) < cint(sample_quantity): if cint(qty) < cint(sample_quantity):
frappe.throw( frappe.throw(
_("Sample quantity {0} cannot be more than received quantity {1}").format(sample_quantity, qty) _("Sample quantity {0} cannot be more than received quantity {1}").format(sample_quantity, qty)
@@ -3916,7 +3920,7 @@ def get_supplied_items(
@frappe.whitelist() @frappe.whitelist()
def get_items_from_subcontract_order(source_name, target_doc=None): def get_items_from_subcontract_order(source_name: str, target_doc: str | Document | None = None):
from erpnext.controllers.subcontracting_controller import make_rm_stock_entry from erpnext.controllers.subcontracting_controller import make_rm_stock_entry
if isinstance(target_doc, str): if isinstance(target_doc, str):

View File

@@ -2,9 +2,12 @@
# License: GNU General Public License v3. See license.txt # License: GNU General Public License v3. See license.txt
from datetime import datetime, timedelta
import frappe import frappe
from frappe import _, bold, json, msgprint from frappe import _, bold, json, msgprint
from frappe.utils import add_to_date, cint, cstr, flt, now from frappe.utils import add_to_date, cint, cstr, flt, now
from frappe.utils.data import DateTimeLikeObject
import erpnext import erpnext
from erpnext.accounts.utils import get_company_default from erpnext.accounts.utils import get_company_default
@@ -16,6 +19,7 @@ from erpnext.stock.doctype.serial_and_batch_bundle.serial_and_batch_bundle impor
get_available_serial_nos, get_available_serial_nos,
) )
from erpnext.stock.doctype.serial_no.serial_no import get_serial_nos from erpnext.stock.doctype.serial_no.serial_no import get_serial_nos
from erpnext.stock.doctype.stock_reconciliation_item.stock_reconciliation_item import StockReconciliationItem
from erpnext.stock.utils import get_combine_datetime, get_incoming_rate, get_stock_balance from erpnext.stock.utils import get_combine_datetime, get_incoming_rate, get_stock_balance
@@ -1260,7 +1264,14 @@ def get_batch_qty_for_stock_reco(
@frappe.whitelist() @frappe.whitelist()
def get_items(warehouse, posting_date, posting_time, company, item_code=None, ignore_empty_stock=False): def get_items(
warehouse: str,
posting_date: DateTimeLikeObject,
posting_time: DateTimeLikeObject,
company: str,
item_code: str | None = None,
ignore_empty_stock: bool | str | int = False,
):
ignore_empty_stock = cint(ignore_empty_stock) ignore_empty_stock = cint(ignore_empty_stock)
items = [] items = []
if item_code and warehouse: if item_code and warehouse:
@@ -1426,13 +1437,13 @@ def get_itemwise_batch(warehouse, posting_date, company, item_code=None):
def get_stock_balance_for( def get_stock_balance_for(
item_code: str, item_code: str,
warehouse: str, warehouse: str,
posting_date, posting_date: DateTimeLikeObject,
posting_time, posting_time: DateTimeLikeObject | timedelta,
batch_no: str | None = None, batch_no: str | None = None,
with_valuation_rate: bool = True, with_valuation_rate: bool = True,
inventory_dimensions_dict=None, inventory_dimensions_dict: dict | None = None,
row=None, row: StockReconciliationItem | str | dict | None = None,
company=None, company: str | None = None,
): ):
frappe.has_permission("Stock Reconciliation", "write", throw=True) frappe.has_permission("Stock Reconciliation", "write", throw=True)
@@ -1519,7 +1530,7 @@ def get_stock_balance_for(
@frappe.whitelist() @frappe.whitelist()
def get_difference_account(purpose, company): def get_difference_account(purpose: str, company: str):
if purpose == "Stock Reconciliation": if purpose == "Stock Reconciliation":
account = get_company_default(company, "stock_adjustment_account") account = get_company_default(company, "stock_adjustment_account")
else: else:

View File

@@ -3,6 +3,7 @@
import json import json
from typing import Any
import frappe import frappe
from frappe import _, throw from frappe import _, throw
@@ -191,7 +192,13 @@ class Warehouse(NestedSet):
@frappe.whitelist() @frappe.whitelist()
def get_children(doctype, parent=None, company=None, is_root=False, include_disabled=False): def get_children(
doctype: str,
parent: str | None = None,
company: str | None = None,
is_root: bool = False,
include_disabled: bool | str = False,
):
if is_root: if is_root:
parent = "" parent = ""
@@ -224,7 +231,7 @@ def add_node():
@frappe.whitelist() @frappe.whitelist()
def convert_to_group_or_ledger(docname=None): def convert_to_group_or_ledger(docname: str | None = None):
if not docname: if not docname:
docname = frappe.form_dict.docname docname = frappe.form_dict.docname
return frappe.get_doc("Warehouse", docname).convert_to_group_or_ledger() return frappe.get_doc("Warehouse", docname).convert_to_group_or_ledger()
@@ -299,7 +306,9 @@ def apply_warehouse_filter(query, sle, filters):
@frappe.whitelist() @frappe.whitelist()
@frappe.validate_and_sanitize_search_inputs @frappe.validate_and_sanitize_search_inputs
def get_warehouses_for_reorder(doctype, txt, searchfield, start, page_len, filters): def get_warehouses_for_reorder(
doctype: str, txt: Any, searchfield: Any, start: int, page_len: int, filters: dict
):
filters = frappe._dict(filters or {}) filters = frappe._dict(filters or {})
if filters.warehouse and not frappe.db.exists("Warehouse", filters.warehouse): if filters.warehouse and not frappe.db.exists("Warehouse", filters.warehouse):

View File

@@ -3,6 +3,7 @@
import json import json
from typing import Any
import frappe import frappe
from frappe import _, throw from frappe import _, throw
@@ -54,7 +55,12 @@ def _preprocess_ctx(ctx):
@frappe.whitelist() @frappe.whitelist()
@erpnext.normalize_ctx_input(ItemDetailsCtx) @erpnext.normalize_ctx_input(ItemDetailsCtx)
def get_item_details(ctx, doc=None, for_validate=False, overwrite_warehouse=True) -> ItemDetails: def get_item_details(
ctx: ItemDetailsCtx,
doc: Document | str | None = None,
for_validate: bool = False,
overwrite_warehouse: bool = True,
):
""" """
ctx = { ctx = {
"item_code": "", "item_code": "",
@@ -649,7 +655,13 @@ def get_barcode_data(items_list=None, item_code=None):
@frappe.whitelist() @frappe.whitelist()
def get_item_tax_info(doc, tax_category, item_codes, item_rates=None, item_tax_templates=None): def get_item_tax_info(
doc: Document | str | None,
tax_category: str,
item_codes: list,
item_rates: dict | None = None,
item_tax_templates: dict | None = None,
):
out = {} out = {}
if item_tax_templates is None: if item_tax_templates is None:
@@ -690,7 +702,7 @@ def get_item_tax_info(doc, tax_category, item_codes, item_rates=None, item_tax_t
@frappe.whitelist() @frappe.whitelist()
@erpnext.normalize_ctx_input(ItemDetailsCtx) @erpnext.normalize_ctx_input(ItemDetailsCtx)
def get_item_tax_template(ctx, item=None, out: ItemDetails | None = None): def get_item_tax_template(ctx: ItemDetailsCtx, item: Document | None = None, out: ItemDetails | None = None):
""" """
Determines item_tax template from item or parent item groups. Determines item_tax template from item or parent item groups.
@@ -821,7 +833,7 @@ def is_within_valid_range(ctx: ItemDetailsCtx, tax) -> bool:
@frappe.whitelist() @frappe.whitelist()
def get_item_tax_map(*, doc: str | dict | Document, tax_template: str | None = None, as_json=True): def get_item_tax_map(*, doc: str | dict | Document, tax_template: str | None = None, as_json: bool = True):
doc = parse_json(doc) doc = parse_json(doc)
item_tax_map = {} item_tax_map = {}
for t in (t for t in (doc.get("taxes") or []) if not t.get("set_by_item_tax_template")): for t in (t for t in (doc.get("taxes") or []) if not t.get("set_by_item_tax_template")):
@@ -838,7 +850,7 @@ def get_item_tax_map(*, doc: str | dict | Document, tax_template: str | None = N
@frappe.whitelist() @frappe.whitelist()
@erpnext.normalize_ctx_input(ItemDetailsCtx) @erpnext.normalize_ctx_input(ItemDetailsCtx)
def calculate_service_end_date(ctx: ItemDetailsCtx, item=None): def calculate_service_end_date(ctx: ItemDetailsCtx, item: Document | None = None):
_preprocess_ctx(ctx) _preprocess_ctx(ctx)
if not item: if not item:
item = frappe.get_cached_doc("Item", ctx.item_code) item = frappe.get_cached_doc("Item", ctx.item_code)
@@ -1156,7 +1168,7 @@ def get_item_price(
@frappe.whitelist() @frappe.whitelist()
def get_batch_based_item_price(pctx: ItemPriceCtx | dict | str, item_code) -> float: def get_batch_based_item_price(pctx: ItemPriceCtx | dict | str, item_code: str):
pctx = parse_json(pctx) pctx = parse_json(pctx)
item_price = get_item_price(pctx, item_code, force_batch_no=True) item_price = get_item_price(pctx, item_code, force_batch_no=True)
@@ -1172,7 +1184,7 @@ def get_batch_based_item_price(pctx: ItemPriceCtx | dict | str, item_code) -> fl
@erpnext.normalize_ctx_input(ItemDetailsCtx) @erpnext.normalize_ctx_input(ItemDetailsCtx)
def get_price_list_rate_for(ctx: ItemDetailsCtx, item_code): def get_price_list_rate_for(ctx: ItemDetailsCtx, item_code: str):
""" """
:param customer: link to Customer DocType :param customer: link to Customer DocType
:param supplier: link to Supplier DocType :param supplier: link to Supplier DocType
@@ -1355,7 +1367,7 @@ def get_pos_profile_item_details_(ctx: ItemDetailsCtx, company, pos_profile=None
@frappe.whitelist() @frappe.whitelist()
def get_pos_profile(company, pos_profile=None, user=None): def get_pos_profile(company: str, pos_profile: str | None = None, user: str | None = None):
if pos_profile: if pos_profile:
return frappe.get_cached_doc("POS Profile", pos_profile) return frappe.get_cached_doc("POS Profile", pos_profile)
@@ -1391,7 +1403,7 @@ def get_pos_profile(company, pos_profile=None, user=None):
@frappe.whitelist() @frappe.whitelist()
def get_conversion_factor(item_code, uom): def get_conversion_factor(item_code: str | None, uom: str):
item = frappe.get_cached_value("Item", item_code, ["variant_of", "stock_uom"], as_dict=True) item = frappe.get_cached_value("Item", item_code, ["variant_of", "stock_uom"], as_dict=True)
if not item_code or not item or uom == item.stock_uom: if not item_code or not item or uom == item.stock_uom:
return {"conversion_factor": 1.0} return {"conversion_factor": 1.0}
@@ -1422,7 +1434,7 @@ def get_conversion_factor(item_code, uom):
@frappe.whitelist() @frappe.whitelist()
def get_projected_qty(item_code, warehouse): def get_projected_qty(item_code: str, warehouse: str):
return { return {
"projected_qty": frappe.db.get_value( "projected_qty": frappe.db.get_value(
"Bin", {"item_code": item_code, "warehouse": warehouse}, "projected_qty" "Bin", {"item_code": item_code, "warehouse": warehouse}, "projected_qty"
@@ -1431,7 +1443,9 @@ def get_projected_qty(item_code, warehouse):
@frappe.whitelist() @frappe.whitelist()
def get_bin_details(item_code, warehouse, company=None, include_child_warehouses=False): def get_bin_details(
item_code: str, warehouse: str | None, company: str | None = None, include_child_warehouses: bool = False
):
bin_details = {"projected_qty": 0, "actual_qty": 0, "reserved_qty": 0} bin_details = {"projected_qty": 0, "actual_qty": 0, "reserved_qty": 0}
if warehouse: if warehouse:
@@ -1472,7 +1486,7 @@ def get_company_total_stock(item_code, company):
@frappe.whitelist() @frappe.whitelist()
def get_batch_qty(batch_no, warehouse, item_code): def get_batch_qty(batch_no: str, warehouse: str, item_code: str):
from erpnext.stock.doctype.batch import batch from erpnext.stock.doctype.batch import batch
if batch_no: if batch_no:
@@ -1481,7 +1495,7 @@ def get_batch_qty(batch_no, warehouse, item_code):
@frappe.whitelist() @frappe.whitelist()
@erpnext.normalize_ctx_input(ItemDetailsCtx) @erpnext.normalize_ctx_input(ItemDetailsCtx)
def apply_price_list(ctx, as_doc=False, doc=None): def apply_price_list(ctx: ItemDetailsCtx, as_doc: bool = False, doc: Document | None = None):
"""Apply pricelist on a document-like dict object and return as """Apply pricelist on a document-like dict object and return as
{'parent': dict, 'children': list} {'parent': dict, 'children': list}
@@ -1581,7 +1595,7 @@ def get_price_list_currency_and_exchange_rate(ctx: ItemDetailsCtx):
@frappe.whitelist() @frappe.whitelist()
def get_default_bom(item_code=None): def get_default_bom(item_code: str | None = None):
def _get_bom(item): def _get_bom(item):
bom = frappe.get_all("BOM", dict(item=item, is_active=True, is_default=True, docstatus=1), limit=1) bom = frappe.get_all("BOM", dict(item=item, is_active=True, is_default=True, docstatus=1), limit=1)
return bom[0].name if bom else None return bom[0].name if bom else None
@@ -1599,7 +1613,7 @@ def get_default_bom(item_code=None):
@frappe.whitelist() @frappe.whitelist()
def get_valuation_rate(item_code, company, warehouse=None): def get_valuation_rate(item_code: str, company: str, warehouse: str | None = None):
if frappe.get_cached_value("Warehouse", warehouse, "is_group"): if frappe.get_cached_value("Warehouse", warehouse, "is_group"):
return {"valuation_rate": 0.0} return {"valuation_rate": 0.0}
@@ -1640,7 +1654,7 @@ def get_gross_profit(out: ItemDetails):
@frappe.whitelist() @frappe.whitelist()
def get_serial_no(_args, serial_nos=None, sales_order=None): def get_serial_no(_args: Any, serial_nos: list | None = None, sales_order: str | None = None):
serial_nos = serial_nos or [] serial_nos = serial_nos or []
return serial_nos return serial_nos

View File

@@ -134,7 +134,7 @@ def get_linked_cancelled_sabb(filters):
@frappe.whitelist() @frappe.whitelist()
def fix_sabb_entries(selected_rows): def fix_sabb_entries(selected_rows: str | list):
if isinstance(selected_rows, str): if isinstance(selected_rows, str):
selected_rows = frappe.parse_json(selected_rows) selected_rows = frappe.parse_json(selected_rows)

View File

@@ -1,6 +1,8 @@
# Copyright (c) 2023, Frappe Technologies Pvt. Ltd. and contributors # Copyright (c) 2023, Frappe Technologies Pvt. Ltd. and contributors
# For license information, please see license.txt # For license information, please see license.txt
from typing import Any
import frappe import frappe
from frappe import _ from frappe import _
@@ -179,7 +181,7 @@ def get_columns(filters, data):
@frappe.whitelist() @frappe.whitelist()
@frappe.validate_and_sanitize_search_inputs @frappe.validate_and_sanitize_search_inputs
def get_voucher_type(doctype, txt, searchfield, start, page_len, filters): def get_voucher_type(doctype: Any, txt: str, searchfield: Any, start: int, page_len: int, filters: dict):
child_doctypes = frappe.get_all( child_doctypes = frappe.get_all(
"DocField", "DocField",
filters={"fieldname": "serial_and_batch_bundle"}, filters={"fieldname": "serial_and_batch_bundle"},
@@ -196,7 +198,7 @@ def get_voucher_type(doctype, txt, searchfield, start, page_len, filters):
@frappe.whitelist() @frappe.whitelist()
@frappe.validate_and_sanitize_search_inputs @frappe.validate_and_sanitize_search_inputs
def get_serial_nos(doctype, txt, searchfield, start, page_len, filters): def get_serial_nos(doctype: Any, txt: str, searchfield: Any, start: int, page_len: int, filters: dict):
query_filters = {} query_filters = {}
if txt: if txt:
@@ -224,7 +226,7 @@ def get_serial_nos(doctype, txt, searchfield, start, page_len, filters):
@frappe.whitelist() @frappe.whitelist()
@frappe.validate_and_sanitize_search_inputs @frappe.validate_and_sanitize_search_inputs
def get_batch_nos(doctype, txt, searchfield, start, page_len, filters): def get_batch_nos(doctype: Any, txt: str, searchfield: Any, start: int, page_len: int, filters: dict):
query_filters = {} query_filters = {}
if filters.get("voucher_no") and txt: if filters.get("voucher_no") and txt:

View File

@@ -174,7 +174,7 @@ def get_columns(filters):
@frappe.whitelist() @frappe.whitelist()
def create_reposting_entries(rows, company): def create_reposting_entries(rows: str | list, company: str):
if isinstance(rows, str): if isinstance(rows, str):
rows = parse_json(rows) rows = parse_json(rows)

View File

@@ -296,7 +296,7 @@ def get_columns():
@frappe.whitelist() @frappe.whitelist()
def create_reposting_entries(rows, item_code=None, warehouse=None): def create_reposting_entries(rows: str | list, item_code: str | None = None, warehouse: str | None = None):
if isinstance(rows, str): if isinstance(rows, str):
rows = parse_json(rows) rows = parse_json(rows)

View File

@@ -100,7 +100,7 @@ def get_data(filters=None):
@frappe.whitelist() @frappe.whitelist()
def update_batch_qty(selected_batches=None): def update_batch_qty(selected_batches: str | None = None):
if not selected_batches: if not selected_batches:
return return

View File

@@ -2,12 +2,14 @@
# License: GNU General Public License v3. See license.txt # License: GNU General Public License v3. See license.txt
import datetime
import json import json
import frappe import frappe
from frappe import _ from frappe import _
from frappe.query_builder.functions import IfNull, Sum from frappe.query_builder.functions import IfNull, Sum
from frappe.utils import cstr, flt, get_link_to_form, get_time, getdate, nowdate, nowtime from frappe.utils import cstr, flt, get_link_to_form, get_time, getdate, nowdate, nowtime
from frappe.utils.data import DateTimeLikeObject
import erpnext import erpnext
from erpnext.stock.doctype.serial_and_batch_bundle.serial_and_batch_bundle import get_available_serial_nos from erpnext.stock.doctype.serial_and_batch_bundle.serial_and_batch_bundle import get_available_serial_nos
@@ -93,13 +95,13 @@ def get_stock_value_on(
@frappe.whitelist() @frappe.whitelist()
def get_stock_balance( def get_stock_balance(
item_code, item_code: str,
warehouse, warehouse: str,
posting_date=None, posting_date: DateTimeLikeObject | None = None,
posting_time=None, posting_time: DateTimeLikeObject | datetime.timedelta | None = None,
with_valuation_rate=False, with_valuation_rate: bool = False,
with_serial_no=False, with_serial_no: bool = False,
inventory_dimensions_dict=None, inventory_dimensions_dict: dict | None = None,
): ):
"""Returns stock balance quantity at given warehouse on given posting date or current date. """Returns stock balance quantity at given warehouse on given posting date or current date.
@@ -165,7 +167,7 @@ def get_serial_nos_data(serial_nos):
@frappe.whitelist() @frappe.whitelist()
def get_latest_stock_qty(item_code, warehouse=None): def get_latest_stock_qty(item_code: str, warehouse: str | None = None):
values, condition = [item_code], "" values, condition = [item_code], ""
if warehouse: if warehouse:
lft, rgt, is_group = frappe.db.get_value("Warehouse", warehouse, ["lft", "rgt", "is_group"]) lft, rgt, is_group = frappe.db.get_value("Warehouse", warehouse, ["lft", "rgt", "is_group"])
@@ -237,7 +239,7 @@ def _create_bin(item_code, warehouse):
@frappe.whitelist() @frappe.whitelist()
def get_incoming_rate(args, raise_error_if_no_rate=True, fallbacks: bool = True): def get_incoming_rate(args: dict | str, raise_error_if_no_rate: bool = True, fallbacks: bool = True):
"""Get Incoming Rate based on valuation method""" """Get Incoming Rate based on valuation method"""
from erpnext.stock.stock_ledger import get_previous_sle, get_valuation_rate from erpnext.stock.stock_ledger import get_previous_sle, get_valuation_rate