:ticket:`3423`
+.. _change_4351:
+
+AssociationProxy now provides standard column operators for a column-oriented target
+------------------------------------------------------------------------------------
+
+Given an :class:`.AssociationProxy` where the target is a database column,
+as opposed to an object reference::
+
+ class User(Base):
+ # ...
+
+ elements = relationship("Element")
+
+ # column-based association proxy
+ values = association_proxy("elements", "value")
+
+ class Element(Base):
+ # ...
+
+ value = Column(String)
+
+The ``User.values`` association proxy refers to the ``Element.value`` column.
+Standard column operations are now available, such as ``like``::
+
+ >>> print(s.query(User).filter(User.values.like('%foo%')))
+ SELECT "user".id AS user_id
+ FROM "user"
+ WHERE EXISTS (SELECT 1
+ FROM element
+ WHERE "user".id = element.user_id AND element.value LIKE :value_1)
+
+``equals``::
+
+ >>> print(s.query(User).filter(User.values == 'foo'))
+ SELECT "user".id AS user_id
+ FROM "user"
+ WHERE EXISTS (SELECT 1
+ FROM element
+ WHERE "user".id = element.user_id AND element.value = :value_1)
+
+When comparing to ``None``, the ``IS NULL`` expression is augmented with
+a test that the related row does not exist at all; this is the same
+behavior as before::
+
+ >>> print(s.query(User).filter(User.values == None))
+ SELECT "user".id AS user_id
+ FROM "user"
+ WHERE (EXISTS (SELECT 1
+ FROM element
+ WHERE "user".id = element.user_id AND element.value IS NULL)) OR NOT (EXISTS (SELECT 1
+ FROM element
+ WHERE "user".id = element.user_id))
+
+Note that the :meth:`.ColumnOperators.contains` operator is in fact a string
+comparison operator; **this is a change in behavior** in that previously,
+the association proxy used ``.contains`` as a list containment operator only.
+With a column-oriented comparison, it now behaves like a "like"::
+
+ >>> print(s.query(User).filter(User.values.contains('foo')))
+ SELECT "user".id AS user_id
+ FROM "user"
+ WHERE EXISTS (SELECT 1
+ FROM element
+ WHERE "user".id = element.user_id AND (element.value LIKE '%' || :value_1 || '%'))
+
+In order to test the ``User.values`` collection for simple membership of the value
+``"foo"``, the equals operator (e.g. ``User.values == 'foo'``) should be used;
+this works in previous versions as well.
+
+When using an object-based association proxy with a collection, the behavior is
+as before, that of testing for collection membership, e.g. given a mapping::
+
+ class User(Base):
+ __tablename__ = 'user'
+
+ id = Column(Integer, primary_key=True)
+ user_elements = relationship("UserElement")
+
+ # object-based association proxy
+ elements = association_proxy("user_elements", "element")
+
+
+ class UserElement(Base):
+ __tablename__ = 'user_element'
+
+ id = Column(Integer, primary_key=True)
+ user_id = Column(ForeignKey("user.id"))
+ element_id = Column(ForeignKey("element.id"))
+ element = relationship("Element")
+
+
+ class Element(Base):
+ __tablename__ = 'element'
+
+ id = Column(Integer, primary_key=True)
+ value = Column(String)
+
+The ``.contains()`` method produces the same expression as before, testing
+the list of ``User.elements`` for the presence of an ``Element`` object::
+
+ >>> print(s.query(User).filter(User.elements.contains(Element(id=1))))
+ SELECT "user".id AS user_id
+ FROM "user"
+ WHERE EXISTS (SELECT 1
+ FROM user_element
+ WHERE "user".id = user_element.user_id AND :param_1 = user_element.element_id)
+
+Overall, the change is enabled based on the architectural change that is
+part of :ref:`change_3423`; as the proxy now spins off additional state when
+an expression is generated, there is both an object-target and a column-target
+version of the :class:`.AssociationProxyInstance` class.
+
+:ticket:`4351`
+
+
.. _change_4246:
FOR UPDATE clause is rendered within the joined eager load subquery as well as outside
--- /dev/null
+.. change::
+ :tags: feature, ext
+ :tickets: 4351
+
+ The :class:`.AssociationProxy` now has standard column comparison operations
+ such as :meth:`.ColumnOperators.like` and
+ :meth:`.ColumnOperators.startswith` available when the target attribute is a
+ plain column - the EXISTS expression that joins to the target table is
+ rendered as usual, but the column expression is then use within the WHERE
+ criteria of the EXISTS. Note that this alters the behavior of the
+ ``.contains()`` method on the association proxy to make use of
+ :meth:`.ColumnOperators.contains` when used on a column-based attribute.
+
+ .. seealso::
+
+ :ref:`change_4351`
+
:undoc-members:
:inherited-members:
+.. autoclass:: ObjectAssociationProxyInstance
+ :members:
+ :inherited-members:
+
+.. autoclass:: ColumnAssociationProxyInstance
+ :members:
+ :inherited-members:
+
.. autodata:: ASSOCIATION_PROXY
from .. import exc, orm, util
from ..orm import collections, interfaces
from ..sql import or_
+from ..sql.operators import ColumnOperators
from .. import inspect
except KeyError:
owner = self._calc_owner(class_)
if owner is not None:
- result = AssociationProxyInstance(self, owner)
+ result = AssociationProxyInstance.for_proxy(self, owner)
setattr(class_, self.key + "_inst", result)
return result
else:
"""
- def __init__(self, parent, owning_class):
+ def __init__(self, parent, owning_class, target_class, value_attr):
self.parent = parent
self.key = parent.key
self.owning_class = owning_class
self.target_collection = parent.target_collection
self.value_attr = parent.value_attr
self.collection_class = None
+ self.target_class = target_class
+ self.value_attr = value_attr
+
+ target_class = None
+ """The intermediary class handled by this
+ :class:`.AssociationProxyInstance`.
+
+ Intercepted append/set/assignment events will result
+ in the generation of new instances of this class.
+
+ """
+
+ @classmethod
+ def for_proxy(cls, parent, owning_class):
+ target_collection = parent.target_collection
+ value_attr = parent.value_attr
+ prop = orm.class_mapper(owning_class).\
+ get_property(target_collection)
+ target_class = prop.mapper.class_
+
+ target_assoc = cls._cls_unwrap_target_assoc_proxy(
+ target_class, value_attr)
+ if target_assoc is not None:
+ return ObjectAssociationProxyInstance(
+ parent, owning_class, target_class, value_attr
+ )
+
+ is_object = getattr(target_class, value_attr).impl.uses_objects
+ if is_object:
+ return ObjectAssociationProxyInstance(
+ parent, owning_class, target_class, value_attr
+ )
+ else:
+ return ColumnAssociationProxyInstance(
+ parent, owning_class, target_class, value_attr
+ )
def _get_property(self):
return orm.class_mapper(self.owning_class).\
def _comparator(self):
return self._get_property().comparator
- @util.memoized_property
- def _unwrap_target_assoc_proxy(self):
- attr = getattr(self.target_class, self.value_attr)
+ @classmethod
+ def _cls_unwrap_target_assoc_proxy(cls, target_class, value_attr):
+ attr = getattr(target_class, value_attr)
if isinstance(attr, (AssociationProxy, AssociationProxyInstance)):
return attr
return None
+ @util.memoized_property
+ def _unwrap_target_assoc_proxy(self):
+ return self._cls_unwrap_target_assoc_proxy(
+ self.target_class, self.value_attr)
+
@property
def remote_attr(self):
"""The 'remote' :class:`.MapperProperty` referenced by this
"""
return (self.local_attr, self.remote_attr)
- @util.memoized_property
- def target_class(self):
- """The intermediary class handled by this
- :class:`.AssociationProxyInstance`.
-
- Intercepted append/set/assignment events will result
- in the generation of new instances of this class.
-
- """
- return self._get_property().mapper.class_
-
@util.memoized_property
def scalar(self):
"""Return ``True`` if this :class:`.AssociationProxyInstance`
return not self._get_property().\
mapper.get_property(self.value_attr).uselist
- @util.memoized_property
+ @property
def _target_is_object(self):
- return getattr(self.target_class, self.value_attr).impl.uses_objects
+ raise NotImplementedError()
def _initialize_scalar_accessors(self):
if self.parent.getset_factory:
return self._criterion_exists(
criterion=criterion, is_has=True, **kwargs)
+
+class ObjectAssociationProxyInstance(AssociationProxyInstance):
+ """an :class:`.AssociationProxyInstance` that has an object as a target.
+ """
+ _target_is_object = True
+
def contains(self, obj):
"""Produce a proxied 'contains' expression using EXISTS.
elif self._target_is_object and self.scalar and \
self._value_is_scalar:
raise exc.InvalidRequestError(
- "contains() doesn't apply to a scalar endpoint; use ==")
+ "contains() doesn't apply to a scalar object endpoint; use ==")
else:
return self._comparator._criterion_exists(**{self.value_attr: obj})
getattr(self.target_class, self.value_attr) != obj)
+class ColumnAssociationProxyInstance(
+ ColumnOperators, AssociationProxyInstance):
+ """an :class:`.AssociationProxyInstance` that has a database column as a
+ target.
+ """
+ _target_is_object = False
+
+ def __eq__(self, other):
+ # special case "is None" to check for no related row as well
+ expr = self._criterion_exists(
+ self.remote_attr.operate(operator.eq, other)
+ )
+ if other is None:
+ return or_(
+ expr, self._comparator == None
+ )
+ else:
+ return expr
+
+ def operate(self, op, *other, **kwargs):
+ return self._criterion_exists(
+ self.remote_attr.operate(op, *other, **kwargs)
+ )
+
+
class _lazy_collection(object):
def __init__(self, obj, target):
self.parent = obj
from sqlalchemy.testing.assertions import expect_warnings
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.ext.declarative import declared_attr
-
+from sqlalchemy.engine import default
class DictCollection(dict):
@collection.appender
cls.session = session
def _equivalent(self, q_proxy, q_direct):
+ proxy_sql = q_proxy.statement.compile(dialect=default.DefaultDialect())
+ direct_sql = q_direct.statement.compile(
+ dialect=default.DefaultDialect())
+ eq_(str(proxy_sql), str(direct_sql))
eq_(q_proxy.all(), q_direct.all())
def test_filter_any_criterion_ul_scalar(self):
def test_filter_ne_nul_nul(self):
Keyword = self.classes.Keyword
+ UserKeyword = self.classes.UserKeyword
self._equivalent(
self.session.query(Keyword).filter(Keyword.user != self.u),
self.session.query(Keyword).filter(
- Keyword.user_keyword.has(Keyword.user != self.u)))
+ Keyword.user_keyword.has(UserKeyword.user != self.u)))
def test_filter_eq_null_nul_nul(self):
UserKeyword, Keyword = self.classes.UserKeyword, self.classes.Keyword
self.session.query(Keyword).filter(
Keyword.user_keyword.has(UserKeyword.user != None)))
- def test_filter_eq_None_nul(self):
+ def test_filter_object_eq_None_nul(self):
+ UserKeyword = self.classes.UserKeyword
+ User = self.classes.User
+
+ self._equivalent(
+ self.session.query(UserKeyword).filter(
+ UserKeyword.singular == None), # noqa
+ self.session.query(UserKeyword).filter(or_(
+ UserKeyword.user.has(User.singular == None),
+ UserKeyword.user_id == None)
+ )
+ )
+
+ def test_filter_column_eq_None_nul(self):
User = self.classes.User
Singular = self.classes.Singular
User.singular_value == None), # noqa
self.session.query(User).filter(or_(
User.singular.has(Singular.value == None),
- User.singular == None)))
+ User.singular == None)
+ )
+ )
- def test_filter_ne_value_nul(self):
+ def test_filter_object_ne_value_nul(self):
+ UserKeyword = self.classes.UserKeyword
+ User = self.classes.User
+ Singular = self.classes.Singular
+
+ s4 = self.session.query(Singular).filter_by(value="singular4").one()
+ self._equivalent(
+ self.session.query(UserKeyword).filter(
+ UserKeyword.singular != s4),
+ self.session.query(UserKeyword).filter(
+ UserKeyword.user.has(User.singular != s4)))
+
+ def test_filter_column_ne_value_nul(self):
User = self.classes.User
Singular = self.classes.Singular
User.singular_keyword.has, keyword="brown"
)
- def test_filter_contains_chained_has_to_any(self):
+ def test_filter_eq_chained_has_to_any(self):
User = self.classes.User
Keyword = self.classes.Keyword
Singular = self.classes.Singular
q1 = self.session.query(User).filter(
- User.singular_keyword.contains("brown")
+ User.singular_keyword == "brown"
)
self.assert_compile(
q1,
User.singular_value.has, singular_value="singular4"
)
- def test_filter_scalar_contains_fails_nul_nul(self):
+ def test_filter_scalar_object_contains_fails_nul_nul(self):
Keyword = self.classes.Keyword
assert_raises(exc.InvalidRequestError,
lambda: Keyword.user.contains(self.u))
- def test_filter_scalar_any_fails_nul_nul(self):
+ def test_filter_scalar_object_any_fails_nul_nul(self):
Keyword = self.classes.Keyword
assert_raises(exc.InvalidRequestError,
lambda: Keyword.user.any(name='user2'))
+ def test_filter_scalar_column_like(self):
+ User = self.classes.User
+ Singular = self.classes.Singular
+
+ self._equivalent(
+ self.session.query(User).filter(User.singular_value.like('foo')),
+ self.session.query(User).filter(
+ User.singular.has(Singular.value.like('foo')),
+ )
+ )
+
+ def test_filter_scalar_column_contains(self):
+ User = self.classes.User
+ Singular = self.classes.Singular
+
+ self._equivalent(
+ self.session.query(User).filter(User.singular_value.contains('foo')),
+ self.session.query(User).filter(
+ User.singular.has(Singular.value.contains('foo')),
+ )
+ )
+
+ def test_filter_scalar_column_eq(self):
+ User = self.classes.User
+ Singular = self.classes.Singular
+
+ self._equivalent(
+ self.session.query(User).filter(User.singular_value == 'foo'),
+ self.session.query(User).filter(
+ User.singular.has(Singular.value == 'foo'),
+ )
+ )
+
+ def test_filter_scalar_column_ne(self):
+ User = self.classes.User
+ Singular = self.classes.Singular
+
+ self._equivalent(
+ self.session.query(User).filter(User.singular_value != 'foo'),
+ self.session.query(User).filter(
+ User.singular.has(Singular.value != 'foo'),
+ )
+ )
+
+ def test_filter_scalar_column_eq_nul(self):
+ User = self.classes.User
+ Singular = self.classes.Singular
+
+ self._equivalent(
+ self.session.query(User).filter(User.singular_value == None),
+ self.session.query(User).filter(or_(
+ User.singular.has(Singular.value == None),
+ User.singular == None
+ ))
+ )
+
def test_filter_collection_has_fails_ul_nul(self):
User = self.classes.User
m = MetaData()
a = Table('a', m, Column('id', Integer, primary_key=True))
b = Table('b', m, Column('id', Integer, primary_key=True),
- Column('aid', Integer, ForeignKey('a.id')))
+ Column('aid', Integer, ForeignKey('a.id')),
+ Column('elem', String))
mapper(A, a, properties={
'orig': relationship(
B,
__tablename__ = 'child'
parent_id = Column(
Integer, ForeignKey(Parent.id), primary_key=True)
+ value = Column(String)
# 2. declarative builds up SubParent, scans through all attributes
# over all classes. Hits Mixin, hits "children", accesses "children"
__tablename__ = 'child'
parent_id = Column(
Integer, ForeignKey(Parent.id), primary_key=True)
+ value = Column(String)
class SubParent(Parent):
__tablename__ = 'subparent'
__tablename__ = 'child'
parent_id = Column(
Integer, ForeignKey(Parent.id), primary_key=True)
+ value = Column(String)
class SubParent(Parent):
__tablename__ = 'subparent'
__tablename__ = 'child'
parent_id = Column(
Integer, ForeignKey(Parent.id), primary_key=True)
+ value = Column(String)
class SubParent(Parent):
__tablename__ = 'subparent'
c_id = Column(ForeignKey('c.id'))
c2_id = Column(ForeignKey('c2.id'))
- def test_any_has(self):
+ def test_column_collection_expressions(self):
B, C, C2 = self.classes("B", "C", "C2")
self.assert_compile(
B.d_values.contains('b1'),
"EXISTS (SELECT 1 FROM d, b WHERE d.b_id = b.id "
- "AND d.value = :value_1)"
+ "AND (d.value LIKE '%' || :value_1 || '%'))"
)
self.assert_compile(
C2.d_values.contains("c2"),
"EXISTS (SELECT 1 FROM d, c2 WHERE d.c2_id = c2.id "
- "AND d.value = :value_1)"
+ "AND (d.value LIKE '%' || :value_1 || '%'))"
)
self.assert_compile(
C.d_values.contains('c1'),
"EXISTS (SELECT 1 FROM d, c WHERE d.c_id = c.id "
- "AND d.value = :value_1)"
+ "AND (d.value LIKE '%' || :value_1 || '%'))"
)
+
class ScopeBehaviorTest(fixtures.DeclarativeMappedTest):
# test some GC scenarios, including issue #4268
from sqlalchemy import exc, util
from sqlalchemy import inspect
from test.orm import _fixtures
-from sqlalchemy.orm import class_mapper, synonym, Session, aliased
+from sqlalchemy.orm import class_mapper, synonym, Session, aliased,\
+ relationship
+from sqlalchemy import ForeignKey
from sqlalchemy.orm.attributes import instance_state, NO_VALUE
from sqlalchemy import testing
from sqlalchemy.orm.util import identity_key
def conv(self, fn):
raise NotImplementedError()
+ class Address(self.classes.Address):
+ pass
+
class SomeSubClass(SomeClass):
@hybrid_property
def upper_name(self):
def foo(self):
raise NotImplementedError()
- t = Table('sometable', MetaData(),
+ m = MetaData()
+ t = Table('sometable', m,
Column('id', Integer, primary_key=True))
- mapper(SomeClass, t)
+ ta = Table('address_t', m,
+ Column('id', Integer, primary_key=True),
+ Column('s_id', ForeignKey('sometable.id'))
+ )
+ mapper(SomeClass, t, properties={
+ "addresses": relationship(Address)
+ })
+ mapper(Address, ta)
mapper(SomeSubClass, inherits=SomeClass)
insp = inspect(SomeSubClass)
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
