Deprecate connection branching

The :meth:`.Connection.connect` method is deprecated as is the concept of
"connection branching", which copies a :class:`.Connection` into a new one
that has a no-op ".close()" method.  This pattern is oriented around the
"connectionless execution" concept which is also being removed in 2.0.

As part of this change we begin to move the internals away from
"connectionless execution" overall.   Remove the "connectionless
execution" concept from the reflection internals and replace with
explicit patterns at the Inspector level.

Fixes: #5131
Change-Id: Id23d28a9889212ac5ae7329b85136157815d3e6f

4 files changed, 241 insertions, 123 deletions

diff --git a/lib/sqlalchemy/engine/base.py b/lib/sqlalchemy/engine/base.py
index 462e5f9ec..29df67dcb 100644
--- a/lib/sqlalchemy/engine/base.py
+++ b/lib/sqlalchemy/engine/base.py
@@ -102,6 +102,15 @@ class Connection(Connectable):
             self.__transaction = None
             self.__savepoint_seq = 0
             self.should_close_with_result = close_with_result
+            if close_with_result:
+                util.warn_deprecated_20(
+                    '"Connectionless" execution, which refers to running '
+                    "SQL commands using the Engine.execute() (or "
+                    "some_statement.execute()) method without "
+                    "calling .connect() or .begin() to get a Connection, is "
+                    "deprecated and will be removed SQLAlchemy 2.0"
+                )
+
             self.__invalid = False
             self.__can_reconnect = True
             self._echo = self.engine._should_log_info()
@@ -489,6 +498,7 @@ class Connection(Connectable):
 
         return self.connection.info
 
+    @util.deprecated_20(":meth:`.Connection.connect`")
     def connect(self, close_with_result=False):
         """Returns a branched version of this :class:`.Connection`.
 
@@ -884,6 +894,12 @@ class Connection(Connectable):
 
         """
         if self.__branch_from:
+            util.warn_deprecated(
+                "The .close() method on a so-called 'branched' connection is "
+                "deprecated as of 1.4, as are 'branched' connections overall, "
+                "and will be removed in a future release.  If this is a "
+                "default-handling function, don't close the connection."
+            )
             try:
                 del self.__connection
             except AttributeError:
@@ -2237,7 +2253,6 @@ class Engine(Connectable, log.Identified):
         resource to be returned to the connection pool.
 
         """
-
         connection = self.connect(close_with_result=True)
         return connection.execute(statement, *multiparams, **params)
 
diff --git a/lib/sqlalchemy/engine/default.py b/lib/sqlalchemy/engine/default.py
index d900a74b8..7d36345fd 100644
--- a/lib/sqlalchemy/engine/default.py
+++ b/lib/sqlalchemy/engine/default.py
@@ -810,12 +810,12 @@ class DefaultExecutionContext(interfaces.ExecutionContext):
         parameters = []
         if compiled.positional:
             for compiled_params in self.compiled_parameters:
-                param = []
-                for key in positiontup:
-                    if key in processors:
-                        param.append(processors[key](compiled_params[key]))
-                    else:
-                        param.append(compiled_params[key])
+                param = [
+                    processors[key](compiled_params[key])
+                    if key in processors
+                    else compiled_params[key]
+                    for key in positiontup
+                ]
                 parameters.append(dialect.execute_sequence_format(param))
         else:
             encode = not dialect.supports_unicode_statements
@@ -948,7 +948,7 @@ class DefaultExecutionContext(interfaces.ExecutionContext):
         else:
             return autocommit
 
-    def _execute_scalar(self, stmt, type_):
+    def _execute_scalar(self, stmt, type_, parameters=None):
         """Execute a string statement on the current cursor, returning a
         scalar result.
 
@@ -965,12 +965,13 @@ class DefaultExecutionContext(interfaces.ExecutionContext):
         ):
             stmt = self.dialect._encoder(stmt)[0]
 
-        if self.dialect.positional:
-            default_params = self.dialect.execute_sequence_format()
-        else:
-            default_params = {}
+        if not parameters:
+            if self.dialect.positional:
+                parameters = self.dialect.execute_sequence_format()
+            else:
+                parameters = {}
 
-        conn._cursor_execute(self.cursor, stmt, default_params, context=self)
+        conn._cursor_execute(self.cursor, stmt, parameters, context=self)
         r = self.cursor.fetchone()[0]
         if type_ is not None:
             # apply type post processors to the result
