--- /dev/null
+.. change::
+ :tags: bug, orm
+ :tickets: 8064
+
+ Fixed issue where using a :func:`_orm.column_property` construct containing
+ a subquery against an already-mapped column attribute would not correctly
+ apply ORM-compilation behaviors to the subquery, including that the "IN"
+ expression added for a single-table inherits expression would fail to be
+ included.
self.columns.add(col, key)
- for col in prop.columns + prop._orig_columns:
+ for col in prop.columns:
for proxy_col in col.proxy_set:
self._columntoproperty[proxy_col] = prop
@HasMemoized.memoized_attribute
def _single_table_criterion(self):
if self.single and self.inherits and self.polymorphic_on is not None:
- return self.polymorphic_on._annotate({"parentmapper": self}).in_(
- [m.polymorphic_identity for m in self.self_and_descendants]
- )
+ return self.polymorphic_on._annotate(
+ {"parententity": self, "parentmapper": self}
+ ).in_([m.polymorphic_identity for m in self.self_and_descendants])
else:
return None
from .interfaces import PropComparator
from .interfaces import StrategizedProperty
from .relationships import Relationship
-from .util import _orm_full_deannotate
from .. import exc as sa_exc
from .. import ForeignKey
from .. import log
_links_to_entity = False
columns: List[NamedColumn[Any]]
- _orig_columns: List[NamedColumn[Any]]
_is_polymorphic_discriminator: bool
comparator_factory: Type[PropComparator[_T]]
__slots__ = (
- "_orig_columns",
"columns",
"group",
"deferred",
attribute_options=attribute_options
)
columns = (column,) + additional_columns
- self._orig_columns = [
- coercions.expect(roles.LabeledColumnExprRole, c) for c in columns
- ]
self.columns = [
- _orm_full_deannotate(
- coercions.expect(roles.LabeledColumnExprRole, c)
- )
- for c in columns
+ coercions.expect(roles.LabeledColumnExprRole, c) for c in columns
]
self.group = group
self.deferred = deferred
def _include_fn(self, elem):
entity = elem._annotations.get("parentmapper", None)
- return not entity or entity.isa(self.mapper)
+ return not entity or entity.common_parent(self.mapper)
class AliasedClass(
from sqlalchemy import true
from sqlalchemy.orm import aliased
from sqlalchemy.orm import Bundle
+from sqlalchemy.orm import column_property
from sqlalchemy.orm import joinedload
from sqlalchemy.orm import relationship
from sqlalchemy.orm import Session
super(EagerDefaultEvalTestPolymorphic, cls).setup_classes(
with_polymorphic="*"
)
+
+
+class ColExprTest(AssertsCompiledSQL, fixtures.TestBase):
+ def test_discrim_on_column_prop(self, registry):
+ Base = registry.generate_base()
+
+ class Employee(Base):
+ __tablename__ = "employee"
+ id = Column(Integer, primary_key=True)
+ type = Column(String(20))
+
+ __mapper_args__ = {
+ "polymorphic_on": "type",
+ "polymorphic_identity": "employee",
+ }
+
+ class Engineer(Employee):
+ __mapper_args__ = {"polymorphic_identity": "engineer"}
+
+ class Company(Base):
+ __tablename__ = "company"
+ id = Column(Integer, primary_key=True)
+
+ max_engineer_id = column_property(
+ select(func.max(Engineer.id)).scalar_subquery()
+ )
+
+ self.assert_compile(
+ select(Company.max_engineer_id),
+ "SELECT (SELECT max(employee.id) AS max_1 FROM employee "
+ "WHERE employee.type IN (__[POSTCOMPILE_type_1])) AS anon_1",
+ )
# has to lazy load the addresses
self.assert_sql_count(testing.db, go, 1)
+ def test_column_property_adaptation(self, decl_base):
+ """test #2316 in support of #8064"""
+
+ class A(decl_base):
+ __tablename__ = "a"
+ id = Column(Integer, primary_key=True)
+ type = Column(String(40), nullable=False)
+ __mapper_args__ = {"polymorphic_on": type}
+
+ A.anything = column_property(A.id + 1000)
+
+ class B(A):
+ __tablename__ = "b"
+ account_id = Column(Integer, ForeignKey("a.id"), primary_key=True)
+ x_id = Column(Integer, ForeignKey("x.id"), nullable=False)
+ __mapper_args__ = {"polymorphic_identity": "named"}
+
+ class X(decl_base):
+ __tablename__ = "x"
+ id = Column(Integer, primary_key=True)
+ b = relationship("B")
+
+ self.assert_compile(
+ select(X).options(joinedload(X.b)),
+ "SELECT x.id, a_1.id AS id_1, a_1.type, a_1.id + :id_2 AS anon_1, "
+ "b_1.account_id, b_1.x_id FROM x "
+ "LEFT OUTER JOIN "
+ "(a AS a_1 JOIN b AS b_1 ON a_1.id = b_1.account_id) "
+ "ON x.id = b_1.x_id",
+ )
+
def test_no_render_in_subquery(self):
"""test #6378"""
)
@testing.combinations((True,), (False,))
- def test_add_column_prop_deannotate(self, autoalias):
+ def test_add_column_prop_adaption(self, autoalias):
+ """test ultimately from #2316 revised for #8064"""
User, users = self.classes.User, self.tables.users
Address, addresses = self.classes.Address, self.tables.addresses
"users_1.id = addresses.user_id",
)
- def test_column_prop_deannotate(self):
- """test that column property deannotates,
- bringing expressions down to the original mapped columns.
+ def test_column_prop_stays_annotated(self):
+ """test ultimately from #2316 revised for #8064.
+
+ previously column_property() would deannotate the given expression,
+ however this interfered with some compilation sceanrios.
+
+
"""
User, users = self.classes.User, self.tables.users
m = self.mapper(User, users)
m.add_property("y", column_property(expr2.scalar_subquery()))
assert User.x.property.columns[0] is not expr
- assert User.x.property.columns[0].element.left is users.c.name
- # a deannotate needs to clone the base, in case
- # the original one referenced annotated elements.
- assert User.x.property.columns[0].element.right is not expr.right
+
+ assert (
+ User.x.property.columns[0].element.left
+ is User.name.comparator.expr
+ )
+
+ assert User.x.property.columns[0].element.right is expr.right
assert User.y.property.columns[0] is not expr2
assert (
- User.y.property.columns[0].element._raw_columns[0] is users.c.name
+ User.y.property.columns[0].element._raw_columns[0]
+ is User.name.expression
)
assert User.y.property.columns[0].element._raw_columns[1] is users.c.id
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
