--- /dev/null
+.. change::
+ :tags: bug, orm, regression
+ :tickets: 6272
+
+ Fixed regression where an attribute that is mapped to a
+ :func:`_orm.synonym` could not be used in column loader options such as
+ :func:`_orm.load_only`.
+
+.. change::
+ :tags: usecase, orm
+ :tickets: 6267
+
+ Established support for :func:`_orm.synoynm` in conjunction with
+ hybrid property, assocaitionproxy is set up completely, including that
+ synonyms can be established linking to these constructs which work
+ fully. This is a behavior that was semi-explicitly disallowed previously,
+ however since it did not fail in every scenario, explicit support
+ for assoc proxy and hybrids has been added.
+
"""
+ _extra_criteria = ()
+
def __init__(
self,
class_,
from .. import sql
from .. import util
from ..sql import expression
+from ..sql import operators
class DescriptorProperty(MapperProperty):
def uses_objects(self):
return getattr(self.parent.class_, self.name).impl.uses_objects
- # TODO: when initialized, check _proxied_property,
+ # TODO: when initialized, check _proxied_object,
# emit a warning if its not a column-based property
@util.memoized_property
- def _proxied_property(self):
+ def _proxied_object(self):
attr = getattr(self.parent.class_, self.name)
if not hasattr(attr, "property") or not isinstance(
attr.property, MapperProperty
):
+ # attribute is a non-MapperProprerty proxy such as
+ # hybrid or association proxy
+ if isinstance(attr, attributes.QueryableAttribute):
+ return attr.comparator
+ elif isinstance(attr, operators.ColumnOperators):
+ return attr
+
raise sa_exc.InvalidRequestError(
"""synonym() attribute "%s.%s" only supports """
"""ORM mapped attributes, got %r"""
return attr.property
def _comparator_factory(self, mapper):
- prop = self._proxied_property
+ prop = self._proxied_object
- if self.comparator_factory:
- comp = self.comparator_factory(prop, mapper)
+ if isinstance(prop, MapperProperty):
+ if self.comparator_factory:
+ comp = self.comparator_factory(prop, mapper)
+ else:
+ comp = prop.comparator_factory(prop, mapper)
+ return comp
else:
- comp = prop.comparator_factory(prop, mapper)
- return comp
+ return prop
def get_history(self, *arg, **kw):
attr = getattr(self.parent.class_, self.name)
from sqlalchemy.testing import assert_raises_message
from sqlalchemy.testing import AssertsCompiledSQL
from sqlalchemy.testing import eq_
+from sqlalchemy.testing import expect_warnings
from sqlalchemy.testing import fixtures
from sqlalchemy.testing import is_
-from sqlalchemy.testing.assertions import expect_warnings
+from sqlalchemy.testing import is_false
from sqlalchemy.testing.fixtures import fixture_session
from sqlalchemy.testing.mock import call
from sqlalchemy.testing.mock import Mock
)
+class SynonymOfProxyTest(AssertsCompiledSQL, fixtures.DeclarativeMappedTest):
+ __dialect__ = "default"
+
+ run_create_tables = None
+
+ @classmethod
+ def setup_classes(cls):
+ from sqlalchemy.orm import synonym
+
+ Base = cls.DeclarativeBasic
+
+ class A(Base):
+ __tablename__ = "a"
+
+ id = Column(Integer, primary_key=True)
+ data = Column(String)
+ bs = relationship("B", backref="a")
+
+ b_data = association_proxy("bs", "data")
+
+ b_data_syn = synonym("b_data")
+
+ class B(Base):
+ __tablename__ = "b"
+ id = Column(Integer, primary_key=True)
+ a_id = Column(ForeignKey("a.id"))
+ data = Column(String)
+
+ def test_hasattr(self):
+ A, B = self.classes("A", "B")
+ is_false(hasattr(A.b_data_syn, "nonexistent"))
+
+ def test_o2m_instance_getter(self):
+ A, B = self.classes("A", "B")
+
+ a1 = A(bs=[B(data="bdata1"), B(data="bdata2")])
+ eq_(a1.b_data_syn, ["bdata1", "bdata2"])
+
+ def test_o2m_expr(self):
+ A, B = self.classes("A", "B")
+
+ self.assert_compile(
+ A.b_data_syn == "foo",
+ "EXISTS (SELECT 1 FROM a, b WHERE a.id = b.a_id "
+ "AND b.data = :data_1)",
+ )
+
+
class ProxyHybridTest(fixtures.DeclarativeMappedTest, AssertsCompiledSQL):
__dialect__ = "default"
from sqlalchemy import func
from sqlalchemy import inspect
from sqlalchemy import Integer
+from sqlalchemy import literal_column
from sqlalchemy import Numeric
from sqlalchemy import select
from sqlalchemy import String
from sqlalchemy.orm import aliased
from sqlalchemy.orm import relationship
from sqlalchemy.orm import Session
+from sqlalchemy.orm import synonym
from sqlalchemy.sql import update
from sqlalchemy.testing import assert_raises_message
from sqlalchemy.testing import AssertsCompiledSQL
from sqlalchemy.testing import eq_
from sqlalchemy.testing import fixtures
from sqlalchemy.testing import is_
+from sqlalchemy.testing import is_false
from sqlalchemy.testing.fixtures import fixture_session
from sqlalchemy.testing.schema import Column
is_(insp.all_orm_descriptors["value"].info, A.value.info)
+class SynonymOfPropertyTest(fixtures.TestBase, AssertsCompiledSQL):
+ __dialect__ = "default"
+
+ def _fixture(self):
+ Base = declarative_base()
+
+ class A(Base):
+ __tablename__ = "a"
+ id = Column(Integer, primary_key=True)
+ _value = Column("value", String)
+
+ @hybrid.hybrid_property
+ def value(self):
+ return self._value
+
+ value_syn = synonym("value")
+
+ @hybrid.hybrid_property
+ def string_value(self):
+ return "foo"
+
+ string_value_syn = synonym("string_value")
+
+ @hybrid.hybrid_property
+ def string_expr_value(self):
+ return "foo"
+
+ @string_expr_value.expression
+ def string_expr_value(cls):
+ return literal_column("'foo'")
+
+ string_expr_value_syn = synonym("string_expr_value")
+
+ return A
+
+ def test_hasattr(self):
+ A = self._fixture()
+
+ is_false(hasattr(A.value_syn, "nonexistent"))
+
+ is_false(hasattr(A.string_value_syn, "nonexistent"))
+
+ is_false(hasattr(A.string_expr_value_syn, "nonexistent"))
+
+ def test_instance_access(self):
+ A = self._fixture()
+
+ a1 = A(_value="hi")
+
+ eq_(a1.value_syn, "hi")
+
+ eq_(a1.string_value_syn, "foo")
+
+ eq_(a1.string_expr_value_syn, "foo")
+
+ def test_expression_property(self):
+ A = self._fixture()
+
+ self.assert_compile(
+ select(A.id, A.value_syn).where(A.value_syn == "value"),
+ "SELECT a.id, a.value FROM a WHERE a.value = :value_1",
+ )
+
+ def test_expression_expr(self):
+ A = self._fixture()
+
+ self.assert_compile(
+ select(A.id, A.string_expr_value_syn).where(
+ A.string_expr_value_syn == "value"
+ ),
+ "SELECT a.id, 'foo' FROM a WHERE 'foo' = :'foo'_1",
+ )
+
+
class MethodExpressionTest(fixtures.TestBase, AssertsCompiledSQL):
__dialect__ = "default"
"orders.isopen AS orders_isopen FROM orders",
)
+ @testing.combinations(("string",), ("attr",))
+ def test_load_only_synonym(self, type_):
+ orders, Order = self.tables.orders, self.classes.Order
+
+ mapper(
+ Order,
+ orders,
+ properties={"desc": synonym("description")},
+ )
+
+ if type_ == "attr":
+ opt = load_only(Order.isopen, Order.desc)
+ else:
+ opt = load_only("isopen", "desc")
+
+ sess = fixture_session()
+ q = sess.query(Order).options(opt)
+ self.assert_compile(
+ q,
+ "SELECT orders.id AS orders_id, orders.description "
+ "AS orders_description, orders.isopen AS orders_isopen "
+ "FROM orders",
+ )
+
def test_load_only_propagate_unbound(self):
self._test_load_only_propagate(False)
assert hasattr(User.x, "comparator")
def test_synonym_of_non_property_raises(self):
- from sqlalchemy.ext.associationproxy import association_proxy
-
class User(object):
- pass
+ @property
+ def x(self):
+ return "hi"
users, Address, addresses = (
self.tables.users,
properties={"y": synonym("x"), "addresses": relationship(Address)},
)
self.mapper(Address, addresses)
- User.x = association_proxy("addresses", "email_address")
assert_raises_message(
sa.exc.InvalidRequestError,
r'synonym\(\) attribute "User.x" only supports ORM mapped '
- "attributes, got .*AssociationProxy",
+ "attributes, got .*property",
getattr,
User.y,
"property",
)
+ assert_raises_message(
+ sa.exc.InvalidRequestError,
+ r'synonym\(\) attribute "User.x" only supports ORM mapped '
+ "attributes, got .*property",
+ lambda: User.y == 10,
+ )
+
def test_synonym_column_location(self):
users, User = self.tables.users, self.classes.User
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