@@ -1288,18 +1289,51 @@ class DefaultExecutionContext(interfaces.ExecutionContext):
             self.current_column = column
             return default.arg(self)
         elif default.is_clause_element:
-            # TODO: expensive branching here should be
-            # pulled into _exec_scalar()
-            conn = self.connection
-            if not default._arg_is_typed:
-                default_arg = expression.type_coerce(default.arg, type_)
-            else:
-                default_arg = default.arg
-            c = expression.select([default_arg]).compile(bind=conn)
-            return conn._execute_compiled(c, (), {}).scalar()
+            return self._exec_default_clause_element(column, default, type_)
         else:
             return default.arg
 
+    def _exec_default_clause_element(self, column, default, type_):
+        # execute a default that's a complete clause element.  Here, we have
+        # to re-implement a miniature version of the compile->parameters->
+        # cursor.execute() sequence, since we don't want to modify the state
+        # of the connection  / result in progress or create new connection/
+        # result objects etc.
+        # .. versionchanged:: 1.4
+
+        if not default._arg_is_typed:
+            default_arg = expression.type_coerce(default.arg, type_)
+        else:
+            default_arg = default.arg
+        compiled = expression.select([default_arg]).compile(
+            dialect=self.dialect
+        )
+        compiled_params = compiled.construct_params()
+        processors = compiled._bind_processors
+        if compiled.positional:
+            positiontup = compiled.positiontup
+            parameters = self.dialect.execute_sequence_format(
+                [
+                    processors[key](compiled_params[key])
+                    if key in processors
+                    else compiled_params[key]
+                    for key in positiontup
+                ]
+            )
+        else:
+            parameters = dict(
+                (
+                    key,
+                    processors[key](compiled_params[key])
+                    if key in processors
+                    else compiled_params[key],
+                )
+                for key in compiled_params
+            )
+        return self._execute_scalar(
+            util.text_type(compiled), type_, parameters=parameters
+        )
+
     current_parameters = None
     """A dictionary of parameters applied to the current row.
 
diff --git a/lib/sqlalchemy/engine/interfaces.py b/lib/sqlalchemy/engine/interfaces.py
index cffaa159b..237eb0f2f 100644
--- a/lib/sqlalchemy/engine/interfaces.py
+++ b/lib/sqlalchemy/engine/interfaces.py
@@ -1092,6 +1092,16 @@ class ExecutionContext(object):
         raise NotImplementedError()
 
 
+@util.deprecated_20_cls(
+    ":class:`.Connectable`",
+    alternative=(
+        "The :class:`.Engine` will be the only Core "
+        "object that features a .connect() method, and the "
+        ":class:`.Connection` will be the only object that features "
+        "an .execute() method."
+    ),
+    constructor=None,
+)
 class Connectable(object):
     """Interface for an object which supports execution of SQL constructs.
 
@@ -1120,34 +1130,6 @@ class Connectable(object):
 
     """
 
-    @util.deprecated(
-        "0.7",
-        "The :meth:`.Connectable.create` method is deprecated and will be "
-        "removed in a future release.  Please use the ``.create()`` method "
-        "on specific schema objects to emit DDL sequences, including "
-        ":meth:`.Table.create`, :meth:`.Index.create`, and "
-        ":meth:`.MetaData.create_all`.",
-    )
-    def create(self, entity, **kwargs):
-        """Emit CREATE statements for the given schema entity.
-        """
-
-        raise NotImplementedError()
-
-    @util.deprecated(
-        "0.7",
-        "The :meth:`.Connectable.drop` method is deprecated and will be "
-        "removed in a future release.  Please use the ``.drop()`` method "
-        "on specific schema objects to emit DDL sequences, including "
-        ":meth:`.Table.drop`, :meth:`.Index.drop`, and "
-        ":meth:`.MetaData.drop_all`.",
-    )
-    def drop(self, entity, **kwargs):
-        """Emit DROP statements for the given schema entity.
-        """
-
-        raise NotImplementedError()
-
     def execute(self, object_, *multiparams, **params):
         """Executes the given construct and returns a :class:`.ResultProxy`."""
         raise NotImplementedError()
diff --git a/lib/sqlalchemy/engine/reflection.py b/lib/sqlalchemy/engine/reflection.py
index d113588bb..25538fddb 100644
--- a/lib/sqlalchemy/engine/reflection.py
+++ b/lib/sqlalchemy/engine/reflection.py
@@ -25,7 +25,11 @@ methods such as get_table_names, get_columns, etc.
    'name' attribute..
 """
 
