Check explicitly for mapped class as secondary

Added a comprehensive check and an informative error message for the case
where a mapped class, or a string mapped class name, is passed to
:paramref:`_orm.relationship.secondary`.  This is an extremely common error
which warrants a clear message.

Additionally, added a new rule to the class registry resolution such that
with regards to the :paramref:`_orm.relationship.secondary` parameter, if a
mapped class and its table are of the identical string name, the
:class:`.Table` will be favored when resolving this parameter.   In all
other cases, the class continues to be favored if a class and table
share the identical name.

Fixes: #5774
Change-Id: I65069d79c1c3897fbd1081a8e1edf3b63b497223

doc/build/changelog/unreleased_13/5774.rst

@@ -0,0 +1,16 @@
+.. change::
+    :tags: bug, orm
+    :tickets: 5774
+    :versions: 1.4.0b2
+
+    Added a comprehensive check and an informative error message for the case
+    where a mapped class, or a string mapped class name, is passed to
+    :paramref:`_orm.relationship.secondary`.  This is an extremely common error
+    which warrants a clear message.
+
+    Additionally, added a new rule to the class registry resolution such that
+    with regards to the :paramref:`_orm.relationship.secondary` parameter, if a
+    mapped class and its table are of the identical string name, the
+    :class:`.Table` will be favored when resolving this parameter.   In all
+    other cases, the class continues to be favored if a class and table
+    share the identical name.

lib/sqlalchemy/orm/clsregistry.py

@@ -315,15 +315,24 @@ def _determine_container(key, value):
 
 
 class _class_resolver(object):
-    __slots__ = "cls", "prop", "arg", "fallback", "_dict", "_resolvers"
+    __slots__ = (
+        "cls",
+        "prop",
+        "arg",
+        "fallback",
+        "_dict",
+        "_resolvers",
+        "favor_tables",
+    )
 
-    def __init__(self, cls, prop, fallback, arg):
+    def __init__(self, cls, prop, fallback, arg, favor_tables=False):
         self.cls = cls
         self.prop = prop
         self.arg = arg
         self.fallback = fallback
         self._dict = util.PopulateDict(self._access_cls)
         self._resolvers = ()
+        self.favor_tables = favor_tables
 
     def _access_cls(self, key):
         cls = self.cls
@@ -333,13 +342,22 @@ class _class_resolver(object):
         decl_class_registry = decl_base._class_registry
         metadata = decl_base.metadata
 
+        if self.favor_tables:
+            if key in metadata.tables:
+                return metadata.tables[key]
+            elif key in metadata._schemas:
+                return _GetTable(key, cls.metadata)
+
         if key in decl_class_registry:
             return _determine_container(key, decl_class_registry[key])
-        elif key in metadata.tables:
-            return metadata.tables[key]
-        elif key in metadata._schemas:
-            return _GetTable(key, cls.metadata)
-        elif (
+
+        if not self.favor_tables:
+            if key in metadata.tables:
+                return metadata.tables[key]
+            elif key in metadata._schemas:
+                return _GetTable(key, cls.metadata)
+
+        if (
             "_sa_module_registry" in decl_class_registry
             and key in decl_class_registry["_sa_module_registry"]
         ):
@@ -412,8 +430,10 @@ def _resolver(cls, prop):
             {"foreign": foreign, "remote": remote}
         )
 
-    def resolve_arg(arg):
-        return _class_resolver(cls, prop, _fallback_dict, arg)
+    def resolve_arg(arg, favor_tables=False):
+        return _class_resolver(
+            cls, prop, _fallback_dict, arg, favor_tables=favor_tables
+        )
 
     def resolve_name(arg):
         return _class_resolver(cls, prop, _fallback_dict, arg)._resolve_name

lib/sqlalchemy/orm/relationships.py

@@ -20,6 +20,7 @@ import re
 import weakref
 
 from . import attributes
+from .base import _is_mapped_class
 from .base import state_str
 from .interfaces import MANYTOMANY
 from .interfaces import MANYTOONE
@@ -2163,9 +2164,13 @@ class RelationshipProperty(StrategizedProperty):
 
             if isinstance(attr_value, util.string_types):
                 setattr(
-                    self, attr, self._clsregistry_resolve_arg(attr_value)()
+                    self,
+                    attr,
+                    self._clsregistry_resolve_arg(
+                        attr_value, favor_tables=attr == "secondary"
+                    )(),
                 )
-            elif callable(attr_value):
+            elif callable(attr_value) and not _is_mapped_class(attr_value):
                 setattr(self, attr, attr_value())
 
         # remove "annotations" which are present if mapped class
