Modernize internal reflection

- Deprecated remaining engine-level introspection and utility methods
including :meth:`.Engine.run_callable`, :meth:`.Engine.transaction`,
:meth:`.Engine.table_names`, :meth:`.Engine.has_table`.   The utility
methods are superseded by modern context-manager patterns, and the table
introspection tasks are suited by the :class:`.Inspector` object.

- The internal dialect method ``Dialect.reflecttable`` has been removed.  A
review of third party dialects has not found any making use of this method,
as it was already documented as one that should not be used by external
dialects.  Additionally, the private ``Engine._run_visitor`` method
is also removed.

- The long-deprecated ``Inspector.get_table_names.order_by`` parameter has
been removed.

- The :paramref:`.Table.autoload_with` parameter now accepts an :class:`.Inspector` object
directly, as well as any :class:`.Engine` or :class:`.Connection` as was the case before.

Fixes: #4755
Change-Id: Iec3a8b0f3e298ba87d532b16fac1e1132f464e21

7 files changed, 124 insertions, 129 deletions

diff --git a/lib/sqlalchemy/engine/base.py b/lib/sqlalchemy/engine/base.py
index 322db8415..a708643b3 100644
--- a/lib/sqlalchemy/engine/base.py
+++ b/lib/sqlalchemy/engine/base.py
@@ -13,6 +13,7 @@ from .interfaces import Connectable
 from .interfaces import ExceptionContext
 from .util import _distill_params
 from .. import exc
+from .. import inspection
 from .. import interfaces
 from .. import log
 from .. import util
@@ -1551,6 +1552,21 @@ class Connection(Connectable):
         else:
             util.reraise(*exc_info)
 
+    def _run_ddl_visitor(self, visitorcallable, element, **kwargs):
+        """run a DDL visitor.
+
+        This method is only here so that the MockConnection can change the
+        options given to the visitor so that "checkfirst" is skipped.
+
+        """
+        visitorcallable(self.dialect, self, **kwargs).traverse_single(element)
+
+    @util.deprecated(
+        "1.4",
+        "The :meth:`.Connection.transaction` method is deprecated and will be "
+        "removed in a future release.  Use the :meth:`.Engine.begin` "
+        "context manager instead.",
+    )
     def transaction(self, callable_, *args, **kwargs):
         r"""Execute the given function within a transaction boundary.
 
@@ -1593,6 +1609,7 @@ class Connection(Connectable):
 
         """
 
+        kwargs["_sa_skip_warning"] = True
         trans = self.begin()
         try:
             ret = self.run_callable(callable_, *args, **kwargs)
@@ -1602,6 +1619,11 @@ class Connection(Connectable):
             with util.safe_reraise():
                 trans.rollback()
 
+    @util.deprecated(
+        "1.4",
+        "The :meth:`.Connection.run_callable` method is deprecated and will "
+        "be removed in a future release.  Use a context manager instead.",
+    )
     def run_callable(self, callable_, *args, **kwargs):
         r"""Given a callable object or function, execute it, passing
         a :class:`.Connection` as the first argument.
@@ -1617,9 +1639,6 @@ class Connection(Connectable):
         """
         return callable_(self, *args, **kwargs)
 
-    def _run_visitor(self, visitorcallable, element, **kwargs):
-        visitorcallable(self.dialect, self, **kwargs).traverse_single(element)
-
 
 class ExceptionContextImpl(ExceptionContext):
     """Implement the :class:`.ExceptionContext` interface."""
@@ -2059,12 +2078,6 @@ class Engine(Connectable, log.Identified):
         else:
             yield connection
 
-    def _run_visitor(
-        self, visitorcallable, element, connection=None, **kwargs
-    ):
-        with self._optional_conn_ctx_manager(connection) as conn:
-            conn._run_visitor(visitorcallable, element, **kwargs)
-
     class _trans_ctx(object):
         def __init__(self, conn, transaction, close_with_result):
             self.conn = conn
@@ -2121,6 +2134,12 @@ class Engine(Connectable, log.Identified):
                 conn.close()
         return Engine._trans_ctx(conn, trans, close_with_result)
 
+    @util.deprecated(
+        "1.4",
+        "The :meth:`.Engine.transaction` method is deprecated and will be "
+        "removed in a future release.  Use the :meth:`.Engine.begin` context "
+        "manager instead.",
+    )
     def transaction(self, callable_, *args, **kwargs):
         r"""Execute the given function within a transaction boundary.
 
@@ -2159,10 +2178,16 @@ class Engine(Connectable, log.Identified):
             :meth:`.Engine.transaction`
 
         """