+import contextlib
+
 from .base import Connectable
+from .base import Connection
+from .base import Engine
 from .. import exc
 from .. import inspection
 from .. import sql
@@ -64,24 +68,27 @@ class Inspector(object):
     fetched metadata.
 
     A :class:`.Inspector` object is usually created via the
-    :func:`.inspect` function::
+    :func:`.inspect` function, which may be passed an :class:`.Engine`
+    or a :class:`.Connection`::
 
         from sqlalchemy import inspect, create_engine
         engine = create_engine('...')
         insp = inspect(engine)
 
-    The inspection method above is equivalent to using the
-    :meth:`.Inspector.from_engine` method, i.e.::
-
-        engine = create_engine('...')
-        insp = Inspector.from_engine(engine)
-
-    Where above, the :class:`~sqlalchemy.engine.interfaces.Dialect` may opt
-    to return an :class:`.Inspector` subclass that provides additional
-    methods specific to the dialect's target database.
+    Where above, the :class:`~sqlalchemy.engine.interfaces.Dialect` associated
+    with the engine may opt to return an :class:`.Inspector` subclass that
+    provides additional methods specific to the dialect's target database.
 
     """
 
+    @util.deprecated(
+        "1.4",
+        "The __init__() method on :class:`.Inspector` is deprecated and "
+        "will be removed in a future release.  Please use the "
+        ":func:`.sqlalchemy.inspect` "
+        "function on an :class:`.Engine` or :class:`.Connection` in order to "
+        "acquire an :class:`.Inspector`.",
+    )
     def __init__(self, bind):
         """Initialize a new :class:`.Inspector`.
 
@@ -94,23 +101,47 @@ class Inspector(object):
         :meth:`.Inspector.from_engine`
 
         """
-        # this might not be a connection, it could be an engine.
-        self.bind = bind
+        return self._init_legacy(bind)
+
+    @classmethod
+    def _construct(cls, init, bind):
 
-        # set the engine
+        if hasattr(bind.dialect, "inspector"):
+            cls = bind.dialect.inspector
+
+        self = cls.__new__(cls)
+        init(self, bind)
+        return self
+
+    def _init_legacy(self, bind):
         if hasattr(bind, "engine"):
-            self.engine = bind.engine
+            self._init_connection(bind)
         else:
-            self.engine = bind
+            self._init_engine(bind)
 
-        if self.engine is bind:
-            # if engine, ensure initialized
-            bind.connect().close()
+    def _init_engine(self, engine):
+        self.bind = self.engine = engine
+        engine.connect().close()
+        self._op_context_requires_connect = True
+        self.dialect = self.engine.dialect
+        self.info_cache = {}
 
+    def _init_connection(self, connection):
+        self.bind = connection
+        self.engine = connection.engine
+        self._op_context_requires_connect = False
         self.dialect = self.engine.dialect
         self.info_cache = {}
 
     @classmethod
+    @util.deprecated(
+        "1.4",
+        "The from_engine() method on :class:`.Inspector` is deprecated and "
+        "will be removed in a future release.  Please use the "
+        ":func:`.sqlalchemy.inspect` "
+        "function on an :class:`.Engine` or :class:`.Connection` in order to "
+        "acquire an :class:`.Inspector`.",
+    )
     def from_engine(cls, bind):
         """Construct a new dialect-specific Inspector object from the given
         engine or connection.
