From 84773a3a32a28d2065985dfa014226d1814e697b Mon Sep 17 00:00:00 2001 From: khushi8112 Date: Tue, 17 Feb 2026 15:56:24 +0530 Subject: [PATCH] refactor(assets): add type annotations in asset module --- erpnext/assets/doctype/asset/asset.py | 82 +++++++++++-------- erpnext/assets/doctype/asset/depreciation.py | 22 +++-- .../asset_capitalization.py | 21 +++-- .../asset_depreciation_schedule.py | 4 +- .../asset_maintenance/asset_maintenance.py | 33 ++++++-- .../doctype/asset_repair/asset_repair.py | 22 ++++- .../asset_value_adjustment.py | 2 +- erpnext/assets/doctype/location/location.py | 4 +- 8 files changed, 122 insertions(+), 68 deletions(-) diff --git a/erpnext/assets/doctype/asset/asset.py b/erpnext/assets/doctype/asset/asset.py index 7190db400d9..f5999d7fc72 100644 --- a/erpnext/assets/doctype/asset/asset.py +++ b/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("{0}") - .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) diff --git a/erpnext/assets/doctype/asset/depreciation.py b/erpnext/assets/doctype/asset/depreciation.py index 832e3736e7d..cdc4462f673 100644 --- a/erpnext/assets/doctype/asset/depreciation.py +++ b/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"] ) diff --git a/erpnext/assets/doctype/asset_capitalization/asset_capitalization.py b/erpnext/assets/doctype/asset_capitalization/asset_capitalization.py index 14353bc75cb..85e7ae613a3 100644 --- a/erpnext/assets/doctype/asset_capitalization/asset_capitalization.py +++ b/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) diff --git a/erpnext/assets/doctype/asset_depreciation_schedule/asset_depreciation_schedule.py b/erpnext/assets/doctype/asset_depreciation_schedule/asset_depreciation_schedule.py index b258fa11301..5b7a32a6cf8 100644 --- a/erpnext/assets/doctype/asset_depreciation_schedule/asset_depreciation_schedule.py +++ b/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: diff --git a/erpnext/assets/doctype/asset_maintenance/asset_maintenance.py b/erpnext/assets/doctype/asset_maintenance/asset_maintenance.py index 0103360b17a..350caccc30e 100644 --- a/erpnext/assets/doctype/asset_maintenance/asset_maintenance.py +++ b/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, ) diff --git a/erpnext/assets/doctype/asset_repair/asset_repair.py b/erpnext/assets/doctype/asset_repair/asset_repair.py index 476b0187bf1..b39c770b034 100644 --- a/erpnext/assets/doctype/asset_repair/asset_repair.py +++ b/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. diff --git a/erpnext/assets/doctype/asset_value_adjustment/asset_value_adjustment.py b/erpnext/assets/doctype/asset_value_adjustment/asset_value_adjustment.py index c033cda05b5..2f71456207a 100644 --- a/erpnext/assets/doctype/asset_value_adjustment/asset_value_adjustment.py +++ b/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) diff --git a/erpnext/assets/doctype/location/location.py b/erpnext/assets/doctype/location/location.py index 03d0980b03f..dbffa4ce923 100644 --- a/erpnext/assets/doctype/location/location.py +++ b/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 = ""