Add resolve_fks=False option for reflection

Added new parameters :paramref:`.Table.resolve_fks` and
:paramref:`.MetaData.reflect.resolve_fks` which when set to False will
disable the automatic reflection of related tables encountered in
:class:`.ForeignKey` objects, which can both reduce SQL overhead for omitted
tables as well as avoid tables that can't be reflected for database-specific
reasons.  Two :class:`.Table` objects present in the same :class:`.MetaData`
collection can still refer to each other even if the reflection of the two
tables occurred separately.

Fixes: #4517
Change-Id: I623baed42042a16c5109e4c8af6b2f64d2d00f95

6 files changed, 210 insertions, 23 deletions

diff --git a/doc/build/changelog/unreleased_13/4517.rst b/doc/build/changelog/unreleased_13/4517.rst
new file mode 100644
index 000000000..308d61f07
--- /dev/null
+++ b/doc/build/changelog/unreleased_13/4517.rst
@@ -0,0 +1,13 @@
+.. change::
+   :tags: feature, schema
+   :tickets: 4517
+
+   Added new parameters :paramref:`.Table.resolve_fks` and
+   :paramref:`.MetaData.reflect.resolve_fks` which when set to False will
+   disable the automatic reflection of related tables encountered in
+   :class:`.ForeignKey` objects, which can both reduce SQL overhead for omitted
+   tables as well as avoid tables that can't be reflected for database-specific
+   reasons.  Two :class:`.Table` objects present in the same :class:`.MetaData`
+   collection can still refer to each other even if the reflection of the two
+   tables occurred separately.
+
diff --git a/lib/sqlalchemy/engine/default.py b/lib/sqlalchemy/engine/default.py
index 58c09a3ac..51e2c4603 100644
--- a/lib/sqlalchemy/engine/default.py
+++ b/lib/sqlalchemy/engine/default.py
@@ -418,11 +418,17 @@ class DefaultDialect(interfaces.Dialect):
         return sqltypes.adapt_type(typeobj, self.colspecs)
 
     def reflecttable(
-        self, connection, table, include_columns, exclude_columns, **opts
+        self,
+        connection,
+        table,
+        include_columns,
+        exclude_columns,
+        resolve_fks,
+        **opts
     ):
         insp = reflection.Inspector.from_engine(connection)
         return insp.reflecttable(
-            table, include_columns, exclude_columns, **opts
+            table, include_columns, exclude_columns, resolve_fks, **opts
         )
 
     def get_pk_constraint(self, conn, table_name, schema=None, **kw):
diff --git a/lib/sqlalchemy/engine/interfaces.py b/lib/sqlalchemy/engine/interfaces.py
index 3142c0c6b..4806e72a5 100644
--- a/lib/sqlalchemy/engine/interfaces.py
+++ b/lib/sqlalchemy/engine/interfaces.py
@@ -197,7 +197,7 @@ class Dialect(object):
         pass
 
     def reflecttable(
-        self, connection, table, include_columns, exclude_columns
+        self, connection, table, include_columns, exclude_columns, resolve_fks
     ):
         """Load table description from the database.
 
diff --git a/lib/sqlalchemy/engine/reflection.py b/lib/sqlalchemy/engine/reflection.py
index 14d647b9a..888149226 100644
--- a/lib/sqlalchemy/engine/reflection.py
+++ b/lib/sqlalchemy/engine/reflection.py
@@ -561,7 +561,12 @@ class Inspector(object):
         )
 
     def reflecttable(
-        self, table, include_columns, exclude_columns=(), _extend_on=None
+        self,
+        table,
+        include_columns,
+        exclude_columns=(),
+        resolve_fks=True,
+        _extend_on=None,
     ):
         """Given a Table object, load its internal constructs based on
         introspection.
@@ -650,6 +655,7 @@ class Inspector(object):
             table,
             cols_by_orig_name,
             exclude_columns,
+            resolve_fks,
             _extend_on,
             reflection_options,
         )
@@ -783,6 +789,7 @@ class Inspector(object):
         table,
         cols_by_orig_name,
         exclude_columns,
+        resolve_fks,
         _extend_on,
         reflection_options,
     ):
@@ -806,29 +813,31 @@ class Inspector(object):
             referred_columns = fkey_d["referred_columns"]
             refspec = []
             if referred_schema is not None:
-                sa_schema.Table(
-                    referred_table,
-                    table.metadata,
-                    autoload=True,
-                    schema=referred_schema,
-                    autoload_with=self.bind,
-                    _extend_on=_extend_on,
-                    **reflection_options
-                )
+                if resolve_fks:
+                    sa_schema.Table(
+                        referred_table,
+                        table.metadata,
+                        autoload=True,
+                        schema=referred_schema,
+                        autoload_with=self.bind,
+                        _extend_on=_extend_on,
+                        **reflection_options
+                    )
                 for column in referred_columns:
                     refspec.append(
                         ".".join([referred_schema, referred_table, column])
                     )
             else:
