refactor: rewrite queries with frappe.qb + refactor type hints

erpnext/hr/doctype/leave_allocation/leave_allocation.py

@@ -296,18 +296,24 @@ class LeaveAllocation(Document):
 
 def get_previous_allocation(from_date, leave_type, employee):
 	"""Returns document properties of previous allocation"""
-	return frappe.db.get_value(
-		"Leave Allocation",
-		filters={
-			"to_date": ("<", from_date),
-			"leave_type": leave_type,
-			"employee": employee,
-			"docstatus": 1,
-		},
-		order_by="to_date DESC",
-		fieldname=["name", "from_date", "to_date", "employee", "leave_type"],
-		as_dict=1,
-	)
+	Allocation = frappe.qb.DocType("Leave Allocation")
+	return (
+		frappe.qb.from_(Allocation)
+		.select(
+			Allocation.name,
+			Allocation.from_date,
+			Allocation.to_date,
+			Allocation.employee,
+			Allocation.leave_type,
+		)
+		.where(
+			(Allocation.employee == employee)
+			& (Allocation.leave_type == leave_type)
+			& (Allocation.to_date < from_date)
+			& (Allocation.docstatus == 1)
+		)
+		.orderby(Allocation.to_date, order=frappe.qb.desc)
+	).run(as_dict=True)
 
 
 def get_leave_allocation_for_period(

erpnext/hr/doctype/leave_application/leave_application.py

@@ -706,20 +706,22 @@ def get_allocation_expiry_for_cf_leaves(
 	employee: str, leave_type: str, to_date: str, from_date: str
 ) -> str:
 	"""Returns expiry of carry forward allocation in leave ledger entry"""
-	expiry = frappe.get_all(
-		"Leave Ledger Entry",
-		filters={
-			"employee": employee,
-			"leave_type": leave_type,
-			"is_carry_forward": 1,
-			"transaction_type": "Leave Allocation",
-			"to_date": ["between", (from_date, to_date)],
-			"docstatus": 1,
-		},
-		fields=["to_date"],
-		limit=1,
-	)
-	return expiry[0]["to_date"] if expiry else ""
+	Ledger = frappe.qb.DocType("Leave Ledger Entry")
+	expiry = (
+		frappe.qb.from_(Ledger)
+		.select(Ledger.to_date)
+		.where(
+			(Ledger.employee == employee)
+			& (Ledger.leave_type == leave_type)
+			& (Ledger.is_carry_forward == 1)
+			& (Ledger.transaction_type == "Leave Allocation")
+			& (Ledger.to_date.between(from_date, to_date))
+			& (Ledger.docstatus == 1)
+		)
+		.limit(1)
+	).run()
+
+	return expiry[0][0] if expiry else ""
 
 
 @frappe.whitelist()
@@ -1018,7 +1020,7 @@ def get_leaves_for_period(
 			if leave_entry.leaves % 1:
 				half_day = 1
 				half_day_date = frappe.db.get_value(
-					"Leave Application", {"name": leave_entry.transaction_name}, "half_day_date"
+					"Leave Application", leave_entry.transaction_name, "half_day_date"
 				)
 
 			leave_days += (

erpnext/hr/report/employee_leave_balance/employee_leave_balance.py

@@ -3,7 +3,6 @@
 
 
 from itertools import groupby
-from typing import Dict, List, Optional, Tuple
 
 import frappe
 from frappe import _
@@ -18,7 +17,7 @@ from erpnext.hr.doctype.leave_application.leave_application import (
 Filters = frappe._dict
 
 
-def execute(filters: Optional[Filters] = None) -> Tuple:
+def execute(filters: Filters | None = None) -> tuple:
 	if filters.to_date <= filters.from_date:
 		frappe.throw(_('"From Date" can not be greater than or equal to "To Date"'))
 
@@ -28,7 +27,7 @@ def execute(filters: Optional[Filters] = None) -> Tuple:
 	return columns, data, None, charts
 
 
-def get_columns() -> List[Dict]:
+def get_columns() -> list[dict]:
 	return [
 		{
 			"label": _("Leave Type"),
@@ -84,8 +83,8 @@ def get_columns() -> List[Dict]:
 	]
 
 
-def get_data(filters: Filters) -> List:
-	leave_types = frappe.get_all("Leave Type", pluck="name", order_by="name")
+def get_data(filters: Filters) -> list:
+	leave_types = get_leave_types()
 	active_employees = get_employees(filters)
 
 	precision = cint(frappe.db.get_single_value("System Settings", "float_precision"))
@@ -131,6 +130,13 @@ def get_data(filters: Filters) -> List:
 	return data
 
 
+def get_leave_types() -> list[str]:
+	LeaveType = frappe.qb.DocType("Leave Type")
+	return (frappe.qb.from_(LeaveType).select(LeaveType.name).orderby(LeaveType.name)).run(
+		pluck="name"
+	)
+
+
 def get_employees(filters: Filters) -> list[dict]:
 	Employee = frappe.qb.DocType("Employee")
 	query = frappe.qb.from_(Employee).select(
@@ -178,7 +184,7 @@ def get_opening_balance(
 
 def get_allocated_and_expired_leaves(
 	from_date: str, to_date: str, employee: str, leave_type: str
-) -> Tuple[float, float, float]:
+) -> tuple[float, float, float]:
 	new_allocation = 0
 	expired_leaves = 0
 	carry_forwarded_leaves = 0
@@ -208,9 +214,9 @@ def get_allocated_and_expired_leaves(
 
 def get_leave_ledger_entries(
 	from_date: str, to_date: str, employee: str, leave_type: str
-) -> List[Dict]:
+) -> list[dict]:
 	ledger = frappe.qb.DocType("Leave Ledger Entry")
-	records = (
+	return (
 		frappe.qb.from_(ledger)
 		.select(
 			ledger.employee,
@@ -236,10 +242,8 @@ def get_leave_ledger_entries(
 		)
 	).run(as_dict=True)
 
-	return records
 
-
-def get_chart_data(data: List, filters: Filters) -> Dict:
+def get_chart_data(data: list, filters: Filters) -> dict:
 	labels = []
 	datasets = []
 	employee_data = data
@@ -259,7 +263,7 @@ def get_chart_data(data: List, filters: Filters) -> Dict:
 	return chart
 
 
-def get_dataset_for_chart(employee_data: List, datasets: List, labels: List) -> List:
+def get_dataset_for_chart(employee_data: list, datasets: list, labels: list) -> list:
 	leaves = []
 	employee_data = sorted(employee_data, key=lambda k: k["employee_name"])