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
(cherry picked from commit
cfcfb7d43c2719ef6e4901dca43de5b492a1f08f)
--- /dev/null
+.. 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.
class _class_resolver(object):
- 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 = self._declarative_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
+
+ if self.favor_tables:
+ if key in cls.metadata.tables:
+ return cls.metadata.tables[key]
+ elif key in cls.metadata._schemas:
+ return _GetTable(key, cls.metadata)
+
if key in cls._decl_class_registry:
return _determine_container(key, cls._decl_class_registry[key])
- elif key in cls.metadata.tables:
- return cls.metadata.tables[key]
- elif key in cls.metadata._schemas:
- return _GetTable(key, cls.metadata)
- elif (
+
+ if not self.favor_tables:
+ if key in cls.metadata.tables:
+ return cls.metadata.tables[key]
+ elif key in cls.metadata._schemas:
+ return _GetTable(key, cls.metadata)
+
+ if (
"_sa_module_registry" in cls._decl_class_registry
and key in cls._decl_class_registry["_sa_module_registry"]
):
fallback = sqlalchemy.__dict__.copy()
fallback.update({"foreign": foreign, "remote": remote})
- def resolve_arg(arg):
- return _class_resolver(cls, prop, fallback, arg)
+ def resolve_arg(arg, favor_tables=False):
+ return _class_resolver(
+ cls, prop, fallback, arg, favor_tables=favor_tables
+ )
def resolve_name(arg):
return _class_resolver(cls, prop, fallback, arg)._resolve_name
):
v = getattr(prop, attr)
if isinstance(v, util.string_types):
- setattr(prop, attr, resolve_arg(v))
+ setattr(
+ prop,
+ attr,
+ resolve_arg(v, favor_tables=attr == "secondary"),
+ )
for attr in ("argument",):
v = getattr(prop, attr)
from . import attributes
from . import dependency
from . import mapper as mapperlib
+from .base import _is_mapped_class
from .base import state_str
from .interfaces import MANYTOMANY
from .interfaces import MANYTOONE
"remote_side",
):
attr_value = getattr(self, attr)
- if util.callable(attr_value):
+ if util.callable(attr_value) and not _is_mapped_class(attr_value):
setattr(self, attr, attr_value())
# remove "annotations" which are present if mapped class
),
)
+ 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:
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 = decl.declarative_base()
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
):
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