@@ -129,13 +160,53 @@ class Inspector(object):
         See the example at :class:`.Inspector`.
 
         """
-        if hasattr(bind.dialect, "inspector"):
-            return bind.dialect.inspector(bind)
-        return Inspector(bind)
+        return cls._construct(cls._init_legacy, bind)
 
     @inspection._inspects(Connectable)
-    def _insp(bind):
-        return Inspector.from_engine(bind)
+    def _connectable_insp(bind):
+        # this method should not be used unless some unusual case
+        # has subclassed "Connectable"
+
+        return Inspector._construct(Inspector._init_legacy, bind)
+
+    @inspection._inspects(Engine)
+    def _engine_insp(bind):
+        return Inspector._construct(Inspector._init_engine, bind)
+
+    @inspection._inspects(Connection)
+    def _connection_insp(bind):
+        return Inspector._construct(Inspector._init_connection, bind)
+
+    @contextlib.contextmanager
+    def _operation_context(self):
+        """Return a context that optimizes for multiple operations on a single
+        transaction.
+
+        This essentially allows connect()/close() to be called if we detected
+        that we're against an :class:`.Engine` and not a :class:`.Connection`.
+
+        """
+        if self._op_context_requires_connect:
+            conn = self.bind.connect()
+        else:
+            conn = self.bind
+        try:
+            yield conn
+        finally:
+            if self._op_context_requires_connect:
+                conn.close()
+
+    @contextlib.contextmanager
+    def _inspection_context(self):
+        """Return an :class:`.Inspector` from this one that will run all
+        operations on a single connection.
+
+        """
+
+        with self._operation_context() as conn:
+            sub_insp = self._construct(self.__class__._init_connection, conn)
+            sub_insp.info_cache = self.info_cache
+            yield sub_insp
 
     @property
     def default_schema_name(self):
@@ -153,9 +224,10 @@ class Inspector(object):
         """
 
         if hasattr(self.dialect, "get_schema_names"):
-            return self.dialect.get_schema_names(
-                self.bind, info_cache=self.info_cache
-            )
+            with self._operation_context() as conn:
+                return self.dialect.get_schema_names(
+                    conn, info_cache=self.info_cache
+                )
         return []
 
     def get_table_names(self, schema=None):
@@ -185,9 +257,10 @@ class Inspector(object):
 
         """
 
-        return self.dialect.get_table_names(
-            self.bind, schema, info_cache=self.info_cache
-        )
+        with self._operation_context() as conn:
+            return self.dialect.get_table_names(
+                conn, schema, info_cache=self.info_cache
+            )
 
     def has_table(self, table_name, schema=None):
         """Return True if the backend has a table of the given name.
@@ -196,7 +269,8 @@ class Inspector(object):
 
         """
         # TODO: info_cache?
-        return self.dialect.has_table(self.bind, table_name, schema)
+        with self._operation_context() as conn:
+            return self.dialect.has_table(conn, table_name, schema)
 
     def get_sorted_table_and_fkc_names(self, schema=None):
         """Return dependency-sorted table and foreign key constraint names in
@@ -222,12 +296,11 @@ class Inspector(object):
              with an already-given :class:`.MetaData`.
 
         """
-        if hasattr(self.dialect, "get_table_names"):
+
+        with self._operation_context() as conn:
             tnames = self.dialect.get_table_names(
-                self.bind, schema, info_cache=self.info_cache
+                conn, schema, info_cache=self.info_cache
             )
-        else:
-            tnames = self.engine.table_names(schema)
 
         tuples = set()
         remaining_fkcs = set()
@@ -263,9 +336,11 @@ class Inspector(object):
         .. versionadded:: 1.0.0
 
         """
-        return self.dialect.get_temp_table_names(
-            self.bind, info_cache=self.info_cache
-        )
+
+        with self._operation_context() as conn:
+            return self.dialect.get_temp_table_names(
+                conn, info_cache=self.info_cache
+            )
 
     def get_temp_view_names(self):
         """return a list of temporary view names for the current bind.
@@ -276,9 +351,10 @@ class Inspector(object):
         .. versionadded:: 1.0.0
 
         """
-        return self.dialect.get_temp_view_names(
-            self.bind, info_cache=self.info_cache
-        )
+        with self._operation_context() as conn:
+            return self.dialect.get_temp_view_names(
+                conn, info_cache=self.info_cache
+            )
 
     def get_table_options(self, table_name, schema=None, **kw):
         """Return a dictionary of options specified when the table of the
@@ -295,9 +371,10 @@ class Inspector(object):
 
         """
         if hasattr(self.dialect, "get_table_options"):
-            return self.dialect.get_table_options(
-                self.bind, table_name, schema, info_cache=self.info_cache, **kw
-            )
+            with self._operation_context() as conn:
+                return self.dialect.get_table_options(
+                    conn, table_name, schema, info_cache=self.info_cache, **kw
+                )
         return {}
 
     def get_view_names(self, schema=None):
@@ -308,9 +385,10 @@ class Inspector(object):
 
         """
 
-        return self.dialect.get_view_names(
-            self.bind, schema, info_cache=self.info_cache
-        )
+        with self._operation_context() as conn:
+            return self.dialect.get_view_names(
+                conn, schema, info_cache=self.info_cache
+            )
 
     def get_view_definition(self, view_name, schema=None):
         """Return definition for `view_name`.