-
+        kwargs["_sa_skip_warning"] = True
         with self.connect() as conn:
             return conn.transaction(callable_, *args, **kwargs)
 
+    @util.deprecated(
+        "1.4",
+        "The :meth:`.Engine.run_callable` method is deprecated and will be "
+        "removed in a future release.  Use the :meth:`.Engine.connect` "
+        "context manager instead.",
+    )
     def run_callable(self, callable_, *args, **kwargs):
         r"""Given a callable object or function, execute it, passing
         a :class:`.Connection` as the first argument.
@@ -2176,9 +2201,14 @@ class Engine(Connectable, log.Identified):
         which one is being dealt with.
 
         """
+        kwargs["_sa_skip_warning"] = True
         with self.connect() as conn:
             return conn.run_callable(callable_, *args, **kwargs)
 
+    def _run_ddl_visitor(self, visitorcallable, element, **kwargs):
+        with self.connect() as conn:
+            conn._run_ddl_visitor(visitorcallable, element, **kwargs)
+
     def execute(self, statement, *multiparams, **params):
         """Executes the given construct and returns a :class:`.ResultProxy`.
 
@@ -2225,6 +2255,12 @@ class Engine(Connectable, log.Identified):
 
         return self._connection_cls(self, close_with_result=close_with_result)
 
+    @util.deprecated(
+        "1.4",
+        "The :meth:`.Engine.table_names` method is deprecated and will be "
+        "removed in a future release.  Please refer to "
+        ":meth:`.Inspector.get_table_names`.",
+    )
     def table_names(self, schema=None, connection=None):
         """Return a list of all table names available in the database.
 
@@ -2232,12 +2268,16 @@ class Engine(Connectable, log.Identified):
 
         :param connection: Optional, use a specified connection.
         """
-
         with self._optional_conn_ctx_manager(connection) as conn:
-            if not schema:
-                schema = self.dialect.default_schema_name
-            return self.dialect.get_table_names(conn, schema)
-
+            insp = inspection.inspect(conn)
+            return insp.get_table_names(schema)
+
+    @util.deprecated(
+        "1.4",
+        "The :meth:`.Engine.has_table` method is deprecated and will be "
+        "removed in a future release.  Please refer to "
+        ":meth:`.Inspector.has_table`.",
+    )
     def has_table(self, table_name, schema=None):
         """Return True if the given backend has a table of the given name.
 
@@ -2250,7 +2290,9 @@ class Engine(Connectable, log.Identified):
             with a schema identifier.
 
         """
-        return self.run_callable(self.dialect.has_table, table_name, schema)
+        with self._optional_conn_ctx_manager(None) as conn:
+            insp = inspection.inspect(conn)
+            return insp.has_table(table_name, schema=schema)
 
     def _wrap_pool_connect(self, fn, connection):
         dialect = self.dialect
diff --git a/lib/sqlalchemy/engine/default.py b/lib/sqlalchemy/engine/default.py
index df5794dd6..9d457b800 100644
--- a/lib/sqlalchemy/engine/default.py
+++ b/lib/sqlalchemy/engine/default.py
@@ -19,7 +19,6 @@ import re
 import weakref
 
 from . import interfaces
-from . import reflection
 from . import result
 from .. import event
 from .. import exc
@@ -416,20 +415,6 @@ class DefaultDialect(interfaces.Dialect):
         """
         return sqltypes.adapt_type(typeobj, self.colspecs)
 
