feat: wrap read only mode SQL errors

frappe/database/database.py

@@ -221,6 +221,15 @@ class Database:
 			elif self.is_timedout(e):
 				raise frappe.QueryTimeoutError(e) from e
 
+			elif self.is_read_only_mode_error(e):
+				frappe.throw(
+					_(
+						"Site is running in read only mode, this action can not be performed right now. Please try again later."
+					),
+					title=_("In Read Only Mode"),
+					exc=frappe.InReadOnlyMode,
+				)
+
 			# TODO: added temporarily
 			elif self.db_type == "postgres":
 				traceback.print_stack()

frappe/database/mariadb/database.py

@@ -32,6 +32,10 @@ class MariaDBExceptionUtil:
 	def is_timedout(e: pymysql.Error) -> bool:
 		return e.args[0] == ER.LOCK_WAIT_TIMEOUT
 
+	@staticmethod
+	def is_read_only_mode_error(e: pymysql.Error) -> bool:
+		return e.args[0] == 1792
+
 	@staticmethod
 	def is_table_missing(e: pymysql.Error) -> bool:
 		return e.args[0] == ER.NO_SUCH_TABLE

frappe/database/postgres/database.py

@@ -12,7 +12,7 @@ from psycopg2.errorcodes import (
 	UNDEFINED_TABLE,
 	UNIQUE_VIOLATION,
 )
-from psycopg2.errors import SequenceGeneratorLimitExceeded, SyntaxError
+from psycopg2.errors import ReadOnlySqlTransaction, SequenceGeneratorLimitExceeded, SyntaxError
 from psycopg2.extensions import ISOLATION_LEVEL_REPEATABLE_READ
 
 import frappe
@@ -55,6 +55,10 @@ class PostgresExceptionUtil:
 		# http://initd.org/psycopg/docs/extensions.html?highlight=datatype#psycopg2.extensions.QueryCanceledError
 		return isinstance(e, psycopg2.extensions.QueryCanceledError)
 
+	@staticmethod
+	def is_read_only_mode_error(e) -> bool:
+		return isinstance(e, ReadOnlySqlTransaction)
+
 	@staticmethod
 	def is_syntax_error(e):
 		return isinstance(e, SyntaxError)

frappe/exceptions.py

@@ -236,6 +236,10 @@ class QueryDeadlockError(Exception):
 	pass
 
 
+class InReadOnlyMode(ValidationError):
+	http_status_code = 503  # temporarily not available
+
+
 class TooManyWritesError(Exception):
 	pass
 

frappe/tests/test_db.py

@@ -460,6 +460,14 @@ class TestDB(FrappeTestCase):
 			# recover transaction to continue other tests
 			raise Exception
 
+	def test_read_only_errors(self):
+		frappe.db.rollback()
+		frappe.db.begin(read_only=True)
+		self.addCleanup(frappe.db.rollback)
+
+		with self.assertRaises(frappe.InReadOnlyMode):
+			frappe.db.set_value("User", "Administrator", "full_name", "Haxor")
+
 	def test_exists(self):
 		dt, dn = "User", "Administrator"
 		self.assertEqual(frappe.db.exists(dt, dn, cache=True), dn)