refactor: overlapping shifts validation

- convert raw query to frappe.qb

- check for overlapping timings if dates overlap

- translation friendly error messages with link to overlapping doc

erpnext/hr/doctype/shift_assignment/shift_assignment.py

@@ -7,79 +7,95 @@ from datetime import datetime, timedelta
 import frappe
 from frappe import _
 from frappe.model.document import Document
-from frappe.utils import cstr, getdate, now_datetime, nowdate
+from frappe.query_builder import Criterion
+from frappe.utils import cstr, get_link_to_form, getdate, now_datetime, nowdate
 
 from erpnext.hr.doctype.employee.employee import get_holiday_list_for_employee
 from erpnext.hr.doctype.holiday_list.holiday_list import is_holiday
 from erpnext.hr.utils import validate_active_employee
 
+class OverlappingShiftError(frappe.ValidationError):
+	pass
 
 class ShiftAssignment(Document):
 	def validate(self):
 		validate_active_employee(self.employee)
-		self.validate_overlapping_dates()
+		self.validate_overlapping_shifts()
 
 		if self.end_date:
 			self.validate_from_to_dates("start_date", "end_date")
 
-	def validate_overlapping_dates(self):
+	def validate_overlapping_shifts(self):
+		overlapping_dates = self.get_overlapping_dates()
+		if len(overlapping_dates):
+			# if dates are overlapping, check if timings are overlapping, else allow
+			overlapping_timings = self.has_overlapping_timings(overlapping_dates[0].shift_type)
+			if overlapping_timings:
+				self.throw_overlap_error(overlapping_dates[0])
+
+	def get_overlapping_dates(self):
 		if not self.name:
 			self.name = "New Shift Assignment"
 
-		condition = """and (
+		shift = frappe.qb.DocType("Shift Assignment")
-				end_date is null
+		query = (
-				or
+			frappe.qb.from_(shift)
-					%(start_date)s between start_date and end_date
+				.select(shift.name, shift.shift_type, shift.start_date, shift.end_date, shift.docstatus, shift.status)
-		"""
+				.where(
-
+					(shift.employee == self.employee)
-		if self.end_date:
+					& (shift.docstatus == 1)
-			condition += """ or
+					& (shift.name != self.name)
-					%(end_date)s between start_date and end_date
+					& (shift.status == "Active")
-					or
+				)
-					start_date between %(start_date)s and %(end_date)s
-				) """
-		else:
-			condition += """ ) """
-
-		assigned_shifts = frappe.db.sql(
-			"""
-			select name, shift_type, start_date ,end_date, docstatus, status
-			from `tabShift Assignment`
-			where
-				employee=%(employee)s and docstatus = 1
-				and name != %(name)s
-				and status = "Active"
-				{0}
-		""".format(
-				condition
-			),
-			{
-				"employee": self.employee,
-				"shift_type": self.shift_type,
-				"start_date": self.start_date,
-				"end_date": self.end_date,
-				"name": self.name,
-			},
-			as_dict=1,
 		)
 
-		if len(assigned_shifts):
+		if self.end_date:
-			self.throw_overlap_error(assigned_shifts[0])
+			query = query.where(
+				Criterion.any([
+					Criterion.any([
+						shift.end_date.isnull(),
+						((self.start_date >= shift.start_date) & (self.start_date <= shift.end_date))
+					]),
+					Criterion.any([
+						((self.end_date >= shift.start_date) & (self.end_date <= shift.end_date)),
+						shift.start_date.between(self.start_date, self.end_date)
+					])
+				])
+			)
+		else:
+			query = query.where(
+				shift.end_date.isnull()
+				| ((self.start_date >= shift.start_date) & (self.start_date <= shift.end_date))
+			)
+
+		return query.run(as_dict=True)
+
+	def has_overlapping_timings(self, overlapping_shift):
+		curr_shift = frappe.db.get_value("Shift Type", self.shift_type, ["start_time", "end_time"], as_dict=True)
+		overlapping_shift = frappe.db.get_value("Shift Type", overlapping_shift, ["start_time", "end_time"], as_dict=True)
+
+		if ((curr_shift.start_time > overlapping_shift.start_time and curr_shift.start_time < overlapping_shift.end_time) or
+			(curr_shift.end_time > overlapping_shift.start_time and curr_shift.end_time < overlapping_shift.end_time) or
+			(curr_shift.start_time <= overlapping_shift.start_time and curr_shift.end_time >= overlapping_shift.end_time)):
+			return True
+		return False
 
 	def throw_overlap_error(self, shift_details):
 		shift_details = frappe._dict(shift_details)
+		msg = None
 		if shift_details.docstatus == 1 and shift_details.status == "Active":
-			msg = _("Employee {0} already has Active Shift {1}: {2}").format(
+			if shift_details.start_date and shift_details.end_date:
-				frappe.bold(self.employee), frappe.bold(self.shift_type), frappe.bold(shift_details.name)
+				msg = _("Employee {0} already has an active Shift {1}: {2} from {3} to {4}").format(frappe.bold(self.employee), frappe.bold(self.shift_type),
-			)
+					get_link_to_form("Shift Assignment", shift_details.name),
-		if shift_details.start_date:
+					getdate(self.start_date).strftime("%d-%m-%Y"),
-			msg += _(" from {0}").format(getdate(self.start_date).strftime("%d-%m-%Y"))
+					getdate(self.end_date).strftime("%d-%m-%Y"))
-			title = "Ongoing Shift"
+			else:
-			if shift_details.end_date:
+				msg = _("Employee {0} already has an active Shift {1}: {2} from {3}").format(frappe.bold(self.employee), frappe.bold(self.shift_type),
-				msg += _(" to {0}").format(getdate(self.end_date).strftime("%d-%m-%Y"))
+					get_link_to_form("Shift Assignment", shift_details.name),
-				title = "Active Shift"
+					getdate(self.start_date).strftime("%d-%m-%Y"))
+
 		if msg:
-			frappe.throw(msg, title=title)
+			frappe.throw(msg, title=_("Overlapping Shifts"), exc=OverlappingShiftError)
 
 
 @frappe.whitelist()