refactor(assets): add type annotations in asset module

erpnext/assets/doctype/asset/asset.py

@@ -4,9 +4,11 @@
 
 import json
 import math
+from typing import Any
 
 import frappe
 from frappe import _
+from frappe.model.document import Document
 from frappe.query_builder.functions import IfNull, Sum
 from frappe.utils import (
 	cint,
@@ -1092,7 +1094,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 +1127,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,
+) -> Document:
 	asset_maintenance = frappe.new_doc("Asset Maintenance")
 	asset_maintenance.update(
 		{
@@ -1140,14 +1148,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,
+) -> Document:
 	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,
+) -> Document:
 	asset_capitalization = frappe.new_doc("Asset Capitalization")
 	asset_capitalization.update(
 		{
@@ -1161,35 +1178,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,
+) -> Document:
 	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,
+) -> list[dict[str, Any]]:
 	asset_category_doc = frappe.get_cached_doc("Asset Category", asset_category)
 	books = []
 	for d in asset_category_doc.finance_books:
@@ -1239,7 +1243,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) -> Document:
 	asset = frappe.get_doc("Asset", asset_name)
 	(
 		fixed_asset_account,
@@ -1281,7 +1285,10 @@ def make_journal_entry(asset_name):
 
 
 @frappe.whitelist()
-def make_asset_movement(assets, purpose=None):
+def make_asset_movement(
+	assets: list[dict[str, Any]],
+	purpose: str = "Transfer",
+) -> dict[str, Any]:
 	import json
 
 	if isinstance(assets, str):
@@ -1291,7 +1298,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 +1320,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,
+) -> float:
 	asset = frappe.get_doc("Asset", asset_name)
 	if not asset.calculate_depreciation:
 		return flt(asset.value_after_depreciation)
@@ -1322,7 +1332,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) -> bool:
 	active_capitalizations = frappe.db.count(
 		"Asset Capitalization", filters={"target_asset": asset, "docstatus": 1}
 	)
@@ -1330,7 +1340,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,
+) -> dict[str, Any]:
 	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 +1366,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) -> Document:
 	"""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

@@ -2,11 +2,15 @@
 # For license information, please see license.txt
 
 
+from typing import Any
+
 import frappe
 from frappe import _
+from frappe.model.document import Document
 from frappe.query_builder import Order
 from frappe.query_builder.functions import Max, Min
 from frappe.utils import (
+	DateTimeLikeObject,
 	add_months,
 	cint,
 	flt,
@@ -161,12 +165,12 @@ 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: dict[str, Any] | None = None,
+) -> Document:
 	frappe.has_permission("Journal Entry", throw=True)
 	date = date or today()
 
@@ -356,7 +360,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 +449,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 +776,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) -> tuple[str, str]:
 	disposal_account, depreciation_cost_center = frappe.get_cached_value(
 		"Company", company, ["disposal_account", "depreciation_cost_center"]
 	)

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
 
@@ -710,24 +711,26 @@ def get_consumed_stock_item_details(ctx: ItemDetailsCtx):
 
 
 @frappe.whitelist()
-def get_warehouse_details(args):
+def get_warehouse_details(args: dict[str, Any]) -> frappe._dict:
 	if isinstance(args, str):
 		args = json.loads(args)
 
 	args = frappe._dict(args)
 
-	out = {}
+	out = frappe._dict()
 	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),
-		}
+		out = frappe._dict(
+			{
+				"actual_qty": get_previous_sle(args).get("qty_after_transaction") or 0,
+				"valuation_rate": get_incoming_rate(args, 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 +776,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 +798,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, Any]):
 	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,7 +281,7 @@ 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:

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,8 +92,12 @@ 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,
+) -> str:
 	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) -> list[dict[str, Any]]:
 	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) -> float:
 	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[str, str],
+) -> list[list[str]]:
 	"""
 	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[str, str],
+) -> list[list[str]]:
 	"""
 	Get expense accounts for non-stock (service) items from the purchase invoice.
 	Used as a query function for link fields.

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) -> dict:
 	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,9 @@ 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
+) -> list[dict]:
 	if parent is None or parent == "All Locations":
 		parent = ""