feat: Skip locked rows while selecting (#24298)

frappe/database/database.py

@@ -437,6 +437,7 @@ class Database:
 		run=True,
 		pluck=False,
 		distinct=False,
+		skip_locked=False,
 	):
 		"""Return a document property or list of properties.
 
@@ -447,6 +448,10 @@ class Database:
 		:param as_dict: Return values as dict.
 		:param debug: Print query in error log.
 		:param order_by: Column to order by
+		:param cache: Use cached results fetched during current job/request
+		:param pluck: pluck first column instead of returning as nested list or dict.
+		:param for_update: All the affected/read rows will be locked.
+		:param skip_locked: Skip selecting currently locked rows.
 
 		Example:
 
@@ -477,6 +482,7 @@ class Database:
 			pluck=pluck,
 			distinct=distinct,
 			limit=1,
+			skip_locked=skip_locked,
 		)
 
 		if not run:
@@ -509,6 +515,7 @@ class Database:
 		pluck=False,
 		distinct=False,
 		limit=None,
+		skip_locked=False,
 	):
 		"""Return multiple document properties.
 
@@ -548,6 +555,8 @@ class Database:
 				distinct=distinct,
 				limit=limit,
 				as_dict=as_dict,
+				skip_locked=skip_locked,
+				for_update=for_update,
 			)
 
 		else:
@@ -568,11 +577,12 @@ class Database:
 						debug=debug,
 						order_by=order_by,
 						update=update,
-						for_update=for_update,
 						run=run,
 						pluck=pluck,
 						distinct=distinct,
 						limit=limit,
+						for_update=for_update,
+						skip_locked=skip_locked,
 					)
 				except Exception as e:
 					if ignore and (frappe.db.is_missing_column(e) or frappe.db.is_table_missing(e)):
@@ -806,6 +816,7 @@ class Database:
 		order_by=None,
 		update=None,
 		for_update=False,
+		skip_locked=False,
 		run=True,
 		pluck=False,
 		distinct=False,
@@ -816,6 +827,7 @@ class Database:
 			filters=filters,
 			order_by=order_by,
 			for_update=for_update,
+			skip_locked=skip_locked,
 			fields=fields,
 			distinct=distinct,
 			limit=limit,
@@ -839,6 +851,8 @@ class Database:
 		distinct=False,
 		limit=None,
 		as_dict=False,
+		for_update=False,
+		skip_locked=False,
 	):
 		if names := list(filter(None, names)):
 			return frappe.qb.get_query(
@@ -849,6 +863,8 @@ class Database:
 				distinct=distinct,
 				limit=limit,
 				validate_filters=True,
+				for_update=for_update,
+				skip_locked=skip_locked,
 			).run(debug=debug, run=run, as_dict=as_dict, pluck=pluck)
 		return {}
 

frappe/database/query.py

@@ -47,6 +47,7 @@ class Engine:
 		delete: bool = False,
 		*,
 		validate_filters: bool = False,
+		skip_locked: bool = False,
 	) -> QueryBuilder:
 		self.is_mariadb = frappe.db.db_type == "mariadb"
 		self.is_postgres = frappe.db.db_type == "postgres"
@@ -83,7 +84,7 @@ class Engine:
 			self.query = self.query.distinct()
 
 		if for_update:
-			self.query = self.query.for_update()
+			self.query = self.query.for_update(skip_locked=skip_locked)
 
 		if group_by:
 			self.query = self.query.groupby(group_by)

frappe/tests/test_db.py

@@ -57,6 +57,17 @@ class TestDB(FrappeTestCase):
 			frappe.db.rollback(save_point=savepoint)
 			self.fail("Long running queries not timing out")
 
+	def test_skip_locking(self):
+		first_conn = frappe.local.db
+		name = frappe.db.get_value("User", "Administrator", "name", for_update=True, skip_locked=True)
+		self.assertEqual(name, "Administrator")
+
+		frappe.connect()  # Create a 2nd connection
+		second_conn = frappe.local.db
+		self.assertIsNot(first_conn, second_conn)
+		name = frappe.db.get_value("User", "Administrator", "name", for_update=True, skip_locked=True)
+		self.assertFalse(name)
+
 	@patch.dict(frappe.conf, {"http_timeout": 20, "enable_db_statement_timeout": 1})
 	def test_db_timeout_computation(self):
 		set_request(method="GET", path="/")