-    def reflecttable(
-        self,
-        connection,
-        table,
-        include_columns,
-        exclude_columns,
-        resolve_fks,
-        **opts
-    ):
-        insp = reflection.Inspector.from_engine(connection)
-        return insp.reflecttable(
-            table, include_columns, exclude_columns, resolve_fks, **opts
-        )
-
     def get_pk_constraint(self, conn, table_name, schema=None, **kw):
         """Compatibility method, adapts the result of get_primary_keys()
         for those dialects which don't implement get_pk_constraint().
diff --git a/lib/sqlalchemy/engine/interfaces.py b/lib/sqlalchemy/engine/interfaces.py
index 4a63e3d84..5bd3b1d3e 100644
--- a/lib/sqlalchemy/engine/interfaces.py
+++ b/lib/sqlalchemy/engine/interfaces.py
@@ -196,28 +196,6 @@ class Dialect(object):
 
         pass
 
-    def reflecttable(
-        self, connection, table, include_columns, exclude_columns, resolve_fks
-    ):
-        """Load table description from the database.
-
-        Given a :class:`.Connection` and a
-        :class:`~sqlalchemy.schema.Table` object, reflect its columns and
-        properties from the database.
-
-        The implementation of this method is provided by
-        :meth:`.DefaultDialect.reflecttable`, which makes use of
-        :class:`.Inspector` to retrieve column information.
-
-        Dialects should **not** seek to implement this method, and should
-        instead implement individual schema inspection operations such as
-        :meth:`.Dialect.get_columns`, :meth:`.Dialect.get_pk_constraint`,
-        etc.
-
-        """
-
-        raise NotImplementedError()
-
     def get_columns(self, connection, table_name, schema=None, **kw):
         """Return information about columns in `table_name`.
 
@@ -450,7 +428,7 @@ class Dialect(object):
         """
         raise NotImplementedError()
 
-    def has_table(self, connection, table_name, schema=None):
+    def has_table(self, connection, table_name, schema=None, **kw):
         """Check the existence of a particular table in the database.
 
         Given a :class:`.Connection` object and a string
@@ -461,7 +439,7 @@ class Dialect(object):
 
         raise NotImplementedError()
 
-    def has_sequence(self, connection, sequence_name, schema=None):
+    def has_sequence(self, connection, sequence_name, schema=None, **kw):
         """Check the existence of a particular sequence in the database.
 
         Given a :class:`.Connection` object and a string
diff --git a/lib/sqlalchemy/engine/mock.py b/lib/sqlalchemy/engine/mock.py
index 439a0ecf6..1ba6881a3 100644
--- a/lib/sqlalchemy/engine/mock.py
+++ b/lib/sqlalchemy/engine/mock.py
@@ -48,7 +48,7 @@ class MockConnection(base.Connectable):
 
         ddl.SchemaDropper(self.dialect, self, **kwargs).traverse_single(entity)
 
-    def _run_visitor(
+    def _run_ddl_visitor(
         self, visitorcallable, element, connection=None, **kwargs
     ):
         kwargs["checkfirst"] = False
diff --git a/lib/sqlalchemy/engine/reflection.py b/lib/sqlalchemy/engine/reflection.py
index dcd1d0313..872a21bb7 100644
--- a/lib/sqlalchemy/engine/reflection.py
+++ b/lib/sqlalchemy/engine/reflection.py
@@ -58,6 +58,7 @@ def cache(fn, self, con, *args, **kw):
     return ret
 
 
+@inspection._self_inspects
 class Inspector(object):
     """Performs database schema inspection.
 
@@ -161,17 +162,7 @@ class Inspector(object):
             )
         return []
 
-    @util.deprecated_params(
-        order_by=(
-            "1.0",
-            "The :paramref:`get_table_names.order_by` parameter is deprecated "
-            "and will be removed in a future release.  Please refer to "
-            ":meth:`.Inspector.get_sorted_table_and_fkc_names` for a "
-            "more comprehensive solution to resolving foreign key cycles "
-            "between tables.",
-        )
-    )
-    def get_table_names(self, schema=None, order_by=None):
+    def get_table_names(self, schema=None):
         """Return all table names in referred to within a particular schema.
 
         The names are expected to be real tables only, not views.
@@ -198,20 +189,18 @@ class Inspector(object):
 
         """
 