-                sa_schema.Table(
-                    referred_table,
-                    table.metadata,
-                    autoload=True,
-                    autoload_with=self.bind,
-                    schema=sa_schema.BLANK_SCHEMA,
-                    _extend_on=_extend_on,
-                    **reflection_options
-                )
+                if resolve_fks:
+                    sa_schema.Table(
+                        referred_table,
+                        table.metadata,
+                        autoload=True,
+                        autoload_with=self.bind,
+                        schema=sa_schema.BLANK_SCHEMA,
+                        _extend_on=_extend_on,
+                        **reflection_options
+                    )
                 for column in referred_columns:
                     refspec.append(".".join([referred_table, column]))
             if "options" in fkey_d:
diff --git a/lib/sqlalchemy/sql/schema.py b/lib/sqlalchemy/sql/schema.py
index e981d7aed..0d3e96d8f 100644
--- a/lib/sqlalchemy/sql/schema.py
+++ b/lib/sqlalchemy/sql/schema.py
@@ -311,6 +311,24 @@ class Table(DialectKWArgs, SchemaItem, TableClause):
         ``Table`` object. Defaults to ``None`` which indicates all columns
         should be reflected.
 
+    :param resolve_fks: Whether or not to reflect :class:`.Table` objects
+        related to this one via :class:`.ForeignKey` objects, when
+        :paramref:`.Table.autoload` or :paramref:`.Table.autoload_with` is
+        specified.   Defaults to True.  Set to False to disable reflection of
+        related tables as :class:`.ForeignKey` objects are encountered; may be
+        used either to save on SQL calls or to avoid issues with related tables
+        that can't be accessed. Note that if a related table is already present
+        in the :class:`.MetaData` collection, or becomes present later, a
+        :class:`.ForeignKey` object associated with this :class:`.Table` will
+        resolve to that table normally.
+
+        .. versionadded:: 1.3
+
+        .. seealso::
+
+            :paramref:`.MetaData.reflect.resolve_fks`
+
+
     :param info: Optional data dictionary which will be populated into the
         :attr:`.SchemaItem.info` attribute of this object.
 
@@ -537,6 +555,7 @@ class Table(DialectKWArgs, SchemaItem, TableClause):
         kwargs.pop("autoload_replace", True)
         _extend_on = kwargs.pop("_extend_on", None)
 
+        resolve_fks = kwargs.pop("resolve_fks", True)
         include_columns = kwargs.pop("include_columns", None)
 
         self.implicit_returning = kwargs.pop("implicit_returning", True)
@@ -559,7 +578,11 @@ class Table(DialectKWArgs, SchemaItem, TableClause):
         # circular foreign keys
         if autoload:
             self._autoload(
-                metadata, autoload_with, include_columns, _extend_on=_extend_on
+                metadata,
+                autoload_with,
+                include_columns,
+                _extend_on=_extend_on,
+                resolve_fks=resolve_fks,
             )
 
         # initialize all the column, etc. objects.  done after reflection to
@@ -572,6 +595,7 @@ class Table(DialectKWArgs, SchemaItem, TableClause):
         autoload_with,
         include_columns,
         exclude_columns=(),
+        resolve_fks=True,
         _extend_on=None,
     ):
 
@@ -581,6 +605,7 @@ class Table(DialectKWArgs, SchemaItem, TableClause):
                 self,
                 include_columns,
                 exclude_columns,
+                resolve_fks,
                 _extend_on=_extend_on,
             )
         else:
@@ -597,6 +622,7 @@ class Table(DialectKWArgs, SchemaItem, TableClause):
                 self,
                 include_columns,
                 exclude_columns,
+                resolve_fks,
                 _extend_on=_extend_on,
             )
 
@@ -636,6 +662,8 @@ class Table(DialectKWArgs, SchemaItem, TableClause):
 
         include_columns = kwargs.pop("include_columns", None)
 
+        resolve_fks = kwargs.pop("resolve_fks", True)
+
         if include_columns is not None:
             for c in self.c:
                 if c.name not in include_columns:
@@ -666,6 +694,7 @@ class Table(DialectKWArgs, SchemaItem, TableClause):
                 autoload_with,
                 include_columns,
                 exclude_columns,
+                resolve_fks,
                 _extend_on=_extend_on,
             )
 
@@ -4065,6 +4094,7 @@ class MetaData(SchemaItem):
         only=None,
         extend_existing=False,
         autoload_replace=True,
