Update transaction / connection handling

step one, do away with __connection attribute and using
awkward AttributeError logic

step two, move all management of "connection._transaction"
into the transaction objects themselves where it's easier
to follow.

build MarkerTransaction that takes the role of
"do-nothing block"

new connection datamodel is: connection._transaction, always
a root, connection._nested_transaction, always a nested.

nested transactions still chain to each other as this
is still sort of necessary but they consider the root
transaction separately, and the marker transactions
not at all.

introduce new InvalidRequestError subclass
PendingRollbackError.  Apply to connection and session
for all cases where a transaction needs to be rolled
back before continuing.   Within Connection,
both PendingRollbackError as well as ResourceClosedError
are now raised directly without being handled by
handle_dbapi_error();  this removes these two exception
cases from the handle_error event handler as well as
from StatementError wrapping, as these two exceptions are
not statement oriented and are instead programmatic
issues, that the application is failing to handle database
errors properly.

Revise savepoints so that when a release fails, they set
themselves as inactive so that their rollback() method
does not throw another exception.

Give savepoints another go on MySQL, can't get release working
however get support for basic round trip going

Fixes: #5327
Change-Id: Ia3cbbf56d4882fcc7980f90519412f1711fae74d

1 files changed, 12 insertions, 0 deletions

diff --git a/lib/sqlalchemy/testing/plugin/plugin_base.py b/lib/sqlalchemy/testing/plugin/plugin_base.py
index 9b2f6911d..f7d0dd3ea 100644
--- a/lib/sqlalchemy/testing/plugin/plugin_base.py
+++ b/lib/sqlalchemy/testing/plugin/plugin_base.py
@@ -412,6 +412,17 @@ def _prep_testing_database(options, file_config):
     if options.dropfirst:
         for cfg in config.Config.all_configs():
             e = cfg.db
+
+            # TODO: this has to be part of provision.py in postgresql
+            if against(cfg, "postgresql"):
+                with e.connect().execution_options(
+                    isolation_level="AUTOCOMMIT"
+                ) as conn:
+                    for xid in conn.execute(
+                        "select gid from pg_prepared_xacts"
+                    ).scalars():
+                        conn.execute("ROLLBACK PREPARED '%s'" % xid)
+
             inspector = inspect(e)
             try:
                 view_names = inspector.get_view_names()
@@ -447,6 +458,7 @@ def _prep_testing_database(options, file_config):
             if config.requirements.schemas.enabled_for_config(cfg):
                 util.drop_all_tables(e, inspector, schema=cfg.test_schema)
 
+            # TODO: this has to be part of provision.py in postgresql
             if against(cfg, "postgresql"):
                 from sqlalchemy.dialects import postgresql