--- /dev/null
+.. change::
+ :tags: bug, orm
+ :tickets: 10564
+
+ The :paramref:`_orm.relationship.secondary` parameter no longer uses Python
+ ``eval()`` to evaluate the given string. This parameter when passed a
+ string should resolve to a table name that's present in the local
+ :class:`.MetaData` collection only, and never needs to be any kind of
+ Python expression otherwise. To use a real deferred callable based on a
+ name that may not be locally present yet, use a lambda instead.
collection associated with the
parent-mapped :class:`_schema.Table`.
- .. warning:: When passed as a Python-evaluable string, the
- argument is interpreted using Python's ``eval()`` function.
- **DO NOT PASS UNTRUSTED INPUT TO THIS STRING**.
- See :ref:`declarative_relationship_eval` for details on
- declarative evaluation of :func:`_orm.relationship` arguments.
+ .. versionchanged:: 2.1 When passed as a string, the argument is
+ interpreted as a string name that should exist directly in the
+ registry of tables. The Python ``eval()`` function is no longer
+ used for the :paramref:`_orm.relationship.secondary` argument when
+ passed as a string.
The :paramref:`_orm.relationship.secondary` keyword argument is
typically applied in the case where the intermediary
"fallback",
"_dict",
"_resolvers",
- "favor_tables",
+ "tables_only",
)
cls: Type[Any]
prop: RelationshipProperty[Any]
fallback: Mapping[str, Any]
arg: str
- favor_tables: bool
+ tables_only: bool
_resolvers: Tuple[Callable[[str], Any], ...]
def __init__(
prop: RelationshipProperty[Any],
fallback: Mapping[str, Any],
arg: str,
- favor_tables: bool = False,
+ tables_only: bool = False,
):
self.cls = cls
self.prop = prop
self.fallback = fallback
self._dict = util.PopulateDict(self._access_cls)
self._resolvers = ()
- self.favor_tables = favor_tables
+ self.tables_only = tables_only
def _access_cls(self, key: str) -> Any:
cls = self.cls
decl_class_registry = decl_base._class_registry
metadata = decl_base.metadata
- if self.favor_tables:
+ if self.tables_only:
if key in metadata.tables:
return metadata.tables[key]
elif key in metadata._schemas:
return _GetTable(key, getattr(cls, "metadata", metadata))
if key in decl_class_registry:
- return _determine_container(key, decl_class_registry[key])
+ dt = _determine_container(key, decl_class_registry[key])
+ if self.tables_only:
+ return dt.cls
+ else:
+ return dt
- if not self.favor_tables:
+ if not self.tables_only:
if key in metadata.tables:
return metadata.tables[key]
elif key in metadata._schemas:
_ModuleMarker, decl_class_registry["_sa_module_registry"]
)
return registry.resolve_attr(key)
- elif self._resolvers:
+
+ if self._resolvers:
for resolv in self._resolvers:
value = resolv(key)
if value is not None:
return rval
def __call__(self) -> Any:
- try:
- x = eval(self.arg, globals(), self._dict)
+ if self.tables_only:
+ try:
+ return self._dict[self.arg]
+ except KeyError as k:
+ self._raise_for_name(self.arg, k)
+ else:
+ try:
+ x = eval(self.arg, globals(), self._dict)
- if isinstance(x, _GetColumns):
- return x.cls
- else:
- return x
- except NameError as n:
- self._raise_for_name(n.args[0], n)
+ if isinstance(x, _GetColumns):
+ return x.cls
+ else:
+ return x
+ except NameError as n:
+ self._raise_for_name(n.args[0], n)
_fallback_dict: Mapping[str, Any] = None # type: ignore
{"foreign": foreign, "remote": remote}
)
- def resolve_arg(arg: str, favor_tables: bool = False) -> _class_resolver:
+ def resolve_arg(arg: str, tables_only: bool = False) -> _class_resolver:
return _class_resolver(
- cls, prop, _fallback_dict, arg, favor_tables=favor_tables
+ cls, prop, _fallback_dict, arg, tables_only=tables_only
)
def resolve_name(
from sqlalchemy import Index
from sqlalchemy import inspect
from sqlalchemy import Integer
+from sqlalchemy import join
from sqlalchemy import literal
from sqlalchemy import select
from sqlalchemy import String
d = relationship(
"D",
- secondary="join(B, D, B.d_id == D.id)."
- "join(C, C.d_id == D.id)",
+ secondary=lambda: join(B, D, B.d_id == D.id).join(
+ C, C.d_id == D.id
+ ),
primaryjoin="and_(A.b_id == B.id, A.id == C.a_id)",
secondaryjoin="D.id == B.d_id",
)
d = relationship(
"D",
- secondary="join(B, C)",
+ secondary=join(B, C),
primaryjoin="A.b_id == B.id",
secondaryjoin="C.d_id == D.id",
uselist=False,
def teardown_test(self):
clear_mappers()
+ @testing.variation("arg_style", ["string", "table", "lambda_"])
+ def test_secondary_arg_styles(self, arg_style):
+ Base = declarative_base()
+
+ c = Table(
+ "c",
+ Base.metadata,
+ Column("a_id", ForeignKey("a.id")),
+ Column("b_id", ForeignKey("b.id")),
+ )
+
+ class A(Base):
+ __tablename__ = "a"
+
+ id = Column(Integer, primary_key=True)
+ data = Column(String)
+
+ if arg_style.string:
+ bs = relationship("B", secondary="c")
+ elif arg_style.table:
+ bs = relationship("B", secondary=c)
+ elif arg_style.lambda_:
+ bs = relationship("B", secondary=lambda: c)
+ else:
+ arg_style.fail()
+
+ class B(Base):
+ __tablename__ = "b"
+ id = Column(Integer, primary_key=True)
+
+ is_(inspect(A).relationships.bs.secondary, c)
+
+ def test_no_eval_in_secondary(self):
+ """test #10564"""
+ Base = declarative_base()
+
+ Table(
+ "c",
+ Base.metadata,
+ Column("a_id", ForeignKey("a.id")),
+ Column("b_id", ForeignKey("b.id")),
+ )
+
+ class A(Base):
+ __tablename__ = "a"
+
+ id = Column(Integer, primary_key=True)
+ data = Column(String)
+
+ bs = relationship("B", secondary="c.c.a_id.table")
+
+ class B(Base):
+ __tablename__ = "b"
+ id = Column(Integer, primary_key=True)
+
+ with expect_raises_message(
+ exc.InvalidRequestError,
+ r"When initializing mapper Mapper\[A\(a\)\], expression "
+ r"'c.c.a_id.table' failed to locate a name \('c.c.a_id.table'\). ",
+ ):
+ Base.registry.configure()
+
@testing.combinations((True,), (False,))
def test_informative_message_on_cls_as_secondary(self, string):
Base = declarative_base()
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