-        if hasattr(self.dialect, "get_table_names"):
-            tnames = self.dialect.get_table_names(
-                self.bind, schema, info_cache=self.info_cache
-            )
-        else:
-            tnames = self.engine.table_names(schema)
-        if order_by == "foreign_key":
-            tuples = []
-            for tname in tnames:
-                for fkey in self.get_foreign_keys(tname, schema):
-                    if tname != fkey["referred_table"]:
-                        tuples.append((fkey["referred_table"], tname))
-            tnames = list(topological.sort(tuples, tnames))
-        return tnames
+        return self.dialect.get_table_names(
+            self.bind, 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.
+
+        .. versionadded:: 1.4
+
+        """
+        # TODO: info_cache?
+        return self.dialect.has_table(self.bind, table_name, schema)
 
     def get_sorted_table_and_fkc_names(self, schema=None):
         """Return dependency-sorted table and foreign key constraint names in
diff --git a/lib/sqlalchemy/sql/schema.py b/lib/sqlalchemy/sql/schema.py
index beebc5b1f..091012896 100644
--- a/lib/sqlalchemy/sql/schema.py
+++ b/lib/sqlalchemy/sql/schema.py
@@ -240,13 +240,14 @@ class Table(DialectKWArgs, SchemaItem, TableClause):
 
             :paramref:`.Table.extend_existing`
 
-    :param autoload_with: An :class:`.Engine` or :class:`.Connection` object
-        with which this :class:`.Table` object will be reflected; when
-        set to a non-None value, it implies that :paramref:`.Table.autoload`
-        is ``True``.   If left unset, but :paramref:`.Table.autoload` is
-        explicitly set to ``True``, an autoload operation will attempt to
-        proceed by locating an :class:`.Engine` or :class:`.Connection` bound
-        to the underlying :class:`.MetaData` object.
+    :param autoload_with: An :class:`.Engine` or :class:`.Connection` object,
+        or a :class:`.Inspector` object as returned by :func:`.inspect`
+        against one, with which this :class:`.Table` object will be reflected.
+        When set to a non-None value, it implies that
+        :paramref:`.Table.autoload` is ``True``.   If left unset, but
+        :paramref:`.Table.autoload` is explicitly set to ``True``, an autoload
+        operation will attempt to proceed by locating an :class:`.Engine` or
+        :class:`.Connection` bound to the underlying :class:`.MetaData` object.
 
         .. seealso::
 
@@ -598,34 +599,23 @@ class Table(DialectKWArgs, SchemaItem, TableClause):
         resolve_fks=True,
         _extend_on=None,
     ):
