--- /dev/null
+.. change::
+ :tags: bug, orm
+ :tickets: 10597
+
+ Fixed issue where use of :func:`_orm.foreign` annotation on a
+ non-initialized :func:`_orm.mapped_column` construct would produce an
+ expression without a type, which was then not updated at initialization
+ time of the actual column, leading to issues such as relationships not
+ determining ``use_get`` appropriately.
+
return c
+class _memoized_property_but_not_nulltype(
+ util.memoized_property["TypeEngine[_T]"]
+):
+ """memoized property, but dont memoize NullType"""
+
+ def __get__(self, obj, cls):
+ if obj is None:
+ return self
+ result = self.fget(obj)
+ if not result._isnull:
+ obj.__dict__[self.__name__] = result
+ return result
+
+
class AnnotatedColumnElement(Annotated):
_Annotated__element: ColumnElement[Any]
"_tq_key_label",
"_tq_label",
"_non_anon_label",
+ "type",
):
self.__dict__.pop(attr, None)
for attr in ("name", "key", "table"):
"""pull 'name' from parent, if not present"""
return self._Annotated__element.name
+ @_memoized_property_but_not_nulltype
+ def type(self):
+ """pull 'type' from parent and don't cache if null.
+
+ type is routinely changed on existing columns within the
+ mapped_column() initialization process, and "type" is also consulted
+ during the creation of SQL expressions. Therefore it can change after
+ it was already retrieved. At the same time we don't want annotated
+ objects having overhead when expressions are produced, so continue
+ to memoize, but only when we have a non-null type.
+
+ """
+ return self._Annotated__element.type
+
@util.memoized_property
def table(self):
"""pull 'table' from parent, if not present"""
identity: Optional[Identity]
def _set_type(self, type_: TypeEngine[Any]) -> None:
+ assert self.type._isnull or type_ is self.type
+
self.type = type_
if isinstance(self.type, SchemaEventTarget):
self.type._set_parent_with_dispatch(self)
from sqlalchemy.orm import declared_attr
from sqlalchemy.orm import deferred
from sqlalchemy.orm import DynamicMapped
+from sqlalchemy.orm import foreign
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import MappedAsDataclass
from sqlalchemy.orm import relationship
+from sqlalchemy.orm import remote
from sqlalchemy.orm import Session
from sqlalchemy.orm import undefer
from sqlalchemy.orm import WriteOnlyMapped
is_(MyClass.__table__.c.data.type, typ)
is_true(MyClass.__table__.c.id.primary_key)
+ @testing.variation("style", ["none", "lambda_", "string", "direct"])
+ def test_foreign_annotation_propagates_correctly(self, decl_base, style):
+ """test #10597"""
+
+ class Parent(decl_base):
+ __tablename__ = "parent"
+ id: Mapped[int] = mapped_column(primary_key=True)
+
+ class Child(decl_base):
+ __tablename__ = "child"
+
+ name: Mapped[str] = mapped_column(primary_key=True)
+
+ if style.none:
+ parent_id: Mapped[int] = mapped_column(ForeignKey("parent.id"))
+ else:
+ parent_id: Mapped[int] = mapped_column()
+
+ if style.lambda_:
+ parent: Mapped[Parent] = relationship(
+ primaryjoin=lambda: remote(Parent.id)
+ == foreign(Child.parent_id),
+ )
+ elif style.string:
+ parent: Mapped[Parent] = relationship(
+ primaryjoin="remote(Parent.id) == "
+ "foreign(Child.parent_id)",
+ )
+ elif style.direct:
+ parent: Mapped[Parent] = relationship(
+ primaryjoin=remote(Parent.id) == foreign(parent_id),
+ )
+ elif style.none:
+ parent: Mapped[Parent] = relationship()
+
+ assert Child.__mapper__.attrs.parent.strategy.use_get
+
@testing.combinations(
(BIGINT(),),
(BIGINT,),
from sqlalchemy.orm import declared_attr
from sqlalchemy.orm import deferred
from sqlalchemy.orm import DynamicMapped
+from sqlalchemy.orm import foreign
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import MappedAsDataclass
from sqlalchemy.orm import relationship
+from sqlalchemy.orm import remote
from sqlalchemy.orm import Session
from sqlalchemy.orm import undefer
from sqlalchemy.orm import WriteOnlyMapped
is_(MyClass.__table__.c.data.type, typ)
is_true(MyClass.__table__.c.id.primary_key)
+ @testing.variation("style", ["none", "lambda_", "string", "direct"])
+ def test_foreign_annotation_propagates_correctly(self, decl_base, style):
+ """test #10597"""
+
+ class Parent(decl_base):
+ __tablename__ = "parent"
+ id: Mapped[int] = mapped_column(primary_key=True)
+
+ class Child(decl_base):
+ __tablename__ = "child"
+
+ name: Mapped[str] = mapped_column(primary_key=True)
+
+ if style.none:
+ parent_id: Mapped[int] = mapped_column(ForeignKey("parent.id"))
+ else:
+ parent_id: Mapped[int] = mapped_column()
+
+ if style.lambda_:
+ parent: Mapped[Parent] = relationship(
+ primaryjoin=lambda: remote(Parent.id)
+ == foreign(Child.parent_id),
+ )
+ elif style.string:
+ parent: Mapped[Parent] = relationship(
+ primaryjoin="remote(Parent.id) == "
+ "foreign(Child.parent_id)",
+ )
+ elif style.direct:
+ parent: Mapped[Parent] = relationship(
+ primaryjoin=remote(Parent.id) == foreign(parent_id),
+ )
+ elif style.none:
+ parent: Mapped[Parent] = relationship()
+
+ assert Child.__mapper__.attrs.parent.strategy.use_get
+
@testing.combinations(
(BIGINT(),),
(BIGINT,),
from sqlalchemy.sql import LABEL_STYLE_DISAMBIGUATE_ONLY
from sqlalchemy.sql import LABEL_STYLE_TABLENAME_PLUS_COL
from sqlalchemy.sql import operators
+from sqlalchemy.sql import sqltypes
from sqlalchemy.sql import table
from sqlalchemy.sql import util as sql_util
from sqlalchemy.sql import visitors
eq_(whereclause.left._annotations, {"foo": "bar"})
eq_(whereclause.right._annotations, {"foo": "bar"})
+ @testing.variation("use_col_ahead_of_time", [True, False])
+ def test_set_type_on_column(self, use_col_ahead_of_time):
+ """test related to #10597"""
+
+ col = Column()
+
+ col_anno = col._annotate({"foo": "bar"})
+
+ if use_col_ahead_of_time:
+ expr = col_anno == bindparam("foo")
+
+ # this could only be fixed if we put some kind of a container
+ # that receives the type directly rather than using NullType;
+ # like a PendingType or something
+
+ is_(expr.right.type._type_affinity, sqltypes.NullType)
+
+ assert "type" not in col_anno.__dict__
+
+ col.name = "name"
+ col._set_type(Integer())
+
+ eq_(col_anno.name, "name")
+ is_(col_anno.type._type_affinity, Integer)
+
+ expr = col_anno == bindparam("foo")
+
+ is_(expr.right.type._type_affinity, Integer)
+
+ assert "type" in col_anno.__dict__
+
@testing.combinations(True, False, None)
def test_setup_inherit_cache(self, inherit_cache_value):
if inherit_cache_value is None:
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