@@ -320,9 +398,10 @@ class Inspector(object):
 
         """
 
-        return self.dialect.get_view_definition(
-            self.bind, view_name, schema, info_cache=self.info_cache
-        )
+        with self._operation_context() as conn:
+            return self.dialect.get_view_definition(
+                conn, view_name, schema, info_cache=self.info_cache
+            )
 
     def get_columns(self, table_name, schema=None, **kw):
         """Return information about columns in `table_name`.
@@ -354,9 +433,10 @@ class Inspector(object):
 
         """
 
-        col_defs = self.dialect.get_columns(
-            self.bind, table_name, schema, info_cache=self.info_cache, **kw
-        )
+        with self._operation_context() as conn:
+            col_defs = self.dialect.get_columns(
+                conn, table_name, schema, info_cache=self.info_cache, **kw
+            )
         for col_def in col_defs:
             # make this easy and only return instances for coltype
             coltype = col_def["type"]
@@ -377,9 +457,10 @@ class Inspector(object):
         primary key information as a list of column names.
         """
 
-        return self.dialect.get_pk_constraint(
-            self.bind, table_name, schema, info_cache=self.info_cache, **kw
-        )["constrained_columns"]
+        with self._operation_context() as conn:
+            return self.dialect.get_pk_constraint(
+                conn, table_name, schema, info_cache=self.info_cache, **kw
+            )["constrained_columns"]
 
     def get_pk_constraint(self, table_name, schema=None, **kw):
         """Return information about primary key constraint on `table_name`.
@@ -401,9 +482,10 @@ class Inspector(object):
          use :class:`.quoted_name`.
 
         """
-        return self.dialect.get_pk_constraint(
-            self.bind, table_name, schema, info_cache=self.info_cache, **kw
-        )
+        with self._operation_context() as conn:
+            return self.dialect.get_pk_constraint(
+                conn, table_name, schema, info_cache=self.info_cache, **kw
+            )
 
     def get_foreign_keys(self, table_name, schema=None, **kw):
         """Return information about foreign_keys in `table_name`.
@@ -436,9 +518,10 @@ class Inspector(object):
 
         """
 
-        return self.dialect.get_foreign_keys(
-            self.bind, table_name, schema, info_cache=self.info_cache, **kw
-        )
+        with self._operation_context() as conn:
+            return self.dialect.get_foreign_keys(
+                conn, table_name, schema, info_cache=self.info_cache, **kw
+            )
 
     def get_indexes(self, table_name, schema=None, **kw):
         """Return information about indexes in `table_name`.
@@ -476,9 +559,10 @@ class Inspector(object):
 
         """
 
-        return self.dialect.get_indexes(
-            self.bind, table_name, schema, info_cache=self.info_cache, **kw
-        )
+        with self._operation_context() as conn:
+            return self.dialect.get_indexes(
+                conn, table_name, schema, info_cache=self.info_cache, **kw
+            )
 
     def get_unique_constraints(self, table_name, schema=None, **kw):
         """Return information about unique constraints in `table_name`.
@@ -501,9 +585,10 @@ class Inspector(object):
 
         """
 
-        return self.dialect.get_unique_constraints(
-            self.bind, table_name, schema, info_cache=self.info_cache, **kw
-        )
+        with self._operation_context() as conn:
+            return self.dialect.get_unique_constraints(
+                conn, table_name, schema, info_cache=self.info_cache, **kw
+            )
 
     def get_table_comment(self, table_name, schema=None, **kw):
         """Return information about the table comment for ``table_name``.
@@ -521,9 +606,10 @@ class Inspector(object):
 
         """
 
-        return self.dialect.get_table_comment(
-            self.bind, table_name, schema, info_cache=self.info_cache, **kw
-        )
+        with self._operation_context() as conn:
+            return self.dialect.get_table_comment(
+                conn, table_name, schema, info_cache=self.info_cache, **kw
+            )
 
     def get_check_constraints(self, table_name, schema=None, **kw):
         """Return information about check constraints in `table_name`.
@@ -554,9 +640,10 @@ class Inspector(object):
 
         """
 
-        return self.dialect.get_check_constraints(
-            self.bind, table_name, schema, info_cache=self.info_cache, **kw
-        )
+        with self._operation_context() as conn:
+            return self.dialect.get_check_constraints(
+                conn, table_name, schema, info_cache=self.info_cache, **kw
+            )
 
     def reflecttable(
         self,