Merge pull request #16727 from ankush/pg_transaction_management

fix: explicitly start a new transaction

frappe/core/doctype/scheduled_job_type/scheduled_job_type.py

@@ -185,9 +185,12 @@ def insert_single_event(frequency: str, event: str, cron_format: str = None):
 	if not frappe.db.exists(
 		"Scheduled Job Type", {"method": event, "frequency": frequency, **cron_expr}
 	):
+		savepoint = "scheduled_job_type_creation"
 		try:
+			frappe.db.savepoint(savepoint)
 			doc.insert()
 		except frappe.DuplicateEntryError:
+			frappe.db.rollback(save_point=savepoint)
 			doc.delete()
 			doc.insert()
 

frappe/custom/doctype/custom_field/custom_field.py

@@ -161,6 +161,7 @@ def create_custom_field(doctype, df, ignore_validate=False, is_system_generated=
 		custom_field.update(df)
 		custom_field.flags.ignore_validate = ignore_validate
 		custom_field.insert()
+		return custom_field
 
 
 def create_custom_fields(custom_fields, ignore_validate=False, update=True):

frappe/database/database.py

@@ -201,6 +201,9 @@ class Database(object):
 
 			elif frappe.conf.db_type == "postgres":
 				# TODO: added temporarily
+				import traceback
+
+				traceback.print_stack()
 				print(e)
 				raise
 
@@ -920,6 +923,9 @@ class Database(object):
 			frappe.call(method[0], *(method[1] or []), **(method[2] or {}))
 
 		self.sql("commit")
+		if frappe.conf.db_type == "postgres":
+			# Postgres requires explicitly starting new transaction
+			self.begin()
 
 		frappe.local.rollback_observers = []
 		self.flush_realtime_log()

frappe/database/postgres/setup_db.py

@@ -6,6 +6,7 @@ import frappe
 def setup_database(force, source_sql=None, verbose=False):
 	root_conn = get_root_connection(frappe.flags.root_login, frappe.flags.root_password)
 	root_conn.commit()
+	root_conn.sql("end")
 	root_conn.sql("DROP DATABASE IF EXISTS `{0}`".format(frappe.conf.db_name))
 	root_conn.sql("DROP USER IF EXISTS {0}".format(frappe.conf.db_name))
 	root_conn.sql("CREATE DATABASE `{0}`".format(frappe.conf.db_name))

frappe/desk/page/setup_wizard/setup_wizard.py

@@ -431,22 +431,12 @@ def make_records(records, debug=False):
 		if doc.meta.get_field(parent_link_field) and not doc.get(parent_link_field):
 			doc.flags.ignore_mandatory = True
 
+		savepoint = "setup_fixtures_creation"
 		try:
-			doc.insert(ignore_permissions=True)
-			frappe.db.commit()
-
-		except frappe.DuplicateEntryError as e:
-			# print("Failed to insert duplicate {0} {1}".format(doctype, doc.name))
-
-			# pass DuplicateEntryError and continue
-			if e.args and e.args[0] == doc.doctype and e.args[1] == doc.name:
-				# make sure DuplicateEntryError is for the exact same doc and not a related doc
-				frappe.clear_messages()
-			else:
-				raise
-
+			frappe.db.savepoint(savepoint)
+			doc.insert(ignore_permissions=True, ignore_if_duplicate=True)
 		except Exception as e:
-			frappe.db.rollback()
+			frappe.db.rollback(save_point=savepoint)
 			exception = record.get("__exception")
 			if exception:
 				config = _dict(exception)

frappe/email/doctype/newsletter/test_newsletter.py

@@ -72,7 +72,7 @@ class TestNewsletterMixin:
 						"doctype": doctype,
 						**email_filters,
 					}
-				).insert()
+				).insert(ignore_if_duplicate=True)
 			except Exception:
 				frappe.db.rollback(save_point=savepoint)
 				frappe.db.update(doctype, email_filters, "unsubscribed", 0)

frappe/tests/test_db.py

@@ -182,10 +182,12 @@ class TestDB(unittest.TestCase):
 		self.assertIn("tabToDo", frappe.flags.touched_tables)
 
 		frappe.flags.touched_tables = set()
-		create_custom_field("ToDo", {"label": "ToDo Custom Field"})
-
+		cf = create_custom_field("ToDo", {"label": "ToDo Custom Field"})
 		self.assertIn("tabToDo", frappe.flags.touched_tables)
 		self.assertIn("tabCustom Field", frappe.flags.touched_tables)
+		if cf:
+			cf.delete()
+		frappe.db.commit()
 		frappe.flags.in_migrate = False
 		frappe.flags.touched_tables.clear()
 
@@ -867,3 +869,18 @@ class TestDDLCommandsPost(unittest.TestCase):
 		self.assertIn(
 			"is null", frappe.db.get_values(user, filters={user.name: ("is", "not set")}, run=False).lower()
 		)
+
+
+@run_only_if(db_type_is.POSTGRES)
+class TestTransactionManagement(unittest.TestCase):
+	def test_create_proper_transactions(self):
+		def _get_transaction_id():
+			return frappe.db.sql("select txid_current()", pluck=True)
+
+		self.assertEqual(_get_transaction_id(), _get_transaction_id())
+
+		frappe.db.rollback()
+		self.assertEqual(_get_transaction_id(), _get_transaction_id())
+
+		frappe.db.commit()
+		self.assertEqual(_get_transaction_id(), _get_transaction_id())

frappe/tests/test_rename_doc.py

@@ -100,8 +100,6 @@ class TestRenameDoc(unittest.TestCase):
 				frappe.delete_doc("DocType", dt)
 				frappe.db.sql_ddl(f"DROP TABLE IF EXISTS `tab{dt}`")
 
-		frappe.delete_doc_if_exists("Renamed Doc", "ToDo")
-
 		# reset original value of developer_mode conf
 		frappe.conf.developer_mode = self._original_developer_flag