Merge pull request #27955 from blaggacao/feat/add-read-only-document-context

feat: add read only document mode

frappe/exceptions.py

@@ -11,6 +11,12 @@ class SiteNotSpecifiedError(Exception):
 		super(Exception, self).__init__(self.message)
 
 
+class DatabaseModificationError(Exception):
+	"""Error raised when attempting to modify the database in a read-only document context."""
+
+	pass
+
+
 class UrlSchemeNotSupported(Exception):
 	pass
 

frappe/model/document.py

@@ -4,7 +4,9 @@ import hashlib
 import json
 import time
 from collections.abc import Generator, Iterable
-from typing import TYPE_CHECKING, Any, Optional
+from contextlib import contextmanager
+from functools import wraps
+from typing import TYPE_CHECKING, Any, Literal, Optional, TypeAlias, Union, overload
 
 from werkzeug.exceptions import NotFound
 
@@ -89,6 +91,61 @@ def get_doc(*args, **kwargs):
 	raise ImportError(doctype)
 
 
+@contextmanager
+def read_only_document(context=None):
+	# Store original methods
+	original_methods = {
+		"save": Document.save,
+		"_save": Document._save,
+		"insert": Document.insert,
+		"delete": Document.delete,
+		"submit": Document.submit,
+		"cancel": Document.cancel,
+		"db_set": Document.db_set,
+	}
+
+	def read_only_method(func):
+		@wraps(func)
+		def wrapper(self, *args, **kwargs):
+			if self.doctype == "Error Log" and func.__name__ == "insert":
+				return original_methods["insert"](self, *args, **kwargs)
+			error_msg = f"Cannot call {func.__name__} in read-only document mode"
+			if context:
+				error_msg += f" ({context})"
+			raise frappe.DatabaseModificationError(error_msg)
+
+		return wrapper
+
+	# Use a thread-local variable to track nested invocations
+	if not hasattr(frappe.local, "read_only_depth"):
+		frappe.local.read_only_depth = 0
+
+	try:
+		# Increment the depth counter
+		frappe.local.read_only_depth += 1
+
+		# Only apply read-only methods if this is the outermost invocation
+		if frappe.local.read_only_depth == 1:
+			# Replace methods with read-only versions
+			for method_name, method in original_methods.items():
+				setattr(Document, method_name, read_only_method(method))
+
+		yield
+
+	finally:
+		# Decrement the depth counter
+		frappe.local.read_only_depth -= 1
+
+		# Only restore original methods if this is the outermost invocation
+		if frappe.local.read_only_depth == 0:
+			# Restore original methods
+			for method_name, method in original_methods.items():
+				setattr(Document, method_name, method)
+
+			# Clean up the thread-local variable
+			del frappe.local.read_only_depth
+
+
 class Document(BaseDocument):
 	"""All controllers inherit from `Document`."""
 

frappe/tests/test_document_ro_mode.py

