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

6 files changed, 41 insertions, 10 deletions

diff --git a/lib/sqlalchemy/sql/base.py b/lib/sqlalchemy/sql/base.py
index d070027c8..2d023c6a6 100644
--- a/lib/sqlalchemy/sql/base.py
+++ b/lib/sqlalchemy/sql/base.py
@@ -461,6 +461,17 @@ class Generative(HasMemoized):
         return s
 
 
+class InPlaceGenerative(HasMemoized):
+    """Provide a method-chaining pattern in conjunction with the
+    @_generative decorator taht mutates in place."""
+
+    def _generate(self):
+        skip = self._memoized_keys
+        for k in skip:
+            self.__dict__.pop(k, None)
+        return self
+
+
 class HasCompileState(Generative):
     """A class that has a :class:`.CompileState` associated with it."""
 
diff --git a/lib/sqlalchemy/sql/compiler.py b/lib/sqlalchemy/sql/compiler.py
index 7ef9f7e6d..bc16b1429 100644
--- a/lib/sqlalchemy/sql/compiler.py
+++ b/lib/sqlalchemy/sql/compiler.py
@@ -434,9 +434,13 @@ class Compiled(object):
                     self.string, schema_translate_map
                 )
 
-    def _execute_on_connection(self, connection, multiparams, params):
+    def _execute_on_connection(
+        self, connection, multiparams, params, execution_options
+    ):
         if self.can_execute:
-            return connection._execute_compiled(self, multiparams, params)
+            return connection._execute_compiled(
+                self, multiparams, params, execution_options
+            )
         else:
             raise exc.ObjectNotExecutableError(self.statement)
 
diff --git a/lib/sqlalchemy/sql/ddl.py b/lib/sqlalchemy/sql/ddl.py
index 4c8250e98..51526173f 100644
--- a/lib/sqlalchemy/sql/ddl.py
+++ b/lib/sqlalchemy/sql/ddl.py
@@ -68,8 +68,12 @@ class DDLElement(roles.DDLRole, Executable, _DDLCompiles):
     dialect = None
     callable_ = None
 
-    def _execute_on_connection(self, connection, multiparams, params):
-        return connection._execute_ddl(self, multiparams, params)
+    def _execute_on_connection(
+        self, connection, multiparams, params, execution_options
+    ):
+        return connection._execute_ddl(
+            self, multiparams, params, execution_options
+        )
 
     def execute(self, bind=None, target=None):
         """Execute this DDL immediately.
diff --git a/lib/sqlalchemy/sql/elements.py b/lib/sqlalchemy/sql/elements.py
index 49bb08644..d8b5a1626 100644
--- a/lib/sqlalchemy/sql/elements.py
+++ b/lib/sqlalchemy/sql/elements.py
@@ -282,9 +282,13 @@ class ClauseElement(
         d.pop("_generate_cache_key", None)
         return d
 
-    def _execute_on_connection(self, connection, multiparams, params):
+    def _execute_on_connection(
+        self, connection, multiparams, params, execution_options
+    ):
         if self.supports_execution:
-            return connection._execute_clauseelement(self, multiparams, params)
+            return connection._execute_clauseelement(
+                self, multiparams, params, execution_options
+            )
         else:
             raise exc.ObjectNotExecutableError(self)
 
diff --git a/lib/sqlalchemy/sql/functions.py b/lib/sqlalchemy/sql/functions.py
index 1b10df954..cedb76f55 100644
--- a/lib/sqlalchemy/sql/functions.py
+++ b/lib/sqlalchemy/sql/functions.py
@@ -115,8 +115,12 @@ class FunctionElement(Executable, ColumnElement, FromClause):
             operator=operators.comma_op, group_contents=True, *args
         ).self_group()
 
-    def _execute_on_connection(self, connection, multiparams, params):
-        return connection._execute_function(self, multiparams, params)
+    def _execute_on_connection(
+        self, connection, multiparams, params, execution_options
+    ):
+        return connection._execute_function(
+            self, multiparams, params, execution_options
+        )
 
     @property
     def columns(self):
diff --git a/lib/sqlalchemy/sql/schema.py b/lib/sqlalchemy/sql/schema.py
index ec8d5a458..eddd62d65 100644
--- a/lib/sqlalchemy/sql/schema.py
+++ b/lib/sqlalchemy/sql/schema.py
@@ -2186,8 +2186,12 @@ class DefaultGenerator(SchemaItem):
             bind = _bind_or_error(self)
         return bind.execute(self, **kwargs)
 
-    def _execute_on_connection(self, connection, multiparams, params):
-        return connection._execute_default(self, multiparams, params)
+    def _execute_on_connection(
+        self, connection, multiparams, params, execution_options
+    ):
+        return connection._execute_default(
+            self, multiparams, params, execution_options
+        )
 
     @property
     def bind(self):