-
-        if autoload_with:
-            autoload_with.run_callable(
-                autoload_with.dialect.reflecttable,
-                self,
-                include_columns,
-                exclude_columns,
-                resolve_fks,
-                _extend_on=_extend_on,
-            )
-        else:
-            bind = _bind_or_error(
+        if autoload_with is None:
+            autoload_with = _bind_or_error(
                 metadata,
                 msg="No engine is bound to this Table's MetaData. "
                 "Pass an engine to the Table via "
-                "autoload_with=<someengine>, "
-                "or associate the MetaData with an engine via "
-                "metadata.bind=<someengine>",
-            )
-            bind.run_callable(
-                bind.dialect.reflecttable,
-                self,
-                include_columns,
-                exclude_columns,
-                resolve_fks,
-                _extend_on=_extend_on,
+                "autoload_with=<someengine_or_connection>",
             )
 
+        insp = inspection.inspect(autoload_with)
+        insp.reflecttable(
+            self,
+            include_columns,
+            exclude_columns,
+            resolve_fks,
+            _extend_on=_extend_on,
+        )
+
     @property
     def _sorted_constraints(self):
         """Return the set of constraints as a list, sorted by creation
@@ -834,15 +824,20 @@ class Table(DialectKWArgs, SchemaItem, TableClause):
             else:
                 return []
 
+    @util.deprecated(
+        "1.4",
+        "The :meth:`.Table.exists` method is deprecated and will be "
+        "removed in a future release.  Please refer to "
+        ":meth:`.Inspector.has_table`.",
+    )
     def exists(self, bind=None):
         """Return True if this table exists."""
 
         if bind is None:
             bind = _bind_or_error(self)
 
-        return bind.run_callable(
-            bind.dialect.has_table, self.name, schema=self.schema
-        )
+        insp = inspection.inspect(bind)
+        return insp.has_table(self.name, schema=self.schema)
 
     def create(self, bind=None, checkfirst=False):
         """Issue a ``CREATE`` statement for this
@@ -857,7 +852,7 @@ class Table(DialectKWArgs, SchemaItem, TableClause):
 
         if bind is None:
             bind = _bind_or_error(self)
-        bind._run_visitor(ddl.SchemaGenerator, self, checkfirst=checkfirst)
+        bind._run_ddl_visitor(ddl.SchemaGenerator, self, checkfirst=checkfirst)
 
     def drop(self, bind=None, checkfirst=False):
         """Issue a ``DROP`` statement for this
@@ -871,7 +866,7 @@ class Table(DialectKWArgs, SchemaItem, TableClause):
         """
         if bind is None:
             bind = _bind_or_error(self)
-        bind._run_visitor(ddl.SchemaDropper, self, checkfirst=checkfirst)
+        bind._run_ddl_visitor(ddl.SchemaDropper, self, checkfirst=checkfirst)
 
     def tometadata(
         self,
@@ -2523,14 +2518,14 @@ class Sequence(roles.StatementRole, DefaultGenerator):
 
         if bind is None:
             bind = _bind_or_error(self)
-        bind._run_visitor(ddl.SchemaGenerator, self, checkfirst=checkfirst)
+        bind._run_ddl_visitor(ddl.SchemaGenerator, self, checkfirst=checkfirst)
 
     def drop(self, bind=None, checkfirst=True):
         """Drops this sequence from the database."""
 
         if bind is None:
             bind = _bind_or_error(self)
-        bind._run_visitor(ddl.SchemaDropper, self, checkfirst=checkfirst)
+        bind._run_ddl_visitor(ddl.SchemaDropper, self, checkfirst=checkfirst)
 
     def _not_a_column_expr(self):
         raise exc.InvalidRequestError(
@@ -3687,7 +3682,7 @@ class Index(DialectKWArgs, ColumnCollectionMixin, SchemaItem):
         """
         if bind is None:
             bind = _bind_or_error(self)
-        bind._run_visitor(ddl.SchemaGenerator, self)
+        bind._run_ddl_visitor(ddl.SchemaGenerator, self)
         return self
 
     def drop(self, bind=None):
@@ -3702,7 +3697,7 @@ class Index(DialectKWArgs, ColumnCollectionMixin, SchemaItem):
         """
         if bind is None:
             bind = _bind_or_error(self)
-        bind._run_visitor(ddl.SchemaDropper, self)
+        bind._run_ddl_visitor(ddl.SchemaDropper, self)
 
     def __repr__(self):
         return "Index(%s)" % (
@@ -4169,10 +4164,10 @@ class MetaData(SchemaItem):
             bind = _bind_or_error(self)
 
         with bind.connect() as conn:
+            insp = inspection.inspect(conn)
 
             reflect_opts = {
-                "autoload": True,
-                "autoload_with": conn,
+                "autoload_with": insp,
                 "extend_existing": extend_existing,
                 "autoload_replace": autoload_replace,
                 "resolve_fks": resolve_fks,
@@ -4187,11 +4182,9 @@ class MetaData(SchemaItem):
             if schema is not None:
                 reflect_opts["schema"] = schema
 
-            available = util.OrderedSet(
-                bind.engine.table_names(schema, connection=conn)
-            )
+            available = util.OrderedSet(insp.get_table_names(schema))
             if views:
-                available.update(bind.dialect.get_view_names(conn, schema))
+                available.update(insp.get_view_names(schema))
 
             if schema is not None:
                 available_w_schema = util.OrderedSet(
@@ -4275,7 +4268,7 @@ class MetaData(SchemaItem):
         """
         if bind is None:
             bind = _bind_or_error(self)
-        bind._run_visitor(
+        bind._run_ddl_visitor(
             ddl.SchemaGenerator, self, checkfirst=checkfirst, tables=tables
         )
 
@@ -4301,11 +4294,17 @@ class MetaData(SchemaItem):
         """
         if bind is None:
             bind = _bind_or_error(self)
-        bind._run_visitor(
+        bind._run_ddl_visitor(
             ddl.SchemaDropper, self, checkfirst=checkfirst, tables=tables
         )
 
 
+@util.deprecated_cls(
+    "1.4",
+    ":class:`.ThreadLocalMetaData` is deprecated and will be removed "
+    "in a future release.",
+    constructor="__init__",
+)
 class ThreadLocalMetaData(MetaData):
     """A MetaData variant that presents a different ``bind`` in every thread.
 
diff --git a/lib/sqlalchemy/util/deprecations.py b/lib/sqlalchemy/util/deprecations.py
index 9af800591..76b0cf56d 100644
--- a/lib/sqlalchemy/util/deprecations.py
+++ b/lib/sqlalchemy/util/deprecations.py
@@ -233,7 +233,9 @@ def _decorate_with_warning(func, wtype, message, docstring_header=None):
 
     @decorator
     def warned(fn, *args, **kwargs):
-        warnings.warn(message, wtype, stacklevel=3)
+        skip_warning = kwargs.pop("_sa_skip_warning", False)
+        if not skip_warning:
+            warnings.warn(message, wtype, stacklevel=3)
         return fn(*args, **kwargs)
 
     doc = func.__doc__ is not None and func.__doc__ or ""