implement include_name hook

Added new hook :paramref:`.EnvironmentContext.configure.include_name`,
which complements the
:paramref:`.EnvironmentContext.configure.include_object` hook by providing
a means of preventing objects of a certain name from being autogenerated
**before** the SQLAlchemy reflection process takes place, and notably
includes explicit support for passing each schema name when
:paramref:`.EnvironmentContext.configure.include_schemas` is set to True.
This is most important especially for enviroments that make use of
:paramref:`.EnvironmentContext.configure.include_schemas` where schemas are
actually databases (e.g. MySQL) in order to prevent reflection sweeps of
the entire server.

The long deprecated
:paramref:`.EnvironmentContext.configure.include_symbol` hook is removed.
The  :paramref:`.EnvironmentContext.configure.include_object`
and  :paramref:`.EnvironmentContext.configure.include_name`
hooks both achieve the goals of this hook.

Change-Id: Idd44a357088a79be94488fdd7a7841bf118d47e2
Fixes: #650

10 files changed, 784 insertions, 183 deletions

diff --git a/alembic/autogenerate/api.py b/alembic/autogenerate/api.py
index db5fe12..7411f9b 100644
--- a/alembic/autogenerate/api.py
+++ b/alembic/autogenerate/api.py
@@ -286,25 +286,18 @@ class AutogenContext(object):
                 % (migration_context.script.env_py_location)
             )
 
-        include_symbol = opts.get("include_symbol", None)
         include_object = opts.get("include_object", None)
+        include_name = opts.get("include_name", None)
 
         object_filters = []
-        if include_symbol:
-
-            def include_symbol_filter(
-                object_, name, type_, reflected, compare_to
-            ):
-                if type_ == "table":
-                    return include_symbol(name, object_.schema)
-                else:
-                    return True
-
-            object_filters.append(include_symbol_filter)
+        name_filters = []
         if include_object:
             object_filters.append(include_object)
+        if include_name:
+            name_filters.append(include_name)
 
         self._object_filters = object_filters
+        self._name_filters = name_filters
 
         self.migration_context = migration_context
         if self.migration_context is not None:
@@ -325,7 +318,40 @@ class AutogenContext(object):
         yield
         self._has_batch = False
 
-    def run_filters(self, object_, name, type_, reflected, compare_to):
+    def run_name_filters(self, name, type_, parent_names):
+        """Run the context's name filters and return True if the targets
+        should be part of the autogenerate operation.
+
+        This method should be run for every kind of name encountered within the
+        reflection side of an autogenerate operation, giving the environment
+        the chance to filter what names should be reflected as database
+        objects.  The filters here are produced directly via the
+        :paramref:`.EnvironmentContext.configure.include_name` parameter.
+
+        """
+
+        if "schema_name" in parent_names:
+            if type_ == "table":
+                table_name = name
+            else:
+                table_name = parent_names["table_name"]
+            schema_name = parent_names["schema_name"]
+            if schema_name:
+                parent_names["schema_qualified_table_name"] = "%s.%s" % (
+                    schema_name,
+                    table_name,
+                )
+            else:
+                parent_names["schema_qualified_table_name"] = table_name
+
+        for fn in self._name_filters:
+
+            if not fn(name, type_, parent_names):
+                return False
+        else:
+            return True
+
+    def run_object_filters(self, object_, name, type_, reflected, compare_to):
         """Run the context's object filters and return True if the targets
         should be part of the autogenerate operation.
 
@@ -333,9 +359,7 @@ class AutogenContext(object):
         an autogenerate operation, giving the environment the chance
         to filter what objects should be included in the comparison.
         The filters here are produced directly via the
-        :paramref:`.EnvironmentContext.configure.include_object`
-        and :paramref:`.EnvironmentContext.configure.include_symbol`
-        functions, if present.
+        :paramref:`.EnvironmentContext.configure.include_object` parameter.
 
         """
         for fn in self._object_filters:
@@ -344,6 +368,8 @@ class AutogenContext(object):
         else:
             return True
 