+        resolve_fks=True,
         **dialect_kwargs
     ):
         r"""Load all available table definitions from the database.
@@ -4111,6 +4141,26 @@ class MetaData(SchemaItem):
 
           .. versionadded:: 0.9.1
 
+        :param resolve_fks: if True, reflect :class:`.Table` objects linked
+         to :class:`.ForeignKey` objects located in each :class:`.Table`.
+         For :meth:`.MetaData.reflect`, this has the effect of reflecting
+         related tables that might otherwise not be in the list of tables
+         being reflected, for example if the referenced table is in a
+         different schema or is omitted via the
+         :paramref:`.MetaData.reflect.only` parameter.  When False,
+         :class:`.ForeignKey` objects are not followed to the :class:`.Table`
+         in which they link, however if the related table is also part of the
+         list of tables that would be reflected in any case, the
+         :class:`.ForeignKey` object will still resolve to its related
+         :class:`.Table` after the :meth:`.MetaData.reflect` operation is
+         complete.   Defaults to True.
+
+         .. versionadded:: 1.3.0
+
+         .. seealso::
+
+            :paramref:`.Table.resolve_fks`
+
         :param \**dialect_kwargs: Additional keyword arguments not mentioned
          above are dialect specific, and passed in the form
          ``<dialectname>_<argname>``.  See the documentation regarding an
@@ -4133,6 +4183,7 @@ class MetaData(SchemaItem):
                 "autoload_with": conn,
                 "extend_existing": extend_existing,
                 "autoload_replace": autoload_replace,
+                "resolve_fks": resolve_fks,
                 "_extend_on": set(),
             }
 
diff --git a/test/engine/test_reflection.py b/test/engine/test_reflection.py
index 41f75b8f8..273cc7356 100644
--- a/test/engine/test_reflection.py
+++ b/test/engine/test_reflection.py
@@ -23,8 +23,10 @@ from sqlalchemy.testing import eq_
 from sqlalchemy.testing import eq_regex
 from sqlalchemy.testing import expect_warnings
 from sqlalchemy.testing import fixtures
+from sqlalchemy.testing import in_
 from sqlalchemy.testing import is_true
 from sqlalchemy.testing import mock
+from sqlalchemy.testing import not_in_
 from sqlalchemy.testing import skip
 from sqlalchemy.testing.schema import Column
 from sqlalchemy.testing.schema import Table
@@ -142,6 +144,112 @@ class ReflectionTest(fixtures.TestBase, ComparesTables):
         assert t1r.c.t2id.references(t2r.c.id)
         assert t1r.c.t3id.references(t3r.c.id)
 
+    @testing.provide_metadata
+    def test_resolve_fks_false_table(self):
+        meta = self.metadata
+        Table(
+            "t1",
+            meta,
+            Column("id", sa.Integer, primary_key=True),
+            Column("t2id", sa.Integer, sa.ForeignKey("t2.id")),
+            test_needs_fk=True,
+        )
+        Table(
+            "t2",
+            meta,
+            Column("id", sa.Integer, primary_key=True),
+            test_needs_fk=True,
+        )
+        meta.create_all()
+        meta2 = MetaData()
+        t1 = Table("t1", meta2, resolve_fks=False, autoload_with=testing.db)
+        in_("t1", meta2.tables)
+        not_in_("t2", meta2.tables)
+
+        assert_raises(
+            sa.exc.NoReferencedTableError,
+            lambda: list(t1.c.t2id.foreign_keys)[0].column,
+        )
+
+        t2 = Table("t2", meta2, autoload_with=testing.db)
+
+        # now it resolves
+        is_true(t1.c.t2id.references(t2.c.id))
+
+    @testing.provide_metadata
+    def test_resolve_fks_false_extend_existing(self):
+        meta = self.metadata
+        Table(
+            "t1",
+            meta,
+            Column("id", sa.Integer, primary_key=True),
+            Column("t2id", sa.Integer, sa.ForeignKey("t2.id")),
+            test_needs_fk=True,
+        )
+        Table(
+            "t2",
+            meta,
+            Column("id", sa.Integer, primary_key=True),
+            test_needs_fk=True,
+        )
+        meta.create_all()
+        meta2 = MetaData()
+        Table("t1", meta2)
+        in_("t1", meta2.tables)
+
+        t1 = Table(
+            "t1",
+            meta2,
+            resolve_fks=False,
+            autoload_with=testing.db,
+            extend_existing=True,
+        )
+        not_in_("t2", meta2.tables)
+
+        assert_raises(
+            sa.exc.NoReferencedTableError,
+            lambda: list(t1.c.t2id.foreign_keys)[0].column,
+        )
+
+        t2 = Table("t2", meta2, autoload_with=testing.db)
+
+        # now it resolves
+        is_true(t1.c.t2id.references(t2.c.id))
+
+    @testing.provide_metadata
+    def test_resolve_fks_false_metadata(self):
+        meta = self.metadata
+        Table(
+            "t1",
+            meta,
+            Column("id", sa.Integer, primary_key=True),
+            Column("t2id", sa.Integer, sa.ForeignKey("t2.id")),
+            test_needs_fk=True,
+        )
+        Table(
+            "t2",
+            meta,
+            Column("id", sa.Integer, primary_key=True),
+            test_needs_fk=True,
+        )
+        meta.create_all()
+        meta2 = MetaData()
+        meta2.reflect(testing.db, resolve_fks=False, only=["t1"])
+        in_("t1", meta2.tables)
+        not_in_("t2", meta2.tables)
+
+        t1 = meta2.tables["t1"]
+
+        assert_raises(
+            sa.exc.NoReferencedTableError,
+            lambda: list(t1.c.t2id.foreign_keys)[0].column,
+        )
+
+        meta2.reflect(testing.db, resolve_fks=False)
+
+        t2 = meta2.tables["t2"]
+        is_true(t1.c.t2id.references(t2.c.id))
+
     def test_nonexistent(self):
         meta = MetaData(testing.db)
         assert_raises(