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

11 files changed, 303 insertions, 146 deletions

diff --git a/lib/sqlalchemy/dialects/postgresql/base.py b/lib/sqlalchemy/dialects/postgresql/base.py
index 45911d4c0..ee81fc020 100644
--- a/lib/sqlalchemy/dialects/postgresql/base.py
+++ b/lib/sqlalchemy/dialects/postgresql/base.py
@@ -2245,9 +2245,6 @@ class PGIdentifierPreparer(compiler.IdentifierPreparer):
 
 
 class PGInspector(reflection.Inspector):
-    def __init__(self, conn):
-        reflection.Inspector.__init__(self, conn)
-
     def get_table_oid(self, table_name, schema=None):
         """Return the OID for the given table name."""
 
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,
diff --git a/lib/sqlalchemy/sql/schema.py b/lib/sqlalchemy/sql/schema.py
index 9f66321fe..79a700ad8 100644
--- a/lib/sqlalchemy/sql/schema.py
+++ b/lib/sqlalchemy/sql/schema.py
@@ -4095,9 +4095,7 @@ class MetaData(SchemaItem):
         if bind is None:
             bind = _bind_or_error(self)
 
-        with bind.connect() as conn:
-            insp = inspection.inspect(conn)
-
+        with inspection.inspect(bind)._inspection_context() as insp:
             reflect_opts = {
                 "autoload_with": insp,
                 "extend_existing": extend_existing,
diff --git a/lib/sqlalchemy/testing/__init__.py b/lib/sqlalchemy/testing/__init__.py
index ab1198da8..582901579 100644
--- a/lib/sqlalchemy/testing/__init__.py
+++ b/lib/sqlalchemy/testing/__init__.py
@@ -20,6 +20,7 @@ from .assertions import eq_  # noqa
 from .assertions import eq_ignore_whitespace  # noqa
 from .assertions import eq_regex  # noqa
 from .assertions import expect_deprecated  # noqa
+from .assertions import expect_deprecated_20  # noqa
 from .assertions import expect_warnings  # noqa
 from .assertions import in_  # noqa
 from .assertions import is_  # noqa
diff --git a/lib/sqlalchemy/testing/assertions.py b/lib/sqlalchemy/testing/assertions.py
index c74259bdf..d055ba86e 100644
--- a/lib/sqlalchemy/testing/assertions.py
+++ b/lib/sqlalchemy/testing/assertions.py
@@ -81,6 +81,10 @@ def expect_deprecated(*messages, **kw):
     return _expect_warnings(sa_exc.SADeprecationWarning, messages, **kw)
 
 
+def expect_deprecated_20(*messages, **kw):
+    return _expect_warnings(sa_exc.RemovedIn20Warning, messages, **kw)
+
+
 def emits_warning_on(db, *messages):
     """Mark a test as emitting a warning on a specific dialect.
 
diff --git a/lib/sqlalchemy/testing/suite/test_reflection.py b/lib/sqlalchemy/testing/suite/test_reflection.py
index f9ff46492..d375f0279 100644
--- a/lib/sqlalchemy/testing/suite/test_reflection.py
+++ b/lib/sqlalchemy/testing/suite/test_reflection.py
@@ -21,7 +21,6 @@ from ... import MetaData
 from ... import String
 from ... import testing
 from ... import types as sql_types
-from ...engine.reflection import Inspector
 from ...schema import DDL
 from ...schema import Index
 from ...sql.elements import quoted_name
@@ -661,7 +660,7 @@ class ComponentReflectionTest(fixtures.TablesTest):
     def test_deprecated_get_primary_keys(self):
         meta = self.metadata
         users = self.tables.users
-        insp = Inspector(meta.bind)
+        insp = inspect(meta.bind)
         assert_raises_message(
             sa_exc.SADeprecationWarning,
             r".*get_primary_keys\(\) method is deprecated",
diff --git a/lib/sqlalchemy/util/__init__.py b/lib/sqlalchemy/util/__init__.py
index d2428bf75..434c5cb79 100644
--- a/lib/sqlalchemy/util/__init__.py
+++ b/lib/sqlalchemy/util/__init__.py
@@ -89,6 +89,7 @@ from .compat import with_metaclass  # noqa
 from .compat import zip_longest  # noqa
 from .deprecations import deprecated  # noqa
 from .deprecations import deprecated_20  # noqa
+from .deprecations import deprecated_20_cls  # noqa
 from .deprecations import deprecated_cls  # noqa
 from .deprecations import deprecated_params  # noqa
 from .deprecations import inject_docstring_text  # noqa
diff --git a/lib/sqlalchemy/util/deprecations.py b/lib/sqlalchemy/util/deprecations.py
index 0db2c72ae..b78a71b1b 100644
--- a/lib/sqlalchemy/util/deprecations.py
+++ b/lib/sqlalchemy/util/deprecations.py
@@ -23,7 +23,7 @@ def warn_deprecated(msg, stacklevel=3):
 
 
 def warn_deprecated_20(msg, stacklevel=3):
-    msg += "(Background on SQLAlchemy 2.0 at: http://sqlalche.me/e/b8d9)"
+    msg += " (Background on SQLAlchemy 2.0 at: http://sqlalche.me/e/b8d9)"
 
     warnings.warn(msg, exc.RemovedIn20Warning, stacklevel=stacklevel)
 
@@ -43,6 +43,23 @@ def deprecated_cls(version, message, constructor="__init__"):
     return decorate
 
 
+def deprecated_20_cls(clsname, alternative=None, constructor="__init__"):
+    message = (
+        ".. deprecated:: 2.0 The %s class is considered legacy as of the "
+        "1.x series of SQLAlchemy and will be removed in 2.0." % clsname
+    )
+
+    if alternative:
+        message += " " + alternative
+
+    def decorate(cls):
+        return _decorate_cls_with_warning(
+            cls, constructor, exc.RemovedIn20Warning, message, message
+        )
+
+    return decorate
+
+
 def deprecated(
     version, message=None, add_deprecation_to_docstring=True, warning=None
 ):
@@ -83,15 +100,13 @@ def deprecated(
 
 def deprecated_20(api_name, alternative=None, **kw):
     message = (
-        "The %s() function/method is considered legacy as of the "
+        "The %s function/method is considered legacy as of the "
         "1.x series of SQLAlchemy and will be removed in 2.0." % api_name
     )
 
     if alternative:
         message += " " + alternative
 
-    message += " (Background on SQLAlchemy 2.0 at: http://sqlalche.me/e/b8d9)"
-
     return deprecated(
         "2.0", message=message, warning=exc.RemovedIn20Warning, **kw
     )
@@ -194,25 +209,36 @@ def _decorate_cls_with_warning(
 ):
     doc = cls.__doc__ is not None and cls.__doc__ or ""
     if docstring_header is not None:
-        docstring_header %= dict(func=constructor)
 
+        if constructor is not None:
+            docstring_header %= dict(func=constructor)
+
+        if issubclass(wtype, exc.RemovedIn20Warning):
+            docstring_header += (
+                " (Background on SQLAlchemy 2.0 at: "
+                ":ref:`migration_20_toplevel`)"
+            )
         doc = inject_docstring_text(doc, docstring_header, 1)
 
         if type(cls) is type:
             clsdict = dict(cls.__dict__)
             clsdict["__doc__"] = doc
+            clsdict.pop("__dict__", None)
             cls = type(cls.__name__, cls.__bases__, clsdict)
-            constructor_fn = clsdict[constructor]
+            if constructor is not None:
+                constructor_fn = clsdict[constructor]
+
         else:
             cls.__doc__ = doc
-            constructor_fn = getattr(cls, constructor)
-
-    setattr(
-        cls,
-        constructor,
-        _decorate_with_warning(constructor_fn, wtype, message, None),
-    )
-
+            if constructor is not None:
+                constructor_fn = getattr(cls, constructor)
+
+        if constructor is not None:
+            setattr(
+                cls,
+                constructor,
+                _decorate_with_warning(constructor_fn, wtype, message, None),
+            )
     return cls
 
 
@@ -221,17 +247,30 @@ def _decorate_with_warning(func, wtype, message, docstring_header=None):
 
     message = _sanitize_restructured_text(message)
 
+    if issubclass(wtype, exc.RemovedIn20Warning):
+        doc_only = (
+            " (Background on SQLAlchemy 2.0 at: "
+            ":ref:`migration_20_toplevel`)"
+        )
+        warning_only = (
+            " (Background on SQLAlchemy 2.0 at: http://sqlalche.me/e/b8d9)"
+        )
+    else:
+        doc_only = warning_only = ""
+
     @decorator
     def warned(fn, *args, **kwargs):
         skip_warning = kwargs.pop("_sa_skip_warning", False)
         if not skip_warning:
-            warnings.warn(message, wtype, stacklevel=3)
+            warnings.warn(message + warning_only, wtype, stacklevel=3)
         return fn(*args, **kwargs)
 
     doc = func.__doc__ is not None and func.__doc__ or ""
     if docstring_header is not None:
         docstring_header %= dict(func=func.__name__)
 
+        docstring_header += doc_only
+
         doc = inject_docstring_text(doc, docstring_header, 1)
 
     decorated = warned(func)