Create initial 2.0 engine implementation

Implemented the SQLAlchemy 2 :func:`.future.create_engine` function which
is used for forwards compatibility with SQLAlchemy 2.   This engine
features always-transactional behavior with autobegin.

Allow execution options per statement execution.  This includes
that the before_execute() and after_execute() events now accept
an additional dictionary with these options, empty if not
passed; a legacy event decorator is added for backwards compatibility
which now also emits a deprecation warning.

Add some basic tests for execution, transactions, and
the new result object.   Build out on a new testing fixture
that swaps in the future engine completely to start with.

Change-Id: I70e7338bb3f0ce22d2f702537d94bb249bd9fb0a
Fixes: #4644

4 files changed, 52 insertions, 38 deletions

diff --git a/lib/sqlalchemy/dialects/mssql/base.py b/lib/sqlalchemy/dialects/mssql/base.py
index b0021e873..df6196bae 100644
--- a/lib/sqlalchemy/dialects/mssql/base.py
+++ b/lib/sqlalchemy/dialects/mssql/base.py
@@ -2544,16 +2544,20 @@ class MSDialect(default.DefaultDialect):
 
     @_db_plus_owner
     def has_table(self, connection, tablename, dbname, owner, schema):
-        columns = ischema.columns
+        tables = ischema.tables
 
-        whereclause = columns.c.table_name == tablename
+        s = sql.select([tables.c.table_name]).where(
+            sql.and_(
+                tables.c.table_type == "BASE TABLE",
+                tables.c.table_name == tablename,
+            )
+        )
 
         if owner:
-            whereclause = sql.and_(
-                whereclause, columns.c.table_schema == owner
-            )
-        s = sql.select([columns], whereclause)
+            s = s.where(tables.c.table_schema == owner)
+
         c = connection.execute(s)
+
         return c.first() is not None
 
     @reflection.cache
@@ -2569,13 +2573,15 @@ class MSDialect(default.DefaultDialect):
     @_db_plus_owner_listing
     def get_table_names(self, connection, dbname, owner, schema, **kw):
         tables = ischema.tables
-        s = sql.select(
-            [tables.c.table_name],
-            sql.and_(
-                tables.c.table_schema == owner,
-                tables.c.table_type == "BASE TABLE",
-            ),
-            order_by=[tables.c.table_name],
+        s = (
+            sql.select([tables.c.table_name])
+            .where(
+                sql.and_(
+                    tables.c.table_schema == owner,
+                    tables.c.table_type == "BASE TABLE",
+                )
+            )
+            .order_by(tables.c.table_name)
         )
         table_names = [r[0] for r in connection.execute(s)]
         return table_names
diff --git a/lib/sqlalchemy/dialects/mysql/base.py b/lib/sqlalchemy/dialects/mysql/base.py
index c7c3bd433..53c916304 100644
--- a/lib/sqlalchemy/dialects/mysql/base.py
+++ b/lib/sqlalchemy/dialects/mysql/base.py
@@ -2379,25 +2379,25 @@ class MySQLDialect(default.DefaultDialect):
             raise
 
     def do_begin_twophase(self, connection, xid):
-        connection.execute(sql.text("XA BEGIN :xid"), xid=xid)
+        connection.execute(sql.text("XA BEGIN :xid"), dict(xid=xid))
 
     def do_prepare_twophase(self, connection, xid):
-        connection.execute(sql.text("XA END :xid"), xid=xid)
-        connection.execute(sql.text("XA PREPARE :xid"), xid=xid)
+        connection.execute(sql.text("XA END :xid"), dict(xid=xid))
+        connection.execute(sql.text("XA PREPARE :xid"), dict(xid=xid))
 
     def do_rollback_twophase(
         self, connection, xid, is_prepared=True, recover=False
     ):
         if not is_prepared:
-            connection.execute(sql.text("XA END :xid"), xid=xid)
-        connection.execute(sql.text("XA ROLLBACK :xid"), xid=xid)
+            connection.execute(sql.text("XA END :xid"), dict(xid=xid))
+        connection.execute(sql.text("XA ROLLBACK :xid"), dict(xid=xid))
 
     def do_commit_twophase(
         self, connection, xid, is_prepared=True, recover=False
     ):
         if not is_prepared:
             self.do_prepare_twophase(connection, xid)
-        connection.execute(sql.text("XA COMMIT :xid"), xid=xid)
+        connection.execute(sql.text("XA COMMIT :xid"), dict(xid=xid))
 
     def do_recover_twophase(self, connection):
         resultset = connection.exec_driver_sql("XA RECOVER")
@@ -2501,8 +2501,7 @@ class MySQLDialect(default.DefaultDialect):
                 "WHERE TABLE_NAME=:name AND "
                 "TABLE_SCHEMA=:schema_name"
             ),
-            name=sequence_name,
-            schema_name=schema,
+            dict(name=sequence_name, schema_name=schema),
         )
         return cursor.first() is not None
 
