Establish flush() for "sub" facade contexts

When converting existing session context manager code to
use enginefacade, normally a flush() is implicit on
a "subtransaction" when it closes.  Enginefacade
has not maintained this behavior.  Ideally, this should
have been on by default when enginefacade was first
released, but for now add this as a behavioral
option for projects that require it.

Co-authored-by: Mike Bayer <mike_mp@zzzcomputing.com>
Closes-Bug: #1664643
Change-Id: I6f4f439928588cff954e749dfa938425892e0931

2 files changed, 69 insertions, 1 deletions

diff --git a/oslo_db/sqlalchemy/enginefacade.py b/oslo_db/sqlalchemy/enginefacade.py
index 057e89e..2519668 100644
--- a/oslo_db/sqlalchemy/enginefacade.py
+++ b/oslo_db/sqlalchemy/enginefacade.py
@@ -154,6 +154,7 @@ class _TransactionFactory(object):
         }
         self._transaction_ctx_cfg = {
             'rollback_reader_sessions': False,
+            'flush_on_subtransaction': False,
             }
         self._facade_cfg = {
             'synchronous_reader': True,
@@ -264,6 +265,13 @@ class _TransactionFactory(object):
          @reader context works the same as @async_reader and will select
          the "slave" database if present.
 
+        :param flush_on_subtransaction: if True, a :class:`.Session` object
+         will have its :meth:`.Session.flush` method invoked whenever a context
+         manager or decorator that is not itself the originator of the top-
+         level or savepoint :class:`.Session` transaction exits - in this way
+         it behaves like a "subtransaction" from a :class:`.Session`
+         perspective.
+
         .. seealso::
 
             :meth:`._TransactionFactory.configure`
@@ -558,7 +566,6 @@ class _TransactionContext(object):
          objects created under this one.  When left as None the actual
          "global" factory is used.
 
-
         """
         self.factory = factory
         self.global_factory = global_factory
@@ -568,6 +575,7 @@ class _TransactionContext(object):
         self.transaction = None
         kw = self.factory._transaction_ctx_cfg
         self.rollback_reader_sessions = kw['rollback_reader_sessions']
+        self.flush_on_subtransaction = kw['flush_on_subtransaction']
 
     @contextlib.contextmanager
     def _connection(self, savepoint=False):
@@ -636,6 +644,8 @@ class _TransactionContext(object):
                     yield self.session
             else:
                 yield self.session
+                if self.flush_on_subtransaction:
+                    self.session.flush()
 
     def _end_session_transaction(self, session):
         if self.mode is _WRITER:
diff --git a/oslo_db/tests/sqlalchemy/test_enginefacade.py b/oslo_db/tests/sqlalchemy/test_enginefacade.py
index a606fae..566dd36 100644
--- a/oslo_db/tests/sqlalchemy/test_enginefacade.py
+++ b/oslo_db/tests/sqlalchemy/test_enginefacade.py
@@ -389,6 +389,12 @@ class MockFacadeTest(oslo_test_base.BaseTestCase):
             makers, enginefacade._WRITER,
             connection, assert_calls)
 
+    def _emit_sub_writer_session(self, session):
+        return self._emit_sub_session(enginefacade._WRITER, session)
+
+    def _emit_sub_reader_session(self, session):
+        return self._emit_sub_session(enginefacade._READER, session)
+
     @contextlib.contextmanager
     def _assert_session(
             self, makers, writer, connection=None, assert_calls=True):
@@ -418,6 +424,13 @@ class MockFacadeTest(oslo_test_base.BaseTestCase):
                 session.mock_calls,
                 self.sessions.element_for_writer(writer).mock_calls)
 
+    @contextlib.contextmanager
+    def _emit_sub_session(self, writer, session):
+        yield session
+        if enginefacade._context_manager.\
+                _factory._transaction_ctx_cfg['flush_on_subtransaction']:
+            session.flush()
+
     def test_dispose_pool(self):
         facade = enginefacade.transaction_context()
 
@@ -771,6 +784,23 @@ class MockFacadeTest(oslo_test_base.BaseTestCase):
                 with self._assert_reader_session(makers) as session:
                     session.execute("test1")
 
+    def test_using_flush_on_nested(self):
+        enginefacade.configure(flush_on_nested=True)
+
+        context = oslo_context.RequestContext()
+
+        with enginefacade.writer.using(context) as session:
+            with enginefacade.writer.using(context) as session:
+                self._assert_ctx_session(context, session)
+                session.execute("test1")
+
+        with self._assert_engines() as engines:
+            with self._assert_makers(engines) as makers:
+                with self._assert_writer_session(makers) as session:
+                    with self._emit_sub_writer_session(
+                            session) as session:
+                        session.execute("test1")
+
     def test_using_writer(self):
         context = oslo_context.RequestContext()
 
@@ -1555,6 +1585,7 @@ class LiveFacadeTest(test_base.DbTestCase):
             'user', metadata,
             Column('id', Integer, primary_key=True),
             Column('name', String(30)),
+            Column('favorite_color', String(10), default='yellow'),
             mysql_engine='InnoDB'
         )
         self.user_table = user_table
@@ -1606,6 +1637,33 @@ class LiveFacadeTest(test_base.DbTestCase):
             session.query(self.User.name).scalar()
         )
 
+    def test_flush_on_subtransaction(self):
+        facade = enginefacade.transaction_context()
+        facade.configure(
+            connection=self.engine.url,
+            flush_on_subtransaction=True)
+        facade.patch_engine(self.engine)
+        context = oslo_context.RequestContext()
+
+        with facade.writer.using(context):
+            with facade.writer.using(context):
+                u = self.User(name="u1")
+                context.session.add(u)
+            self.assertEqual('yellow', u.favorite_color)
+
+    def test_flush_on_subtransaction_default_off(self):
+        context = oslo_context.RequestContext()
+        facade = enginefacade.transaction_context()
+        facade.configure(connection=self.engine.url)
+        facade.patch_engine(self.engine)
+
+        with facade.writer.using(context):
+            with facade.writer.using(context):
+                u = self.User(name="u1")
+                context.session.add(u)
+            self.assertIsNone(u.favorite_color)
+        self.assertEqual('yellow', u.favorite_color)
+
     def test_context_deepcopy_on_session(self):
         context = oslo_context.RequestContext()
         with enginefacade.writer.using(context) as session: