From 9c47822eafc92634e5654ebdb6aa0ec7a03a7c5b Mon Sep 17 00:00:00 2001
From: Lance Edgar <lance@edbob.org>
Date: Mon, 4 Apr 2022 13:07:23 -0500
Subject: [PATCH] Add transaction logic for rattail->rattail datasync

per newer conventions
---
 rattail/datasync/rattail.py | 45 +++++++++++++++++--------------------
 1 file changed, 21 insertions(+), 24 deletions(-)

diff --git a/rattail/datasync/rattail.py b/rattail/datasync/rattail.py
index 7a81bd7f..9a3f06b3 100644
--- a/rattail/datasync/rattail.py
+++ b/rattail/datasync/rattail.py
@@ -426,35 +426,32 @@ class FromRattailToRattailExportConsumer(FromRattailToRattailBase):
     def make_target_session(self):
         return self.app.make_session(bind=self.target_engine)
 
-    def process_changes(self, session, changes):
-        local_session = self.app.make_session()
-        target_session = self.make_target_session()
+    def begin_transaction(self):
+        self.local_session = self.app.make_session()
+        self.target_session = self.make_target_session()
 
-        try:
-            if self.runas_username:
-                target_session.set_continuum_user(self.runas_username)
+    def rollback_transaction(self):
+        self.target_session.rollback()
+        self.target_session.close()
+        del self.target_session
 
-            # update all importers with current sessions
-            for importer in six.itervalues(self.importers):
-                importer.host_session = local_session
-                importer.session = target_session
+        self.local_session.rollback()
+        self.local_session.close()
+        del self.local_session
 
-            # delegate processing to importer(s)
-            for change in changes:
-                self.invoke_importer(session, change)
+    def commit_transaction(self):
+        # TODO: should we ever commit local?
+        #self.local_session.commit()
+        self.target_session.commit()
 
-            # TODO: should we ever commit local?
-            #local_session.commit()
-            target_session.commit()
+    def pre_process_changes(self, session, changes):
+        if self.runas_username:
+            self.target_session.set_continuum_user(self.runas_username)
 
-        except:
-            log.exception("failed to process changes")
-            local_session.rollback()
-            target_session.rollback()
-
-        finally:
-            local_session.close()
-            target_session.close()
+        # update all importers with current sessions
+        for importer in six.itervalues(self.importers):
+            importer.host_session = self.local_session
+            importer.session = self.target_session
 
 
 class FromRattailToRattailImportConsumer(FromRattailToRattailBase):