+    run_filters = run_object_filters
+
     @util.memoized_property
     def sorted_tables(self):
         """Return an aggregate of the :attr:`.MetaData.sorted_tables` collection(s).
diff --git a/alembic/autogenerate/compare.py b/alembic/autogenerate/compare.py
index b82225d..1b36565 100644
--- a/alembic/autogenerate/compare.py
+++ b/alembic/autogenerate/compare.py
@@ -47,6 +47,10 @@ def _produce_net_changes(autogen_context, upgrade_ops):
     else:
         schemas = [None]
 
+    schemas = {
+        s for s in schemas if autogen_context.run_name_filters(s, "schema", {})
+    }
+
     comparators.dispatch("schema", autogen_context.dialect.name)(
         autogen_context, upgrade_ops, schemas
     )
@@ -63,13 +67,20 @@ def _autogen_for_tables(autogen_context, upgrade_ops, schemas):
     )
     version_table = autogen_context.migration_context.version_table
 
-    for s in schemas:
-        tables = set(inspector.get_table_names(schema=s))
-        if s == version_table_schema:
+    for schema_name in schemas:
+        tables = set(inspector.get_table_names(schema=schema_name))
+        if schema_name == version_table_schema:
             tables = tables.difference(
                 [autogen_context.migration_context.version_table]
             )
-        conn_table_names.update(zip([s] * len(tables), tables))
+
+        conn_table_names.update(
+            (schema_name, tname)
+            for tname in tables
+            if autogen_context.run_name_filters(
+                tname, "table", {"schema_name": schema_name}
+            )
+        )
 
     metadata_table_names = OrderedSet(
         [(table.schema, table.name) for table in autogen_context.sorted_tables]
@@ -125,7 +136,7 @@ def _compare_tables(
     for s, tname in metadata_table_names.difference(conn_table_names):
         name = "%s.%s" % (s, tname) if s else tname
         metadata_table = tname_to_table[(s, tname)]
-        if autogen_context.run_filters(
+        if autogen_context.run_object_filters(
             metadata_table, tname, "table", False, None
         ):
             upgrade_ops.ops.append(
@@ -162,7 +173,7 @@ def _compare_tables(
                 # fmt: on
             )
             sqla_compat._reflect_table(inspector, t, None)
-        if autogen_context.run_filters(t, tname, "table", True, None):
+        if autogen_context.run_object_filters(t, tname, "table", True, None):
 
             modify_table_ops = ops.ModifyTableOps(tname, [], schema=s)
 
@@ -201,7 +212,7 @@ def _compare_tables(
         metadata_table = tname_to_table[(s, tname)]
         conn_table = existing_metadata.tables[name]
 
-        if autogen_context.run_filters(
+        if autogen_context.run_object_filters(
             metadata_table, tname, "table", False, conn_table
         ):
 
@@ -286,11 +297,17 @@ def _compare_columns(
     metadata_cols_by_name = dict(
         (c.name, c) for c in metadata_table.c if not c.system
     )
-    conn_col_names = dict((c.name, c) for c in conn_table.c)
+    conn_col_names = dict(
+        (c.name, c)
+        for c in conn_table.c
+        if autogen_context.run_name_filters(
+            c.name, "column", {"table_name": tname, "schema_name": schema}
+        )
+    )
     metadata_col_names = OrderedSet(sorted(metadata_cols_by_name))
 
     for cname in metadata_col_names.difference(conn_col_names):
-        if autogen_context.run_filters(
+        if autogen_context.run_object_filters(
             metadata_cols_by_name[cname], cname, "column", False, None
         ):
             modify_table_ops.ops.append(
@@ -303,7 +320,7 @@ def _compare_columns(
     for colname in metadata_col_names.intersection(conn_col_names):
         metadata_col = metadata_cols_by_name[colname]
         conn_col = conn_table.c[colname]
-        if not autogen_context.run_filters(
+        if not autogen_context.run_object_filters(
             metadata_col, colname, "column", False, conn_col
         ):
             continue
@@ -325,7 +342,7 @@ def _compare_columns(
     yield
 
     for cname in set(conn_col_names).difference(metadata_col_names):
-        if autogen_context.run_filters(
+        if autogen_context.run_object_filters(
             conn_table.c[cname], cname, "column", True, None
         ):
             modify_table_ops.ops.append(
@@ -471,6 +488,15 @@ def _compare_indexes_and_uniques(
                 # not being present
                 pass
             else:
+                conn_uniques = [
+                    uq
+                    for uq in conn_uniques
+                    if autogen_context.run_name_filters(
+                        uq["name"],
+                        "unique_constraint",
+                        {"table_name": tname, "schema_name": schema},
+                    )
+                ]
                 for uq in conn_uniques:
                     if uq.get("duplicates_index"):
                         unique_constraints_duplicate_unique_indexes = True
@@ -478,6 +504,16 @@ def _compare_indexes_and_uniques(
             conn_indexes = inspector.get_indexes(tname, schema=schema)
         except NotImplementedError:
             pass
+        else:
+            conn_indexes = [
+                ix
+                for ix in conn_indexes
+                if autogen_context.run_name_filters(
+                    ix["name"],
+                    "index",
+                    {"table_name": tname, "schema_name": schema},
+                )
+            ]
 
         # 2. convert conn-level objects from raw inspector records
         # into schema objects
@@ -578,7 +614,7 @@ def _compare_indexes_and_uniques(
 
     def obj_added(obj):
         if obj.is_index:
-            if autogen_context.run_filters(
+            if autogen_context.run_object_filters(
                 obj.const, obj.name, "index", False, None
             ):
                 modify_ops.ops.append(ops.CreateIndexOp.from_index(obj.const))
@@ -595,7 +631,7 @@ def _compare_indexes_and_uniques(
             if is_create_table or is_drop_table:
                 # unique constraints are created inline with table defs
                 return
-            if autogen_context.run_filters(
+            if autogen_context.run_object_filters(
                 obj.const, obj.name, "unique_constraint", False, None
             ):
                 modify_ops.ops.append(
@@ -615,7 +651,7 @@ def _compare_indexes_and_uniques(
                 # be sure what we're doing here
                 return
 
-            if autogen_context.run_filters(
+            if autogen_context.run_object_filters(
                 obj.const, obj.name, "index", True, None
             ):
                 modify_ops.ops.append(ops.DropIndexOp.from_index(obj.const))
@@ -627,7 +663,7 @@ def _compare_indexes_and_uniques(
                 # if the whole table is being dropped, we don't need to
                 # consider unique constraint separately
                 return
-            if autogen_context.run_filters(
+            if autogen_context.run_object_filters(
                 obj.const, obj.name, "unique_constraint", True, None
             ):
                 modify_ops.ops.append(
@@ -641,7 +677,7 @@ def _compare_indexes_and_uniques(
 
     def obj_changed(old, new, msg):
         if old.is_index:
-            if autogen_context.run_filters(
+            if autogen_context.run_object_filters(
                 new.const, new.name, "index", False, old.const
             ):
                 log.info(
@@ -653,7 +689,7 @@ def _compare_indexes_and_uniques(
                 modify_ops.ops.append(ops.DropIndexOp.from_index(old.const))
                 modify_ops.ops.append(ops.CreateIndexOp.from_index(new.const))
         else:
-            if autogen_context.run_filters(
+            if autogen_context.run_object_filters(
                 new.const, new.name, "unique_constraint", False, old.const
             ):
                 log.info(
@@ -1128,7 +1164,15 @@ def _compare_foreign_keys(
         if isinstance(fk, sa_schema.ForeignKeyConstraint)
     )
 
-    conn_fks = inspector.get_foreign_keys(tname, schema=schema)
+    conn_fks = [
+        fk
+        for fk in inspector.get_foreign_keys(tname, schema=schema)
+        if autogen_context.run_name_filters(
+            fk["name"],
+            "foreign_key_constraint",
+            {"table_name": tname, "schema_name": schema},
+        )
+    ]
 
     backend_reflects_fk_options = conn_fks and "options" in conn_fks[0]
 
@@ -1161,7 +1205,7 @@ def _compare_foreign_keys(
     )
 
     def _add_fk(obj, compare_to):
-        if autogen_context.run_filters(
+        if autogen_context.run_object_filters(
             obj.const, obj.name, "foreign_key_constraint", False, compare_to
         ):
             modify_table_ops.ops.append(
@@ -1177,7 +1221,7 @@ def _compare_foreign_keys(
             )
 
     def _remove_fk(obj, compare_to):
-        if autogen_context.run_filters(
+        if autogen_context.run_object_filters(
             obj.const, obj.name, "foreign_key_constraint", True, compare_to
         ):
             modify_table_ops.ops.append(
diff --git a/alembic/runtime/environment.py b/alembic/runtime/environment.py
index c9d623c..ed251b9 100644
--- a/alembic/runtime/environment.py
+++ b/alembic/runtime/environment.py
@@ -298,7 +298,7 @@ class EnvironmentContext(util.ModuleClsProxy):
         template_args=None,
         render_as_batch=False,
         target_metadata=None,
-        include_symbol=None,
+        include_name=None,
         include_object=None,
         include_schemas=False,
         process_revision_directives=None,
@@ -521,6 +521,49 @@ class EnvironmentContext(util.ModuleClsProxy):
 
             :paramref:`.EnvironmentContext.configure.compare_type`
 
+        :param include_name: A callable function which is given
+         the chance to return ``True`` or ``False`` for any database reflected
+         object based on its name, including database schema names when
+         the :paramref:`.EnvironmentContext.configure.include_schemas` flag
+         is set to ``True``.
+
+         The function accepts the following positional arguments:
+
+         * ``name``: the name of the object, such as schema name or table name.
+           Will be ``None`` when indicating the default schema name of the
+           database connection.
+         * ``type``: a string describing the type of object; currently
+           ``"schema"``, ``"table"``, ``"column"``, ``"index"``,
+           ``"unique_constraint"``, or ``"foreign_key_constraint"``
+         * ``parent_names``: a dictionary of "parent" object names, that are
+           relative to the name being given.  Keys in this dictionary may
+           include:  ``"schema_name"``, ``"table_name"``.
+
+         E.g.::
+
+            def include_name(name, type_, parent_names):
+                if type_ == "schema":
+                    return name in ["schema_one", "schema_two"]
+                else:
+                    return True
+
+            context.configure(
+                # ...
+                include_schemas = True,
+                include_name = include_name
+            )
+
+         .. versionadded:: 1.5
+
+         .. seealso::
+
+            :ref:`autogenerate_include_hooks`
+
+            :paramref:`.EnvironmentContext.configure.include_object`
+
+            :paramref:`.EnvironmentContext.configure.include_schemas`
+
+
         :param include_object: A callable function which is given
          the chance to return ``True`` or ``False`` for any object,
          indicating if the given object should be considered in the
@@ -539,14 +582,6 @@ class EnvironmentContext(util.ModuleClsProxy):
          * ``type``: a string describing the type of object; currently
            ``"table"``, ``"column"``, ``"index"``, ``"unique_constraint"``,
            or ``"foreign_key_constraint"``
-
-           .. versionadded:: 0.7.0 Support for indexes and unique constraints
-              within the
-              :paramref:`~.EnvironmentContext.configure.include_object` hook.
-
-           .. versionadded:: 0.7.1 Support for foreign keys within the
-              :paramref:`~.EnvironmentContext.configure.include_object` hook.
-
          * ``reflected``: ``True`` if the given object was produced based on
            table reflection, ``False`` if it's from a local :class:`.MetaData`
            object.
@@ -568,46 +603,25 @@ class EnvironmentContext(util.ModuleClsProxy):
                 include_object = include_object
             )
 
-         :paramref:`.EnvironmentContext.configure.include_object` can also
-         be used to filter on specific schemas to include or omit, when
-         the :paramref:`.EnvironmentContext.configure.include_schemas`
-         flag is set to ``True``.   The :attr:`.Table.schema` attribute
-         on each :class:`.Table` object reflected will indicate the name of the
-         schema from which the :class:`.Table` originates.
+         For the use case of omitting specific schemas from a target database
+         when :paramref:`.EnvironmentContext.configure.include_schemas` is
+         set to ``True``, the :attr:`~sqlalchemy.schema.Table.schema`
+         attribute can be checked for each :class:`~sqlalchemy.schema.Table`
+         object passed to the hook, however it is much more efficient
+         to filter on schemas before reflection of objects takes place
+         using the :paramref:`.EnvironmentContext.configure.include_name`
+         hook.
 
          .. versionadded:: 0.6.0
 
          .. seealso::
 
-            :paramref:`.EnvironmentContext.configure.include_schemas`
-
-        :param include_symbol: A callable function which, given a table name
-         and schema name (may be ``None``), returns ``True`` or ``False``,
-         indicating if the given table should be considered in the
-         autogenerate sweep.
-
-         .. deprecated:: 0.6.0
-            :paramref:`.EnvironmentContext.configure.include_symbol`
-            is superceded by the more generic
-            :paramref:`.EnvironmentContext.configure.include_object`
-            parameter.
-
-         E.g.::
+            :ref:`autogenerate_include_hooks`
 
-            def include_symbol(tablename, schema):
-                return tablename not in ("skip_table_one", "skip_table_two")
-
-            context.configure(
-                # ...
-                include_symbol = include_symbol
-            )
-
-         .. seealso::
+            :paramref:`.EnvironmentContext.configure.include_name`
 
             :paramref:`.EnvironmentContext.configure.include_schemas`
 
-            :paramref:`.EnvironmentContext.configure.include_object`
-
         :param render_as_batch: if True, commands which alter elements
          within a table will be placed under a ``with batch_alter_table():``
          directive, so that batch migrations will take place.
@@ -623,12 +637,16 @@ class EnvironmentContext(util.ModuleClsProxy):
          :meth:`~sqlalchemy.engine.reflection.Inspector.get_schema_names`
          method, and include all differences in tables found across all
          those schemas.  When using this option, you may want to also
-         use the :paramref:`.EnvironmentContext.configure.include_object`
-         option to specify a callable which
+         use the :paramref:`.EnvironmentContext.configure.include_name`
+         parameter to specify a callable which
          can filter the tables/schemas that get included.
 
          .. seealso::
 
+            :ref:`autogenerate_include_hooks`
+
+            :paramref:`.EnvironmentContext.configure.include_name`
+
             :paramref:`.EnvironmentContext.configure.include_object`
 
         :param render_item: Callable that can be used to override how
@@ -790,7 +808,7 @@ class EnvironmentContext(util.ModuleClsProxy):
             opts["template_args"].update(template_args)
         opts["transaction_per_migration"] = transaction_per_migration
         opts["target_metadata"] = target_metadata
-        opts["include_symbol"] = include_symbol
+        opts["include_name"] = include_name
         opts["include_object"] = include_object
         opts["include_schemas"] = include_schemas
         opts["render_as_batch"] = render_as_batch
diff --git a/docs/build/autogenerate.rst b/docs/build/autogenerate.rst
index 46dde52..4858f34 100644
--- a/docs/build/autogenerate.rst
+++ b/docs/build/autogenerate.rst
@@ -224,6 +224,212 @@ with the same schema/name combination, an error is raised.
   autogeneration of multiple :class:`~sqlalchemy.schema.MetaData`
   collections.
 
+.. _autogenerate_include_hooks:
+
+Controlling What to be Autogenerated
+------------------------------------
+
+The autogenerate process scans across all table objects within
+the database that is referred towards by the current database connection
+in use.
+
+The list of objects that are scanned in the target database connection include:
+
+* The "default" schema currently referred towards by the database connection.
+
+* If the :paramref:`.EnvironmentContext.configure.include_schemas` is set to
+  ``True``, all non-default "schemas", which are those names returned by the
+  :meth:`~sqlalchemy.engine.reflection.Inspector.get_schema_names` method of
+  :class:`~sqlalchemy.engine.reflection.Inspector`.    The SQLAlchemy document
+  :ref:`sqla:schema_table_schema_name` discusses the concept of a
+  "schema" in detail.
+
+* Within each "schema", all tables present are scanned using the
+  :meth:`~sqlalchemy.engine.reflection.Inspector.get_table_names` method of
+  :class:`~sqlalchemy.engine.reflection.Inspector`.
+
+* Within each "table", most sub-objects of the each
+  :class:`~sqlalchemy.schema.Table` construct are scanned, including columns
+  and some forms of constraints.  This process ultimately involves the use of
+  methods on :class:`~sqlalchemy.engine.reflection.Inspector` including
+  :meth:`~sqlalchemy.engine.reflection.Inspector.get_columns`,
+  :meth:`~sqlalchemy.engine.reflection.Inspector.get_indexes`,
+  :meth:`~sqlalchemy.engine.reflection.Inspector.get_unique_constraints`,
+  :meth:`~sqlalchemy.engine.reflection.Inspector.get_foreign_keys` (as of this
+  writing, CHECK constraints and primary key constraints are not yet included).
+
+Omitting Schema Names from the Autogenerate Process
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+As the above set of database objects are typically to be compared to the contents of
+a single :class:`~sqlalchemy.schema.MetaData` object, particularly when the
+:paramref:`.EnvironmentContext.configure.include_schemas` flag is enabled
+there is an important need to filter out unwanted "schemas", which for some
+database backends might be the list of all the databases present.   This
+filtering is best performed using the :paramref:`.EnvironmentContext.configure.include_name`
+hook, which provides for a callable that may return a boolean true/false
+indicating if a particular schema name should be included::
+
+    def include_name(name, type_, parent_names):
+        if type_ == "schema":
+            # note this will not include the default schema
+            return name in ["schema_one", "schema_two"]
+        else:
+            return True
+
+    context.configure(
+        # ...
+        include_schemas = True,
+        include_name = include_name
+    )
+
+Above, when the list of schema names is first retrieved, the names will be
+filtered through the above ``include_name`` function so that only schemas
+named ``"schema_one"`` and ``"schema_two"`` will be considered by the
+autogenerate process.
+
+In order to include **the default schema**, that is, the schema that is
+referred towards by the database connection **without** any explicit
+schema being specified, the name passed to the hook is ``None``.  To alter
+our above example to also include the default schema, we compare to
+``None`` as well::
+
+    def include_name(name, type_, parent_names):
+        if type_ == "schema":
+            # this **will* include the default schema
+            return name in [None, "schema_one", "schema_two"]
+        else:
+            return True
+
+    context.configure(
+        # ...
+        include_schemas = True,
+        include_name = include_name
+    )
+
+Omitting Table Names from the Autogenerate Process
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+The :paramref:`.EnvironmentContext.configure.include_name` hook is also
+most appropriate to limit the names of tables in the target database
+to be considered.  If a target database has many tables that are not
+part of the :class:`~sqlalchemy.schema.MetaData`, the autogenerate process
+will normally assume these are extraneous tables in the database to be
+dropped, and it will generate a :meth:`.Operations.drop_table` operation
+for each.  To prevent this, the :paramref:`.EnvironmentContext.configure.include_name`
+hook may be used to search for each name within the
+:attr:`~sqlalchemy.schema.MetaData.tables` collection of the
+:class:`~sqlalchemy.schema.MetaData` object and ensure names
+which aren't present are not included::
+
+    target_metadata = MyModel.metadata
+
+    def include_name(name, type_, parent_names):
+        if type_ == "table":
+            return name in target_metadata.tables
+        else:
+            return True
+
+    context.configure(
+        # ...
+        target_metadata = target_metadata,
+        include_name = include_name,
+        include_schemas = False
+    )
+
+The above example is limited to table names present in the default schema only.
+In order to search within a :class:`~sqlalchemy.schema.MetaData` collection for
+schema-qualified table names as well, a table present in the non
+default schema will be present under a name of the form
+``<schemaname>.<tablename>``.   The
+:paramref:`.EnvironmentContext.configure.include_name` hook will present
+this schema name on a per-tablename basis in the ``parent_names`` dictionary,
+using the key ``"schema_name"`` that refers to the name of the
+schema currently being considered, or ``None`` if the schema is the default
+schema of the database connection::
+
+    # example fragment
+
+    if parent_names["schema_name"] is None:
+        return name in target_metadata.tables
+    else:
+        # build out schema-qualified name explicitly...
+        return (
+            "%s.%s" % (parent_names["schema_name"], name) in
+            target_metadata.tables
+        )
+
+However more simply, the ``parent_names`` dictionary will also include
+the dot-concatenated name already constructed under the key
+``"schema_qualified_table_name"``, which will also be suitably formatted
+for tables in the default schema as well with the dot omitted.  So the
+full example of omitting tables with schema support may look like::
+
+    target_metadata = MyModel.metadata
+
+    def include_name(name, type_, parent_names):
+        if type == "schema":
+            return name in [None, "schema_one", "schema_two"]
+        elif type_ == "table":
+            # use schema_qualified_table_name directly
+            return (
+                parent_names["schema_qualified_table_name"] in
+                target_metadata.tables
+            )
+        else:
+            return True
+
+    context.configure(
+        # ...
+        target_metadata = target_metadata,
+        include_name = include_name,
+        include_schemas = True
+    )
+
+The ``parent_names`` dictionary will also include the key ``"table_name"``
+when the name being considered is that of a column or constraint object
+local to a particular table.
+
+The :paramref:`.EnvironmentContext.configure.include_name` hook only refers
+to **reflected** objects, and not those located within the target
+:class:`~sqlalchemy.schema.MetaData` collection.   For more fine-grained
+rules that include both :class:`~sqlalchemy.schema.MetaData` and reflected
+object, the :paramref:`.EnvironmentContext.configure.include_object` hook
+discussed in the next section is more appropriate.
+
+.. versionadded:: 1.5 added the :paramref:`.EnvironmentContext.configure.include_name`
+    hook.
+
+Omitting Based on Object
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+The :paramref:`.EnvironmentContext.configure.include_object` hook provides
+for object-level inclusion/exclusion rules based on the :class:`~sqlalchemy.schema.Table`
+object being reflected as well as the elements within it.   This hook can
+be used to limit objects both from the local :class:`~sqlalchemy.schema.MetaData`
+collection as well as from the target database.   The limitation is that when
+it reports on objects in the database, it will have fully reflected that object,
+which can be expensive if a large number of objects will be omitted.  The
+example below refers to a fine-grained rule that will skip changes on
+:class:`~sqlalchemy.schema.Column` objects that have a user-defined flag
+``skip_autogenerate`` placed into the :attr:`~sqlalchemy.schema.Column.info`
+dictionary::
+
+    def include_object(object, name, type_, reflected, compare_to):
+        if (type_ == "column" and
+            not reflected and
+            object.info.get("skip_autogenerate", False)):
+            return False
+        else:
+            return True
+
+    context.configure(
+        # ...
+        include_object = include_object
+    )
+
+
+
 Comparing and Rendering Types
 ------------------------------
 
diff --git a/docs/build/unreleased/650.rst b/docs/build/unreleased/650.rst
new file mode 100644
index 0000000..96daa0c
--- /dev/null
+++ b/docs/build/unreleased/650.rst
@@ -0,0 +1,29 @@
+.. change::
+    :tags: feature, autogenerate
+    :tickets: 650
+
+    Added new hook :paramref:`.EnvironmentContext.configure.include_name`,
+    which complements the
+    :paramref:`.EnvironmentContext.configure.include_object` hook by providing
+    a means of preventing objects of a certain name from being autogenerated
+    **before** the SQLAlchemy reflection process takes place, and notably
+    includes explicit support for passing each schema name when
+    :paramref:`.EnvironmentContext.configure.include_schemas` is set to True.
+    This is most important especially for enviroments that make use of
+    :paramref:`.EnvironmentContext.configure.include_schemas` where schemas are
+    actually databases (e.g. MySQL) in order to prevent reflection sweeps of
+    the entire server.
+
+    .. seealso::
+
+        :ref:`autogenerate_include_hooks` - new documentation section
+
+.. change::
+    :tags: removed, autogenerate
+
+    The long deprecated
+    :paramref:`.EnvironmentContext.configure.include_symbol` hook is removed.
+    The  :paramref:`.EnvironmentContext.configure.include_object`
+    and  :paramref:`.EnvironmentContext.configure.include_name`
+    hooks both achieve the goals of this hook.
+
diff --git a/tests/_autogen_fixtures.py b/tests/_autogen_fixtures.py
index e81eb5d..f8916fa 100644
--- a/tests/_autogen_fixtures.py
+++ b/tests/_autogen_fixtures.py
@@ -42,6 +42,8 @@ def _default_include_object(obj, name, type_, reflected, compare_to):
 
 _default_object_filters = _default_include_object
 
+_default_name_filters = None
+
 
 class ModelOne(object):
     __requires__ = ("unique_constraint_reflection",)
@@ -235,6 +237,7 @@ class AutogenTest(_ComparesFKs):
             "alembic_module_prefix": "op.",
             "sqlalchemy_module_prefix": "sa.",
             "include_object": _default_object_filters,
+            "include_name": _default_name_filters,
         }
         if self.configure_opts:
             ctx_opts.update(self.configure_opts)
@@ -247,11 +250,15 @@ class AutogenTest(_ComparesFKs):
     def tearDown(self):
         self.conn.close()
 
-    def _update_context(self, object_filters=None, include_schemas=None):
+    def _update_context(
+        self, object_filters=None, name_filters=None, include_schemas=None
+    ):
         if include_schemas is not None:
             self.autogen_context.opts["include_schemas"] = include_schemas
         if object_filters is not None:
             self.autogen_context._object_filters = [object_filters]
+        if name_filters is not None:
+            self.autogen_context._name_filters = [name_filters]
         return self.autogen_context
 
 
@@ -263,6 +270,7 @@ class AutogenFixtureTest(_ComparesFKs):
         include_schemas=False,
         opts=None,
         object_filters=_default_object_filters,
+        name_filters=_default_name_filters,
         return_ops=False,
         max_identifier_length=None,
     ):
@@ -288,6 +296,7 @@ class AutogenFixtureTest(_ComparesFKs):
                     "alembic_module_prefix": "op.",
                     "sqlalchemy_module_prefix": "sa.",
                     "include_object": object_filters,
+                    "include_name": name_filters,
                     "include_schemas": include_schemas,
                 }
                 if opts:
diff --git a/tests/test_autogen_composition.py b/tests/test_autogen_composition.py
index d6664b2..c37ecba 100644
--- a/tests/test_autogen_composition.py
+++ b/tests/test_autogen_composition.py
@@ -260,7 +260,7 @@ class AutogenerateDiffTestWSchema(ModelOne, AutogenTest, TestBase):
                 "downgrade_token": "downgrades",
                 "alembic_module_prefix": "op.",
                 "sqlalchemy_module_prefix": "sa.",
-                "include_symbol": lambda name, schema: False,
+                "include_object": lambda name, *args: False,
             },
         )
         template_args = {}
diff --git a/tests/test_autogen_diffs.py b/tests/test_autogen_diffs.py
index 4ef7696..96184be 100644
--- a/tests/test_autogen_diffs.py
+++ b/tests/test_autogen_diffs.py
@@ -51,6 +51,7 @@ from alembic.testing import TestBase
 from alembic.testing.env import clear_staging_env
 from alembic.testing.env import staging_env
 from alembic.util import CommandError
+from ._autogen_fixtures import _default_name_filters
 from ._autogen_fixtures import _default_object_filters
 from ._autogen_fixtures import AutogenFixtureTest
 from ._autogen_fixtures import AutogenTest
@@ -98,6 +99,61 @@ class AutogenCrossSchemaTest(AutogenTest, TestBase):
         eq_(diffs[0][0], "add_table")
         eq_(diffs[0][1].schema, None)
 
+    def test_default_schema_omitted_by_table_name_upgrade(self):
+        def include_name(name, type_, parent_names):
+            if type_ == "table":
+                retval = name in ["t1", "t6"]
+                if retval:
+                    eq_(parent_names["schema_name"], None)
+                    eq_(parent_names["schema_qualified_table_name"], name)
+                else:
+                    eq_(parent_names["schema_name"], config.test_schema)
+                    eq_(
+                        parent_names["schema_qualified_table_name"],
+                        "%s.%s" % (config.test_schema, name),
+                    )
+                return retval
+            else:
+                return True
+
+        self._update_context(name_filters=include_name, include_schemas=True)
+        uo = ops.UpgradeOps(ops=[])
+        autogenerate._produce_net_changes(self.autogen_context, uo)
+
+        diffs = uo.as_diffs()
+        eq_(
+            {(d[0], d[1].name) for d in diffs},
+            {
+                ("add_table", "t3"),
+                ("add_table", "t4"),
+                ("remove_table", "t1"),
+                ("add_table", "t7"),
+            },
+        )
+
+    def test_default_schema_omitted_by_schema_name_upgrade(self):
+        def include_name(name, type_, parent_names):
+            if type_ == "schema":
+                assert not parent_names
+                return name is None
+            else:
+                return True
+
+        self._update_context(name_filters=include_name, include_schemas=True)
+        uo = ops.UpgradeOps(ops=[])
+        autogenerate._produce_net_changes(self.autogen_context, uo)
+
+        diffs = uo.as_diffs()
+        eq_(
+            {(d[0], d[1].name) for d in diffs},
+            {
+                ("add_table", "t3"),
+                ("add_table", "t4"),
+                ("remove_table", "t1"),
+                ("add_table", "t7"),
+            },
+        )
+
     def test_alt_schema_included_upgrade(self):
         def include_object(obj, name, type_, reflected, compare_to):
             if type_ == "table":
@@ -115,6 +171,32 @@ class AutogenCrossSchemaTest(AutogenTest, TestBase):
         eq_(diffs[0][0], "add_table")
         eq_(diffs[0][1].schema, config.test_schema)
 
+    def test_alt_schema_included_by_schema_name(self):
+        def include_name(name, type_, parent_names):
+            if type_ == "schema":
+                assert not parent_names
+                return name == config.test_schema
+            else:
+                return True
+
+        self._update_context(name_filters=include_name, include_schemas=True)
+        uo = ops.UpgradeOps(ops=[])
+        autogenerate._produce_net_changes(self.autogen_context, uo)
+
+        # does not include "t1" in drops because t1 is in default schema
+        # includes "t6" in adds because t6 is in default schema, was omitted,
+        # so reflection added it
+        diffs = uo.as_diffs()
+        eq_(
+            {(d[0], d[1].name) for d in diffs},
+            {
+                ("add_table", "t3"),
+                ("add_table", "t6"),
+                ("add_table", "t4"),
+                ("remove_table", "t2"),
+            },
+        )
+
     def test_default_schema_omitted_downgrade(self):
         def include_object(obj, name, type_, reflected, compare_to):
             if type_ == "table":
@@ -393,12 +475,23 @@ class AutogenerateDiffTest(ModelOne, AutogenTest, TestBase):
         eq_(diffs[10][3].table.name, "user")
         assert isinstance(diffs[10][3].type, String)
 
-    def test_include_symbol(self):
-
-        diffs = []
-
-        def include_symbol(name, schema=None):
-            return name in ("address", "order")
+    def test_include_object(self):
+        def include_object(obj, name, type_, reflected, compare_to):
+            assert obj.name == name
+            if type_ == "table":
+                if reflected:
+                    assert obj.metadata is not self.m2
+                else:
+                    assert obj.metadata is self.m2
+                return name in ("address", "order", "user")
+            elif type_ == "column":
+                if reflected:
+                    assert obj.table.metadata is not self.m2
+                else:
+                    assert obj.table.metadata is self.m2
+                return name != "street"
+            else:
+                return True
 
         context = MigrationContext.configure(
             connection=self.bind.connect(),
@@ -406,7 +499,7 @@ class AutogenerateDiffTest(ModelOne, AutogenTest, TestBase):
                 "compare_type": True,
                 "compare_server_default": True,
                 "target_metadata": self.m2,
-                "include_symbol": include_symbol,
+                "include_object": include_object,
             },
         )
 
@@ -414,29 +507,39 @@ class AutogenerateDiffTest(ModelOne, AutogenTest, TestBase):
             context, context.opts["target_metadata"]
         )
 
-        alter_cols = set(
-            [
-                d[2]
+        alter_cols = (
+            set(
+                [
+                    d[2]
+                    for d in self._flatten_diffs(diffs)
+                    if d[0].startswith("modify")
+                ]
+            )
+            .union(
+                d[3].name
+                for d in self._flatten_diffs(diffs)
+                if d[0] == "add_column"
+            )
+            .union(
+                d[1].name
                 for d in self._flatten_diffs(diffs)
-                if d[0].startswith("modify")
-            ]
+                if d[0] == "add_table"
+            )
         )
-        eq_(alter_cols, set(["order"]))
+        eq_(alter_cols, set(["user_id", "order", "user"]))
 
-    def test_include_object(self):
-        def include_object(obj, name, type_, reflected, compare_to):
-            assert obj.name == name
+    def test_include_name(self):
+        all_names = set()
+
+        def include_name(name, type_, parent_names):
+            all_names.add((name, type_, parent_names.get("table_name", None)))
             if type_ == "table":
-                if reflected:
-                    assert obj.metadata is not self.m2
-                else:
-                    assert obj.metadata is self.m2
+                eq_(
+                    parent_names,
+                    {"schema_name": None, "schema_qualified_table_name": name},
+                )
                 return name in ("address", "order", "user")
             elif type_ == "column":
-                if reflected:
-                    assert obj.table.metadata is not self.m2
-                else:
-                    assert obj.table.metadata is self.m2
                 return name != "street"
             else:
                 return True
@@ -447,13 +550,32 @@ class AutogenerateDiffTest(ModelOne, AutogenTest, TestBase):
                 "compare_type": True,
                 "compare_server_default": True,
                 "target_metadata": self.m2,
-                "include_object": include_object,
+                "include_name": include_name,
             },
         )
 
         diffs = autogenerate.compare_metadata(
             context, context.opts["target_metadata"]
         )
+        eq_(
+            all_names,
+            {
+                (None, "schema", None),
+                ("user", "table", None),
+                ("id", "column", "user"),
+                ("name", "column", "user"),
+                ("a1", "column", "user"),
+                ("pw", "column", "user"),
+                ("pw_idx", "index", "user"),
+                ("order", "table", None),
+                ("order_id", "column", "order"),
+                ("amount", "column", "order"),
+                ("address", "table", None),
+                ("id", "column", "address"),
+                ("email_address", "column", "address"),
+                ("extra", "table", None),
+            },
+        )
 
         alter_cols = (
             set(
@@ -474,7 +596,7 @@ class AutogenerateDiffTest(ModelOne, AutogenTest, TestBase):
                 if d[0] == "add_table"
             )
         )
-        eq_(alter_cols, set(["user_id", "order", "user"]))
+        eq_(alter_cols, {"user_id", "order", "user", "street", "item"})
 
     def test_skip_null_type_comparison_reflected(self):
         ac = ops.AlterColumnOp("sometable", "somecol")
@@ -754,6 +876,7 @@ class CompareMetadataToInspectorTest(TestBase):
             "alembic_module_prefix": "op.",
             "sqlalchemy_module_prefix": "sa.",
             "include_object": _default_object_filters,
+            "include_name": _default_name_filters,
         }
         if self.configure_opts:
             ctx_opts.update(self.configure_opts)
@@ -1325,42 +1448,62 @@ class CompareMetadataTest(ModelOne, AutogenTest, TestBase):
         eq_(diffs[1][2], "order")
         eq_(diffs[1][3], metadata.tables["order"].c.user_id)
 
-    def test_compare_metadata_include_symbol(self):
+    def test_compare_metadata_include_name(self):
         metadata = self.m2
 
-        def include_symbol(table_name, schema_name):
-            return table_name in ("extra", "order")
+        all_names = set()
+
+        def include_name(name, type_, parent_names):
+            all_names.add((name, type_, parent_names.get("table_name", None)))
+            if type_ == "table":
+                return name in ("extra", "order")
+            elif type_ == "column":
+                return name != "amount"
+            else:
+                return True
 
         context = MigrationContext.configure(
             connection=self.bind.connect(),
             opts={
                 "compare_type": True,
                 "compare_server_default": True,
-                "include_symbol": include_symbol,
+                "include_name": include_name,
             },
         )
 
         diffs = autogenerate.compare_metadata(context, metadata)
+        eq_(
+            all_names,
+            {
+                ("user", "table", None),
+                ("order", "table", None),
+                ("address", "table", None),
+                (None, "schema", None),
+                ("amount", "column", "order"),
+                ("extra", "table", None),
+                ("order_id", "column", "order"),
+            },
+        )
 
-        eq_(diffs[0][0], "remove_table")
-        eq_(diffs[0][1].name, "extra")
-
-        eq_(diffs[1][0], "add_column")
-        eq_(diffs[1][1], None)
-        eq_(diffs[1][2], "order")
-        eq_(diffs[1][3], metadata.tables["order"].c.user_id)
-
-        eq_(diffs[2][0][0], "modify_type")
-        eq_(diffs[2][0][1], None)
-        eq_(diffs[2][0][2], "order")
-        eq_(diffs[2][0][3], "amount")
-        eq_(repr(diffs[2][0][5]), "NUMERIC(precision=8, scale=2)")
-        eq_(repr(diffs[2][0][6]), "Numeric(precision=10, scale=2)")
-
-        eq_(diffs[2][1][0], "modify_nullable")
-        eq_(diffs[2][1][2], "order")
-        eq_(diffs[2][1][5], False)
-        eq_(diffs[2][1][6], True)
+        eq_(
+            {
+                (
+                    d[0],
+                    d[3].name if d[0] == "add_column" else d[1].name,
+                    d[2] if d[0] == "add_column" else None,
+                )
+                for d in diffs
+            },
+            {
+                ("remove_table", "extra", None),
+                ("add_fk", None, None),
+                ("add_column", "amount", "order"),
+                ("add_table", "user", None),
+                ("add_table", "item", None),
+                ("add_column", "user_id", "order"),
+                ("add_table", "address", None),
+            },
+        )
 
     def test_compare_metadata_as_sql(self):
         context = MigrationContext.configure(
diff --git a/tests/test_autogen_fks.py b/tests/test_autogen_fks.py
index 86c1c44..9a44c4b 100644
--- a/tests/test_autogen_fks.py
+++ b/tests/test_autogen_fks.py
@@ -7,6 +7,7 @@ from sqlalchemy import MetaData
 from sqlalchemy import String
 from sqlalchemy import Table
 
+from alembic.testing import combinations
 from alembic.testing import config
 from alembic.testing import eq_
 from alembic.testing import mock
@@ -445,8 +446,9 @@ class IncludeHooksTest(AutogenFixtureTest, TestBase):
     __backend__ = True
     __requires__ = ("fk_names",)
 
+    @combinations(("object",), ("name",))
     @config.requirements.no_name_normalize
-    def test_remove_connection_fk(self):
+    def test_remove_connection_fk(self, hook_type):
         m1 = MetaData()
         m2 = MetaData()
 
@@ -484,15 +486,37 @@ class IncludeHooksTest(AutogenFixtureTest, TestBase):
             mysql_engine="InnoDB",
         )
 
-        def include_object(object_, name, type_, reflected, compare_to):
-            return not (
-                isinstance(object_, ForeignKeyConstraint)
-                and type_ == "foreign_key_constraint"
-                and reflected
-                and name == "fk1"
-            )
-
-        diffs = self._fixture(m1, m2, object_filters=include_object)
+        if hook_type == "object":
+
+            def include_object(object_, name, type_, reflected, compare_to):
+                return not (
+                    isinstance(object_, ForeignKeyConstraint)
+                    and type_ == "foreign_key_constraint"
+                    and reflected
+                    and name == "fk1"
+                )
+
+            diffs = self._fixture(m1, m2, object_filters=include_object)
+        elif hook_type == "name":
+
+            def include_name(name, type_, parent_names):
+                if name == "fk1":
+                    if type_ == "index":  # MariaDB thing
+                        return True
+                    eq_(type_, "foreign_key_constraint")
+                    eq_(
+                        parent_names,
+                        {
+                            "schema_name": None,
+                            "table_name": "t",
+                            "schema_qualified_table_name": "t",
+                        },
+                    )
+                    return False
+                else:
+                    return True
+
+            diffs = self._fixture(m1, m2, name_filters=include_name)
 
         self._assert_fk_diff(
             diffs[0],
@@ -558,8 +582,9 @@ class IncludeHooksTest(AutogenFixtureTest, TestBase):
         )
         eq_(len(diffs), 1)
 
+    @combinations(("object",), ("name",))
     @config.requirements.no_name_normalize
-    def test_change_fk(self):
+    def test_change_fk(self, hook_type):
         m1 = MetaData()
         m2 = MetaData()
 
@@ -623,28 +648,57 @@ class IncludeHooksTest(AutogenFixtureTest, TestBase):
             )
         )
 
-        def include_object(object_, name, type_, reflected, compare_to):
-            return not (
-                isinstance(object_, ForeignKeyConstraint)
-                and type_ == "foreign_key_constraint"
-                and name == "fk1"
-            )
+        if hook_type == "object":
 
-        diffs = self._fixture(m1, m2, object_filters=include_object)
+            def include_object(object_, name, type_, reflected, compare_to):
+                return not (
+                    isinstance(object_, ForeignKeyConstraint)
+                    and type_ == "foreign_key_constraint"
+                    and name == "fk1"
+                )
 
-        self._assert_fk_diff(
-            diffs[0], "remove_fk", "t", ["y"], "ref_a", ["a"], name="fk2"
-        )
-        self._assert_fk_diff(
-            diffs[1],
-            "add_fk",
-            "t",
-            ["y", "z"],
-            "ref_b",
-            ["a", "b"],
-            name="fk2",
-        )
-        eq_(len(diffs), 2)
+            diffs = self._fixture(m1, m2, object_filters=include_object)
+        elif hook_type == "name":
+
+            def include_name(name, type_, parent_names):
+                if type_ == "index":
+                    return True  # MariaDB thing
+
+                if name == "fk1":
+                    eq_(type_, "foreign_key_constraint")
+                    eq_(
+                        parent_names,
+                        {
+                            "schema_name": None,
+                            "table_name": "t",
+                            "schema_qualified_table_name": "t",
+                        },
+                    )
+                    return False
+                else:
+                    return True
+
+            diffs = self._fixture(m1, m2, name_filters=include_name)
+
+        if hook_type == "object":
+            self._assert_fk_diff(
+                diffs[0], "remove_fk", "t", ["y"], "ref_a", ["a"], name="fk2"
+            )
+            self._assert_fk_diff(
+                diffs[1],
+                "add_fk",
+                "t",
+                ["y", "z"],
+                "ref_b",
+                ["a", "b"],
+                name="fk2",
+            )
+            eq_(len(diffs), 2)
+        elif hook_type == "name":
+            eq_(
+                {(d[0], d[1].name) for d in diffs},
+                {("add_fk", "fk2"), ("add_fk", "fk1"), ("remove_fk", "fk2")},
+            )
 
 
 class AutogenerateFKOptionsTest(AutogenFixtureTest, TestBase):
diff --git a/tests/test_autogen_indexes.py b/tests/test_autogen_indexes.py
index 943e61a..bc55206 100644
--- a/tests/test_autogen_indexes.py
+++ b/tests/test_autogen_indexes.py
@@ -13,6 +13,7 @@ from sqlalchemy import Table
 from sqlalchemy import UniqueConstraint
 
 from alembic.testing import assertions
+from alembic.testing import combinations
 from alembic.testing import config
 from alembic.testing import eq_
 from alembic.testing import TestBase
@@ -1308,7 +1309,8 @@ class NoUqReportsIndAsUqTest(NoUqReflectionIndexTest):
 class IncludeHooksTest(AutogenFixtureTest, TestBase):
     __backend__ = True
 
-    def test_remove_connection_index(self):
+    @combinations(("name",), ("object",))
+    def test_remove_connection_index(self, hook_type):
         m1 = MetaData()
         m2 = MetaData()
 
@@ -1318,25 +1320,59 @@ class IncludeHooksTest(AutogenFixtureTest, TestBase):
 
         Table("t", m2, Column("x", Integer), Column("y", Integer))
 
-        def include_object(object_, name, type_, reflected, compare_to):
-            if type_ == "unique_constraint":
-                return False
-            return not (
-                isinstance(object_, Index)
-                and type_ == "index"
-                and reflected
-                and name == "ix1"
-            )
+        if hook_type == "object":
 
-        diffs = self._fixture(m1, m2, object_filters=include_object)
+            def include_object(object_, name, type_, reflected, compare_to):
+                if type_ == "unique_constraint":
+                    return False
+                return not (
+                    isinstance(object_, Index)
+                    and type_ == "index"
+                    and reflected
+                    and name == "ix1"
+                )
+
+            diffs = self._fixture(m1, m2, object_filters=include_object)
+        elif hook_type == "name":
+            all_names = set()
+
+            def include_name(name, type_, parent_names):
+                all_names.add((name, type_))
+                if name == "ix1":
+                    eq_(type_, "index")
+                    eq_(
+                        parent_names,
+                        {
+                            "table_name": "t",
+                            "schema_name": None,
+                            "schema_qualified_table_name": "t",
+                        },
+                    )
+                    return False
+                else:
+                    return True
+
+            diffs = self._fixture(m1, m2, name_filters=include_name)
+            eq_(
+                all_names,
+                {
+                    ("ix1", "index"),
+                    ("ix2", "index"),
+                    ("y", "column"),
+                    ("t", "table"),
+                    (None, "schema"),
+                    ("x", "column"),
+                },
+            )
 
         eq_(diffs[0][0], "remove_index")
         eq_(diffs[0][1].name, "ix2")
         eq_(len(diffs), 1)
 
+    @combinations(("name",), ("object",))
     @config.requirements.unique_constraint_reflection
     @config.requirements.reflects_unique_constraints_unambiguously
-    def test_remove_connection_uq(self):
+    def test_remove_connection_uq(self, hook_type):
         m1 = MetaData()
         m2 = MetaData()
 
@@ -1351,17 +1387,53 @@ class IncludeHooksTest(AutogenFixtureTest, TestBase):
 
         Table("t", m2, Column("x", Integer), Column("y", Integer))
 
-        def include_object(object_, name, type_, reflected, compare_to):
-            if type_ == "index":
-                return False
-            return not (
-                isinstance(object_, UniqueConstraint)
-                and type_ == "unique_constraint"
-                and reflected
-                and name == "uq1"
-            )
+        if hook_type == "object":
 
-        diffs = self._fixture(m1, m2, object_filters=include_object)
+            def include_object(object_, name, type_, reflected, compare_to):
+                if type_ == "index":
+                    return False
+                return not (
+                    isinstance(object_, UniqueConstraint)
+                    and type_ == "unique_constraint"
+                    and reflected
+                    and name == "uq1"
+                )
+
+            diffs = self._fixture(m1, m2, object_filters=include_object)
+        elif hook_type == "name":
+            all_names = set()
+
+            def include_name(name, type_, parent_names):
+                if type_ == "index":
+                    return False  # PostgreSQL thing
+
+                all_names.add((name, type_))
+
+                if name == "uq1":
+                    eq_(type_, "unique_constraint")
+                    eq_(
+                        parent_names,
+                        {
+                            "table_name": "t",
+                            "schema_name": None,
+                            "schema_qualified_table_name": "t",
+                        },
+                    )
+                    return False
+                return True
+
+            diffs = self._fixture(m1, m2, name_filters=include_name)
+            eq_(
+                all_names,
+                {
+                    ("t", "table"),
+                    (None, "schema"),
+                    ("uq2", "unique_constraint"),
+                    ("x", "column"),
+                    ("y", "column"),
+                    ("uq1", "unique_constraint"),
+                },
+            )
 
         eq_(diffs[0][0], "remove_constraint")
         eq_(diffs[0][1].name, "uq2")