Merge pull request #52734 from khushi8112/type-hint-for-asset-module

refactor(assets): add type annotations in asset module

erpnext/assets/doctype/asset/asset.py

@@ -1092,7 +1092,7 @@ def get_asset_naming_series():
 
 
 @frappe.whitelist()
-def make_sales_invoice(asset, item_code, company, sell_qty, serial_no=None):
+def make_sales_invoice(asset: str, item_code: str, company: str, sell_qty: int, serial_no: str | None = None):
 	asset_doc = frappe.get_doc("Asset", asset)
 	si = frappe.new_doc("Sales Invoice")
 	si.company = company
@@ -1125,7 +1125,13 @@ def make_sales_invoice(asset, item_code, company, sell_qty, serial_no=None):
 
 
 @frappe.whitelist()
-def create_asset_maintenance(asset, item_code, item_name, asset_category, company):
+def create_asset_maintenance(
+	asset: str,
+	item_code: str,
+	item_name: str,
+	asset_category: str,
+	company: str,
+):
 	asset_maintenance = frappe.new_doc("Asset Maintenance")
 	asset_maintenance.update(
 		{
@@ -1140,14 +1146,23 @@ def create_asset_maintenance(asset, item_code, item_name, asset_category, compan
 
 
 @frappe.whitelist()
-def create_asset_repair(company, asset, asset_name):
+def create_asset_repair(
+	company: str,
+	asset: str,
+	asset_name: str,
+):
 	asset_repair = frappe.new_doc("Asset Repair")
 	asset_repair.update({"company": company, "asset": asset, "asset_name": asset_name})
 	return asset_repair
 
 
 @frappe.whitelist()
-def create_asset_capitalization(company, asset, asset_name, item_code):
+def create_asset_capitalization(
+	company: str,
+	asset: str,
+	asset_name: str,
+	item_code: str,
+):
 	asset_capitalization = frappe.new_doc("Asset Capitalization")
 	asset_capitalization.update(
 		{
@@ -1161,35 +1176,22 @@ def create_asset_capitalization(company, asset, asset_name, item_code):
 
 
 @frappe.whitelist()
-def create_asset_value_adjustment(asset, asset_category, company):
+def create_asset_value_adjustment(
+	asset: str,
+	asset_category: str,
+	company: str,
+):
 	asset_value_adjustment = frappe.new_doc("Asset Value Adjustment")
 	asset_value_adjustment.update({"asset": asset, "company": company, "asset_category": asset_category})
 	return asset_value_adjustment
 
 
 @frappe.whitelist()
-def transfer_asset(args):
-	args = json.loads(args)
-
-	if args.get("serial_no"):
-		args["quantity"] = len(args.get("serial_no").split("\n"))
-
-	movement_entry = frappe.new_doc("Asset Movement")
-	movement_entry.update(args)
-	movement_entry.insert()
-	movement_entry.submit()
-
-	frappe.db.commit()
-
-	frappe.msgprint(
-		_("Asset Movement record {0} created")
-		.format("<a href='/app/Form/Asset Movement/{0}'>{0}</a>")
-		.format(movement_entry.name)
-	)
-
-
-@frappe.whitelist()
-def get_item_details(item_code, asset_category, net_purchase_amount):
+def get_item_details(
+	item_code: str,
+	asset_category: str,
+	net_purchase_amount: float,
+):
 	asset_category_doc = frappe.get_cached_doc("Asset Category", asset_category)
 	books = []
 	for d in asset_category_doc.finance_books:
@@ -1239,7 +1241,7 @@ def get_asset_account(account_name, asset=None, asset_category=None, company=Non
 
 
 @frappe.whitelist()
-def make_journal_entry(asset_name):
+def make_journal_entry(asset_name: str):
 	asset = frappe.get_doc("Asset", asset_name)
 	(
 		fixed_asset_account,
@@ -1281,7 +1283,10 @@ def make_journal_entry(asset_name):
 
 
 @frappe.whitelist()
-def make_asset_movement(assets, purpose=None):
+def make_asset_movement(
+	assets: list[dict] | str,
+	purpose: str = "Transfer",
+):
 	import json
 
 	if isinstance(assets, str):
@@ -1291,7 +1296,7 @@ def make_asset_movement(assets, purpose=None):
 		frappe.throw(_("At least one asset has to be selected."))
 
 	asset_movement = frappe.new_doc("Asset Movement")
-	asset_movement.quantity = len(assets)
+	asset_movement.purpose = purpose
 	for asset in assets:
 		asset = frappe.get_doc("Asset", asset.get("name"))
 		asset_movement.company = asset.get("company")
@@ -1313,7 +1318,10 @@ def is_cwip_accounting_enabled(asset_category):
 
 
 @frappe.whitelist()
-def get_asset_value_after_depreciation(asset_name, finance_book=None):
+def get_asset_value_after_depreciation(
+	asset_name: str,
+	finance_book: str | None = None,
+):
 	asset = frappe.get_doc("Asset", asset_name)
 	if not asset.calculate_depreciation:
 		return flt(asset.value_after_depreciation)
@@ -1322,7 +1330,7 @@ def get_asset_value_after_depreciation(asset_name, finance_book=None):
 
 
 @frappe.whitelist()
-def has_active_capitalization(asset):
+def has_active_capitalization(asset: str):
 	active_capitalizations = frappe.db.count(
 		"Asset Capitalization", filters={"target_asset": asset, "docstatus": 1}
 	)
@@ -1330,7 +1338,11 @@ def has_active_capitalization(asset):
 
 
 @frappe.whitelist()
-def get_values_from_purchase_doc(purchase_doc_name, item_code, doctype):
+def get_values_from_purchase_doc(
+	purchase_doc_name: str,
+	item_code: str,
+	doctype: str,
+):
 	purchase_doc = frappe.get_doc(doctype, purchase_doc_name)
 	matching_items = [item for item in purchase_doc.items if item.item_code == item_code]
 
@@ -1352,7 +1364,7 @@ def get_values_from_purchase_doc(purchase_doc_name, item_code, doctype):
 
 
 @frappe.whitelist()
-def split_asset(asset_name, split_qty):
+def split_asset(asset_name: str, split_qty: int):
 	"""Split an asset into two based on the given quantity."""
 	existing_asset = frappe.get_doc("Asset", asset_name)
 	split_qty = cint(split_qty)

erpnext/assets/doctype/asset/depreciation.py

@@ -7,6 +7,7 @@ from frappe import _
 from frappe.query_builder import Order
 from frappe.query_builder.functions import Max, Min
 from frappe.utils import (
+	DateTimeLikeObject,
 	add_months,
 	cint,
 	flt,
@@ -161,11 +162,11 @@ def get_depr_cost_center_and_series():
 
 @frappe.whitelist()
 def make_depreciation_entry(
-	depr_schedule_name,
-	date=None,
-	sch_start_idx=None,
-	sch_end_idx=None,
-	accounting_dimensions=None,
+	depr_schedule_name: str,
+	date: DateTimeLikeObject | None = None,
+	sch_start_idx: int | None = None,
+	sch_end_idx: int | None = None,
+	accounting_dimensions: list[dict] | None = None,
 ):
 	frappe.has_permission("Journal Entry", throw=True)
 	date = date or today()
@@ -356,7 +357,7 @@ def get_message_for_depr_entry_posting_error(asset_links, error_log_links):
 
 
 @frappe.whitelist()
-def scrap_asset(asset_name, scrap_date=None):
+def scrap_asset(asset_name: str, scrap_date: DateTimeLikeObject | None = None):
 	asset = frappe.get_doc("Asset", asset_name)
 	scrap_date = getdate(scrap_date) or getdate(today())
 	asset.db_set("disposal_date", scrap_date)
@@ -445,7 +446,7 @@ def create_journal_entry_for_scrap(asset, scrap_date):
 
 
 @frappe.whitelist()
-def restore_asset(asset_name):
+def restore_asset(asset_name: str):
 	asset = frappe.get_doc("Asset", asset_name)
 	reverse_depreciation_entry_made_on_disposal(asset)
 	reset_depreciation_schedule(asset, get_note_for_restore(asset))
@@ -772,7 +773,7 @@ def get_profit_gl_entries(
 
 
 @frappe.whitelist()
-def get_disposal_account_and_cost_center(company):
+def get_disposal_account_and_cost_center(company: str):
 	disposal_account, depreciation_cost_center = frappe.get_cached_value(
 		"Company", company, ["disposal_account", "depreciation_cost_center"]
 	)
@@ -788,9 +789,9 @@ def get_disposal_account_and_cost_center(company):
 @frappe.whitelist()
 def get_value_after_depreciation_on_disposal_date(
 	asset: str,
-	disposal_date: str,
+	disposal_date: DateTimeLikeObject,
 	finance_book: str | None = None,
-) -> float:
+):
 	asset_doc = frappe.get_doc("Asset", asset)
 
 	if asset_doc.asset_type == "Composite Component":

erpnext/assets/doctype/asset_capitalization/asset_capitalization.js

@@ -396,7 +396,7 @@ erpnext.assets.AssetCapitalization = class AssetCapitalization extends erpnext.s
 				method: "erpnext.assets.doctype.asset_capitalization.asset_capitalization.get_warehouse_details",
 				child: item,
 				args: {
-					args: {
+					ctx: {
 						item_code: item.item_code,
 						warehouse: cstr(item.warehouse),
 						qty: -1 * flt(item.stock_qty),

erpnext/assets/doctype/asset_capitalization/asset_capitalization.py

@@ -2,6 +2,7 @@
 # For license information, please see license.txt
 
 import json
+from typing import Any
 
 import frappe
 
@@ -609,7 +610,7 @@ class AssetCapitalization(StockController):
 
 
 @frappe.whitelist()
-def get_target_item_details(item_code: str | None = None, company: str | None = None) -> frappe._dict:
+def get_target_item_details(item_code: str | None = None, company: str | None = None):
 	out = frappe._dict()
 
 	# Get Item Details
@@ -635,7 +636,7 @@ def get_target_item_details(item_code: str | None = None, company: str | None =
 
 
 @frappe.whitelist()
-def get_target_asset_details(asset: str | None = None, company: str | None = None) -> frappe._dict:
+def get_target_asset_details(asset: str | None = None, company: str | None = None):
 	out = frappe._dict()
 
 	# Get Asset Details
@@ -710,24 +711,22 @@ def get_consumed_stock_item_details(ctx: ItemDetailsCtx):
 
 
 @frappe.whitelist()
-def get_warehouse_details(args):
-	if isinstance(args, str):
-		args = json.loads(args)
-
-	args = frappe._dict(args)
-
-	out = {}
-	if args.warehouse and args.item_code:
-		out = {
-			"actual_qty": get_previous_sle(args).get("qty_after_transaction") or 0,
-			"valuation_rate": get_incoming_rate(args, raise_error_if_no_rate=False),
-		}
+@erpnext.normalize_ctx_input(ItemDetailsCtx)
+def get_warehouse_details(ctx: ItemDetailsCtx) -> frappe._dict:
+	out = frappe._dict()
+	if ctx.warehouse and ctx.item_code:
+		out = frappe._dict(
+			{
+				"actual_qty": get_previous_sle(ctx).get("qty_after_transaction") or 0,
+				"valuation_rate": get_incoming_rate(ctx, raise_error_if_no_rate=False),
+			}
+		)
 	return out
 
 
 @frappe.whitelist()
 @erpnext.normalize_ctx_input(ItemDetailsCtx)
-def get_consumed_asset_details(ctx):
+def get_consumed_asset_details(ctx: ItemDetailsCtx) -> frappe._dict:
 	out = frappe._dict()
 
 	asset_details = frappe._dict()
@@ -773,7 +772,7 @@ def get_consumed_asset_details(ctx):
 
 @frappe.whitelist()
 @erpnext.normalize_ctx_input(ItemDetailsCtx)
-def get_service_item_details(ctx):
+def get_service_item_details(ctx: ItemDetailsCtx) -> frappe._dict:
 	out = frappe._dict()
 
 	item = frappe._dict()
@@ -795,7 +794,7 @@ def get_service_item_details(ctx):
 
 
 @frappe.whitelist()
-def get_items_tagged_to_wip_composite_asset(params):
+def get_items_tagged_to_wip_composite_asset(params: dict | str):
 	if isinstance(params, str):
 		params = json.loads(params)
 

erpnext/assets/doctype/asset_depreciation_schedule/asset_depreciation_schedule.py

@@ -271,7 +271,7 @@ def get_asset_shift_factors_map():
 
 
 @frappe.whitelist()
-def get_depr_schedule(asset_name, status, finance_book=None):
+def get_depr_schedule(asset_name: str, status: str, finance_book: str | None = None):
 	asset_depr_schedule_doc = get_asset_depr_schedule_doc(asset_name, status, finance_book)
 
 	if not asset_depr_schedule_doc:
@@ -281,13 +281,13 @@ def get_depr_schedule(asset_name, status, finance_book=None):
 
 
 @frappe.whitelist()
-def get_asset_depr_schedule_doc(asset_name, status=None, finance_book=None):
+def get_asset_depr_schedule_doc(asset_name: str, status: str | None = None, finance_book: str | None = None):
 	asset_depr_schedule = get_asset_depr_schedule_name(asset_name, status, finance_book)
 
 	if not asset_depr_schedule:
 		return
 
-	asset_depr_schedule_doc = frappe.get_doc("Asset Depreciation Schedule", asset_depr_schedule[0].name)
+	asset_depr_schedule_doc = frappe.get_doc("Asset Depreciation Schedule", asset_depr_schedule)
 
 	return asset_depr_schedule_doc
 
@@ -299,21 +299,23 @@ def get_asset_depr_schedule_name(asset_name, status=None, finance_book=None):
 	]
 
 	if status:
-		if isinstance(status, str):
-			status = [status]
-		filters.append(["status", "in", status])
+		status_list = [status] if isinstance(status, str) else status
+		filters.append(["status", "in", status_list])
 
-	if finance_book:
-		filters.append(["finance_book", "=", finance_book])
-	else:
-		filters.append(["finance_book", "is", "not set"])
+	finance_book_filter = (
+		["finance_book", "=", finance_book] if finance_book else ["finance_book", "is", "not set"]
+	)
+	filters.append(finance_book_filter)
 
-	return frappe.get_all(
+	depreciation_schedules = frappe.get_all(
 		doctype="Asset Depreciation Schedule",
 		filters=filters,
+		fields=["name"],
 		limit=1,
 	)
 
+	return depreciation_schedules[0].name if depreciation_schedules else None
+
 
 def is_first_day_of_the_month(date):
 	first_day_of_the_month = get_first_day(date)

erpnext/assets/doctype/asset_maintenance/asset_maintenance.py

@@ -2,11 +2,13 @@
 # For license information, please see license.txt
 
 
+from typing import Any
+
 import frappe
 from frappe import _, throw
 from frappe.desk.form import assign_to
 from frappe.model.document import Document
-from frappe.utils import add_days, add_months, add_years, getdate, nowdate
+from frappe.utils import DateTimeLikeObject, add_days, add_months, add_years, getdate, nowdate
 
 
 class AssetMaintenance(Document):
@@ -90,7 +92,11 @@ def assign_tasks(asset_maintenance_name, assign_to_member, maintenance_task, nex
 
 @frappe.whitelist()
 def calculate_next_due_date(
-	periodicity, start_date=None, end_date=None, last_completion_date=None, next_due_date=None
+	periodicity: str,
+	start_date: DateTimeLikeObject | None = None,
+	end_date: DateTimeLikeObject | None = None,
+	last_completion_date: DateTimeLikeObject | None = None,
+	next_due_date: DateTimeLikeObject | None = None,
 ):
 	if not start_date and not last_completion_date:
 		start_date = frappe.utils.now()
@@ -164,19 +170,30 @@ def update_maintenance_log(asset_maintenance, item_code, item_name, task):
 
 @frappe.whitelist()
 @frappe.validate_and_sanitize_search_inputs
-def get_team_members(doctype, txt, searchfield, start, page_len, filters):
+def get_team_members(
+	doctype: str,
+	txt: str,
+	searchfield: str,
+	start: int,
+	page_len: int,
+	filters: dict[str, Any],
+) -> list[tuple[str]]:
 	return frappe.db.get_values(
-		"Maintenance Team Member", {"parent": filters.get("maintenance_team")}, "team_member"
+		"Maintenance Team Member",
+		{"parent": filters.get("maintenance_team")},
+		"team_member",
 	)
 
 
 @frappe.whitelist()
-def get_maintenance_log(asset_name):
+def get_maintenance_log(asset_name: str):
 	return frappe.db.sql(
 		"""
         select maintenance_status, count(asset_name) as count, asset_name
         from `tabAsset Maintenance Log`
-        where asset_name=%s group by maintenance_status""",
-		(asset_name),
+        where asset_name=%s
+        group by maintenance_status
+        """,
+		(asset_name,),
 		as_dict=1,
 	)

erpnext/assets/doctype/asset_repair/asset_repair.py

@@ -5,7 +5,7 @@ import frappe
 from frappe import _
 from frappe.query_builder import DocType
 from frappe.query_builder.functions import Sum
-from frappe.utils import cint, flt, get_link_to_form, getdate, time_diff_in_hours
+from frappe.utils import DateTimeLikeObject, cint, flt, get_link_to_form, getdate, time_diff_in_hours
 
 import erpnext
 from erpnext.accounts.doctype.accounting_dimension.accounting_dimension import (
@@ -448,14 +448,21 @@ class AssetRepair(AccountsController):
 
 
 @frappe.whitelist()
-def get_downtime(failure_date, completion_date):
+def get_downtime(failure_date: DateTimeLikeObject, completion_date: DateTimeLikeObject):
 	downtime = time_diff_in_hours(completion_date, failure_date)
 	return round(downtime, 2)
 
 
 @frappe.whitelist()
 @frappe.validate_and_sanitize_search_inputs
-def get_purchase_invoice(doctype, txt, searchfield, start, page_len, filters):
+def get_purchase_invoice(
+	doctype: str,
+	txt: str,
+	searchfield: str,
+	start: int,
+	page_len: int,
+	filters: dict,
+):
 	"""
 	Get Purchase Invoices that have expense accounts for non-stock items.
 	Only returns invoices with at least one non-stock, non-fixed-asset item with an expense account.
@@ -490,7 +497,14 @@ def get_purchase_invoice(doctype, txt, searchfield, start, page_len, filters):
 
 @frappe.whitelist()
 @frappe.validate_and_sanitize_search_inputs
-def get_expense_accounts(doctype, txt, searchfield, start, page_len, filters):
+def get_expense_accounts(
+	doctype: str,
+	txt: str,
+	searchfield: str,
+	start: int,
+	page_len: int,
+	filters: dict,
+):
 	"""
 	Get expense accounts for non-stock (service) items from the purchase invoice.
 	Used as a query function for link fields.
@@ -548,7 +562,7 @@ def _get_expense_accounts_for_purchase_invoice(purchase_invoice: str) -> list[st
 @frappe.whitelist()
 def get_unallocated_repair_cost(
 	purchase_invoice: str, expense_account: str, exclude_asset_repair: str | None = None
-) -> float:
+):
 	"""
 	Calculate the unused repair cost for a purchase invoice and expense account.
 	"""

erpnext/assets/doctype/asset_value_adjustment/asset_value_adjustment.py

@@ -227,6 +227,6 @@ class AssetValueAdjustment(Document):
 
 
 @frappe.whitelist()
-def get_value_of_accounting_dimensions(asset_name):
+def get_value_of_accounting_dimensions(asset_name: str):
 	dimension_fields = [*frappe.get_list("Accounting Dimension", pluck="fieldname"), "cost_center"]
 	return frappe.db.get_value("Asset", asset_name, fieldname=dimension_fields, as_dict=True)

erpnext/assets/doctype/location/location.py

@@ -211,7 +211,7 @@ def _ring_area(coords):
 
 
 @frappe.whitelist()
-def get_children(doctype, parent=None, location=None, is_root=False):
+def get_children(doctype: str, parent: str | None = None, location: str | None = None, is_root: bool = False):
 	if parent is None or parent == "All Locations":
 		parent = ""