@@ -2750,7 +2749,7 @@ class MySQLDialect(default.DefaultDialect):
                     :table_data;
                 """
                 ).bindparams(sql.bindparam("table_data", expanding=True)),
-                table_data=col_tuples,
+                dict(table_data=col_tuples),
             )
 
             # in casing=0, table name and schema name come back in their
diff --git a/lib/sqlalchemy/dialects/oracle/base.py b/lib/sqlalchemy/dialects/oracle/base.py
index 50fa71d7e..e0d33cf37 100644
--- a/lib/sqlalchemy/dialects/oracle/base.py
+++ b/lib/sqlalchemy/dialects/oracle/base.py
@@ -1467,8 +1467,10 @@ class OracleDialect(default.DefaultDialect):
                 "SELECT table_name FROM all_tables "
                 "WHERE table_name = :name AND owner = :schema_name"
             ),
-            name=self.denormalize_name(table_name),
-            schema_name=self.denormalize_name(schema),
+            dict(
+                name=self.denormalize_name(table_name),
+                schema_name=self.denormalize_name(schema),
+            ),
         )
         return cursor.first() is not None
 
@@ -1481,8 +1483,10 @@ class OracleDialect(default.DefaultDialect):
                 "WHERE sequence_name = :name AND "
                 "sequence_owner = :schema_name"
             ),
-            name=self.denormalize_name(sequence_name),
-            schema_name=self.denormalize_name(schema),
+            dict(
+                name=self.denormalize_name(sequence_name),
+                schema_name=self.denormalize_name(schema),
+            ),
         )
         return cursor.first() is not None
 
@@ -1525,7 +1529,7 @@ class OracleDialect(default.DefaultDialect):
         q += " AND ".join(clauses)
 
         result = connection.execution_options(future_result=True).execute(
-            sql.text(q), **params
+            sql.text(q), params
         )
         if desired_owner:
             row = result.mappings().first()
@@ -1621,7 +1625,7 @@ class OracleDialect(default.DefaultDialect):
             "OWNER = :owner " "AND IOT_NAME IS NULL " "AND DURATION IS NULL"
         )
 
-        cursor = connection.execute(sql.text(sql_str), owner=schema)
+        cursor = connection.execute(sql.text(sql_str), dict(owner=schema))
         return [self.normalize_name(row[0]) for row in cursor]
 
     @reflection.cache
@@ -1641,14 +1645,16 @@ class OracleDialect(default.DefaultDialect):
             "AND DURATION IS NOT NULL"
         )
 
-        cursor = connection.execute(sql.text(sql_str), owner=schema)
+        cursor = connection.execute(sql.text(sql_str), dict(owner=schema))
         return [self.normalize_name(row[0]) for row in cursor]
 
     @reflection.cache
     def get_view_names(self, connection, schema=None, **kw):
         schema = self.denormalize_name(schema or self.default_schema_name)
         s = sql.text("SELECT view_name FROM all_views WHERE owner = :owner")
-        cursor = connection.execute(s, owner=self.denormalize_name(schema))
+        cursor = connection.execute(
+            s, dict(owner=self.denormalize_name(schema))
+        )
         return [self.normalize_name(row[0]) for row in cursor]
 
     @reflection.cache
@@ -1687,7 +1693,7 @@ class OracleDialect(default.DefaultDialect):
             text += " AND owner = :owner "
         text = text % {"dblink": dblink, "columns": ", ".join(columns)}
 
-        result = connection.execute(sql.text(text), **params)
+        result = connection.execute(sql.text(text), params)
 
         enabled = dict(DISABLED=False, ENABLED=True)
 
@@ -1752,7 +1758,7 @@ class OracleDialect(default.DefaultDialect):
         text += " ORDER BY col.column_id"
         text = text % {"dblink": dblink, "char_length_col": char_length_col}
 
-        c = connection.execute(sql.text(text), **params)
+        c = connection.execute(sql.text(text), params)
 
         for row in c:
             colname = self.normalize_name(row[0])
@@ -1842,7 +1848,8 @@ class OracleDialect(default.DefaultDialect):
         """
 
         c = connection.execute(
-            sql.text(COMMENT_SQL), table_name=table_name, schema_name=schema
+            sql.text(COMMENT_SQL),
+            dict(table_name=table_name, schema_name=schema),
         )
         return {"text": c.scalar()}
 
@@ -1890,7 +1897,7 @@ class OracleDialect(default.DefaultDialect):
         text = text % {"dblink": dblink}
 
         q = sql.text(text)
-        rp = connection.execute(q, **params)
+        rp = connection.execute(q, params)
         indexes = []
         last_index_name = None
         pk_constraint = self.get_pk_constraint(
@@ -1987,7 +1994,7 @@ class OracleDialect(default.DefaultDialect):
         )
 
         text = text % {"dblink": dblink}
-        rp = connection.execute(sql.text(text), **params)
+        rp = connection.execute(sql.text(text), params)
         constraint_data = rp.fetchall()
         return constraint_data
 
@@ -2215,7 +2222,7 @@ class OracleDialect(default.DefaultDialect):
             text += " AND owner = :schema"
             params["schema"] = schema
 
-        rp = connection.execute(sql.text(text), **params).scalar()
+        rp = connection.execute(sql.text(text), params).scalar()
         if rp:
             if util.py2k:
                 rp = rp.decode(self.encoding)
diff --git a/lib/sqlalchemy/dialects/postgresql/base.py b/lib/sqlalchemy/dialects/postgresql/base.py
index f9b3d9b95..20540ac02 100644
--- a/lib/sqlalchemy/dialects/postgresql/base.py
+++ b/lib/sqlalchemy/dialects/postgresql/base.py
@@ -2753,7 +2753,7 @@ class PGDialect(default.DefaultDialect):
         s = s.columns(oid=sqltypes.Integer)
         if schema:
             s = s.bindparams(sql.bindparam("schema", type_=sqltypes.Unicode))
-        c = connection.execute(s, table_name=table_name, schema=schema)
+        c = connection.execute(s, dict(table_name=table_name, schema=schema))
         table_oid = c.scalar()
         if table_oid is None:
             raise exc.NoSuchTableError(table_name)
@@ -3519,7 +3519,9 @@ class PGDialect(default.DefaultDialect):
                 pgd.objoid = :table_oid
         """
 
-        c = connection.execute(sql.text(COMMENT_SQL), table_oid=table_oid)
+        c = connection.execute(
+            sql.text(COMMENT_SQL), dict(table_oid=table_oid)
+        )
         return {"text": c.scalar()}
 
     @reflection.cache