@@ -0,0 +1,191 @@
+import unittest
+from contextlib import contextmanager
+
+import frappe
+from frappe.model.document import Document, read_only_document
+from frappe.tests.utils import FrappeTestCase
+
+
+class TestReadOnlyDocument(FrappeTestCase):
+	def setUp(self):
+		# Create a test document
+		self.test_doc = frappe.get_doc({"doctype": "ToDo", "description": "Test ToDo"})
+		self.test_doc.insert()
+
+	def tearDown(self):
+		# Delete the test document
+		frappe.delete_doc("ToDo", self.test_doc.name)
+
+	def test_read_only_save(self):
+		with read_only_document():
+			with self.assertRaises(frappe.DatabaseModificationError):
+				self.test_doc.save()
+
+	def test_read_only_insert(self):
+		with read_only_document():
+			with self.assertRaises(frappe.DatabaseModificationError):
+				frappe.get_doc({"doctype": "ToDo", "description": "Another Test ToDo"}).insert()
+
+	def test_read_only_delete(self):
+		with read_only_document():
+			with self.assertRaises(frappe.DatabaseModificationError):
+				self.test_doc.delete()
+
+	def test_read_only_submit(self):
+		with read_only_document():
+			with self.assertRaises(frappe.DatabaseModificationError):
+				self.test_doc.submit()
+
+	def test_read_only_cancel(self):
+		with read_only_document():
+			with self.assertRaises(frappe.DatabaseModificationError):
+				self.test_doc.cancel()
+
+	def test_read_only_db_set(self):
+		with read_only_document():
+			with self.assertRaises(frappe.DatabaseModificationError):
+				self.test_doc.db_set("status", "Closed")
+
+	def test_read_only_nested_calls(self):
+		def nested_save():
+			self.test_doc.save()
+
+		with read_only_document():
+			with self.assertRaises(frappe.DatabaseModificationError):
+				nested_save()
+
+	def test_read_only_context_manager_restoration(self):
+		original_save = Document.save
+
+		with read_only_document():
+			self.assertNotEqual(Document.save, original_save)
+
+		self.assertEqual(Document.save, original_save)
+
+	def test_nested_read_only_document(self):
+		# Check that read_only_depth is not set initially
+		self.assertFalse(hasattr(frappe.local, "read_only_depth"))
+
+		with read_only_document():
+			# Check that read_only_depth is set to 1
+			self.assertEqual(frappe.local.read_only_depth, 1)
+
+			# Attempt to modify the document
+			with self.assertRaises(frappe.DatabaseModificationError):
+				self.test_doc.description = "Modified in outer context"
+				self.test_doc.save()
+
+			with read_only_document():
+				# Check that read_only_depth is incremented to 2
+				self.assertEqual(frappe.local.read_only_depth, 2)
+
+				# Attempt to modify the document in nested context
+				with self.assertRaises(frappe.DatabaseModificationError):
+					self.test_doc.description = "Modified in inner context"
+					self.test_doc.save()
+
+			# Check that read_only_depth is back to 1 after nested context
+			self.assertEqual(frappe.local.read_only_depth, 1)
+
+			# Attempt to modify the document again
+			with self.assertRaises(frappe.DatabaseModificationError):
+				self.test_doc.description = "Modified after inner context"
+				self.test_doc.save()
+
+		# Check that read_only_depth is removed after all contexts are closed
+		self.assertFalse(hasattr(frappe.local, "read_only_depth"))
+
+		# Verify that document can be modified outside read_only_document
+		self.test_doc.description = "Modified outside read_only_document"
+		self.test_doc.save()
+		self.assertEqual(self.test_doc.description, "Modified outside read_only_document")
+
+	def test_error_log_exception_in_read_only(self):
+		with read_only_document():
+			# Attempt to insert an Error Log
+			error_log = frappe.get_doc({"doctype": "Error Log", "error": "Test error in read-only mode"})
+
+			# This should not raise an exception
+			error_log.insert()
+
+			# Verify that the Error Log was inserted
+			self.assertTrue(error_log.name)
+
+			# Attempt to modify a different document
+			with self.assertRaises(frappe.DatabaseModificationError):
+				self.test_doc.description = "Modified in read-only mode"
+				self.test_doc.save()
+
+		# Clean up the inserted Error Log
+		frappe.delete_doc("Error Log", error_log.name)
+
+	def test_read_only_multiple_documents(self):
+		doc1 = frappe.get_doc({"doctype": "ToDo", "description": "Test ToDo 1"})
+		doc2 = frappe.get_doc({"doctype": "ToDo", "description": "Test ToDo 2"})
+
+		with read_only_document():
+			with self.assertRaises(frappe.DatabaseModificationError):
+				doc1.insert()
+			with self.assertRaises(frappe.DatabaseModificationError):
+				doc2.insert()
+
+	def test_read_only_custom_method(self):
+		class CustomDocument(Document):
+			def custom_save(self):
+				self.save()
+
+		custom_doc = CustomDocument({"doctype": "ToDo", "description": "Custom Test ToDo"})
+
+		with read_only_document():
+			with self.assertRaises(frappe.DatabaseModificationError):
+				custom_doc.custom_save()
+
+	def test_read_only_exception_handling(self):
+		@contextmanager
+		def exception_raiser():
+			raise Exception("Test exception")
+			yield
+
+		try:
+			with read_only_document(), exception_raiser():
+				pass
+		except Exception:
+			pass
+
+		# Ensure methods are restored even if an exception occurs
+		self.assertEqual(Document.save, self.test_doc.__class__.save)
+
+	def test_read_only_nested_context_managers(self):
+		original_save = Document.save
+
+		with read_only_document():
+			self.assertNotEqual(Document.save, original_save)
+
+			with read_only_document():
+				self.assertNotEqual(Document.save, original_save)
+
+			self.assertNotEqual(Document.save, original_save)
+
+		self.assertEqual(Document.save, original_save)
+
+	def test_read_only_method_call_details(self):
+		with read_only_document():
+			with self.assertRaises(frappe.DatabaseModificationError) as cm:
+				self.test_doc.save()
+
+			self.assertIn("Cannot call save in read-only document mode", str(cm.exception))
+
+	def test_read_only_does_not_affect_reads(self):
+		with read_only_document():
+			# These operations should not raise exceptions
+			doc = frappe.get_doc("ToDo", self.test_doc.name)
+			self.assertEqual(doc.description, "Test ToDo")
+
+			docs = frappe.get_all("ToDo", filters={"name": self.test_doc.name})
+			self.assertEqual(len(docs), 1)
+
+	def test_read_only_with_new_document_instance(self):
+		with read_only_document():
+			new_doc = frappe.new_doc("ToDo")
+			with self.assertRaises(frappe.DatabaseModificationError):
+				new_doc.insert()