@@ -2183,6 +2188,15 @@ class RelationshipProperty(StrategizedProperty):
                     ),
                 )
 
+        if self.secondary is not None and _is_mapped_class(self.secondary):
+            raise sa_exc.ArgumentError(
+                "secondary argument %s passed to to relationship() %s must "
+                "be a Table object or other FROM clause; can't send a mapped "
+                "class directly as rows in 'secondary' are persisted "
+                "independently of a class that is mapped "
+                "to that same table." % (self.secondary, self)
+            )
+
         # ensure expressions in self.order_by, foreign_keys,
         # remote_side are all columns, not strings.
         if self.order_by is not False and self.order_by is not None:

test/orm/declarative/test_basic.py

@@ -872,8 +872,8 @@ class DeclarativeTest(DeclarativeTestBase):
             props = relationship(
                 "Prop",
                 secondary="user_to_prop",
-                primaryjoin="User.id==user_to_prop.c.u" "ser_id",
-                secondaryjoin="user_to_prop.c.prop_id=" "=Prop.id",
+                primaryjoin="User.id==user_to_prop.c.user_id",
+                secondaryjoin="user_to_prop.c.prop_id==Prop.id",
                 backref="users",
             )
 
@@ -895,6 +895,59 @@ class DeclarativeTest(DeclarativeTestBase):
             class_mapper(User).get_property("props").secondary is user_to_prop
         )
 
+    def test_string_dependency_resolution_table_over_class(self):
+        # test for second half of #5774
+        class User(Base, fixtures.ComparableEntity):
+
+            __tablename__ = "users"
+            id = Column(Integer, primary_key=True)
+            name = Column(String(50))
+            props = relationship(
+                "Prop",
+                secondary="Secondary",
+                backref="users",
+            )
+
+        class Prop(Base, fixtures.ComparableEntity):
+
+            __tablename__ = "props"
+            id = Column(Integer, primary_key=True)
+            name = Column(String(50))
+
+        # class name and table name match
+        class Secondary(Base):
+            __tablename__ = "Secondary"
+            user_id = Column(Integer, ForeignKey("users.id"), primary_key=True)
+            prop_id = Column(Integer, ForeignKey("props.id"), primary_key=True)
+
+        configure_mappers()
+        assert (
+            class_mapper(User).get_property("props").secondary
+            is Secondary.__table__
+        )
+
+    def test_string_dependency_resolution_class_over_table(self):
+        # test for second half of #5774
+        class User(Base, fixtures.ComparableEntity):
+
+            __tablename__ = "users"
+            id = Column(Integer, primary_key=True)
+            name = Column(String(50))
+            secondary = relationship(
+                "Secondary",
+            )
+
+        # class name and table name match
+        class Secondary(Base):
+            __tablename__ = "Secondary"
+            user_id = Column(Integer, ForeignKey("users.id"), primary_key=True)
+
+        configure_mappers()
+        assert (
+            class_mapper(User).get_property("secondary").mapper
+            is Secondary.__mapper__
+        )
+
     def test_string_dependency_resolution_schemas(self):
         Base = declarative_base()
 

test/orm/test_relationships.py

@@ -4355,6 +4355,47 @@ class AmbiguousFKResolutionTest(_RelationshipErrors, fixtures.MappedTest):
         sa.orm.configure_mappers()
 
 
+class SecondaryArgTest(fixtures.TestBase):
+    def teardown(self):
+        clear_mappers()
+
+    @testing.combinations((True,), (False,))
+    def test_informative_message_on_cls_as_secondary(self, string):
+        Base = declarative_base()
+
+        class C(Base):
+            __tablename__ = "c"
+            id = Column(Integer, primary_key=True)
+            a_id = Column(ForeignKey("a.id"))
+            b_id = Column(ForeignKey("b.id"))
+
+        if string:
+            c_arg = "C"
+        else:
+            c_arg = C
+
+        class A(Base):
+            __tablename__ = "a"
+
+            id = Column(Integer, primary_key=True)
+            data = Column(String)
+            bs = relationship("B", secondary=c_arg)
+
+        class B(Base):
+            __tablename__ = "b"
+            id = Column(Integer, primary_key=True)
+
+        assert_raises_message(
+            exc.ArgumentError,
+            r"secondary argument <class .*C.*> passed to to "
+            r"relationship\(\) A.bs "
+            "must be a Table object or other FROM clause; can't send a "
+            "mapped class directly as rows in 'secondary' are persisted "
+            "independently of a class that is mapped to that same table.",
+            configure_mappers,
+        )
+
+
 class SecondaryNestedJoinTest(
     fixtures.MappedTest, AssertsCompiledSQL, testing.AssertsExecutionResults
 ):