from sqlalchemy import log, util
import sqlalchemy.exceptions as sa_exc
-
+from sqlalchemy.sql import operators
from sqlalchemy.orm import (
- attributes, object_session, util as mapperutil, strategies,
+ attributes, object_session, util as mapperutil, strategies, object_mapper
)
from sqlalchemy.orm.query import Query
from sqlalchemy.orm.util import _state_has_identity, has_identity
def __init__(self, attr, state):
Query.__init__(self, attr.target_mapper, None)
- self.instance = state.obj()
+ self.instance = instance = state.obj()
self.attr = attr
+ mapper = object_mapper(instance)
+ prop = mapper.get_property(self.attr.key, resolve_synonyms=True)
+ self._criterion = prop.compare(
+ operators.eq,
+ instance,
+ value_is_parent=True,
+ alias_secondary=False)
+
+ if self.attr.order_by:
+ self._order_by = self.attr.order_by
+
def __session(self):
sess = object_session(self.instance)
if sess is not None and self.autoflush and sess.autoflush and self.instance in sess:
query = self.query_class(self.attr.target_mapper, session=sess)
else:
query = sess.query(self.attr.target_mapper)
- query = query.with_parent(instance, self.attr.key)
-
- if self.attr.order_by:
- query = query.order_by(*self.attr.order_by)
+
+ query._criterion = self._criterion
+ query._order_by = self._order_by
+
return query
def append(self, item):
- self.attr.append(attributes.instance_state(self.instance), attributes.instance_dict(self.instance), item, None)
+ self.attr.append(
+ attributes.instance_state(self.instance),
+ attributes.instance_dict(self.instance), item, None)
def remove(self, item):
- self.attr.remove(attributes.instance_state(self.instance), attributes.instance_dict(self.instance), item, None)
+ self.attr.remove(
+ attributes.instance_state(self.instance),
+ attributes.instance_dict(self.instance), item, None)
class AppenderQuery(AppenderMixin, Query):
self.prop.parent.compile()
return self.prop
- def compare(self, op, value, value_is_parent=False):
+ def compare(self, op, value, value_is_parent=False, alias_secondary=True):
if op == operators.eq:
if value is None:
if self.uselist:
return ~sql.exists([1], self.primaryjoin)
else:
- return self._optimized_compare(None, value_is_parent=value_is_parent)
+ return self._optimized_compare(None,
+ value_is_parent=value_is_parent,
+ alias_secondary=alias_secondary)
else:
- return self._optimized_compare(value, value_is_parent=value_is_parent)
+ return self._optimized_compare(value,
+ value_is_parent=value_is_parent,
+ alias_secondary=alias_secondary)
else:
return op(self.comparator, value)
- def _optimized_compare(self, value, value_is_parent=False, adapt_source=None):
+ def _optimized_compare(self, value, value_is_parent=False,
+ adapt_source=None, alias_secondary=True):
if value is not None:
value = attributes.instance_state(value)
return self._get_strategy(strategies.LazyLoader).\
- lazy_clause(value, reverse_direction=not value_is_parent, alias_secondary=True, adapt_source=adapt_source)
+ lazy_clause(value,
+ reverse_direction=not value_is_parent,
+ alias_secondary=alias_secondary, adapt_source=adapt_source)
def __str__(self):
return str(self.parent.class_.__name__) + "." + self.key
from sqlalchemy.test.schema import Table, Column
from sqlalchemy.orm import mapper, relation, create_session, Query, attributes
from sqlalchemy.orm.dynamic import AppenderMixin
-from sqlalchemy.test.testing import eq_
+from sqlalchemy.test.testing import eq_, AssertsCompiledSQL
from sqlalchemy.util import function_named
from test.orm import _base, _fixtures
-class DynamicTest(_fixtures.FixtureTest):
+class DynamicTest(_fixtures.FixtureTest, AssertsCompiledSQL):
@testing.resolve_artifact_names
def test_basic(self):
mapper(User, users, properties={
q.filter(User.id==7).all())
eq_(self.static.user_address_result, q.all())
+ @testing.resolve_artifact_names
+ def test_statement(self):
+ """test that the .statement accessor returns the actual statement that
+ would render, without any _clones called."""
+
+ mapper(User, users, properties={
+ 'addresses':dynamic_loader(mapper(Address, addresses))
+ })
+ sess = create_session()
+ q = sess.query(User)
+
+ u = q.filter(User.id==7).first()
+ self.assert_compile(
+ u.addresses.statement,
+ "SELECT addresses.id, addresses.user_id, addresses.email_address FROM "
+ "addresses WHERE :param_1 = addresses.user_id",
+ use_default_dialect=True
+ )
+
@testing.resolve_artifact_names
def test_order_by(self):
mapper(User, users, properties={
})
sess = create_session()
u = sess.query(User).get(8)
- eq_(list(u.addresses.order_by(desc(Address.email_address))), [Address(email_address=u'ed@wood.com'), Address(email_address=u'ed@lala.com'), Address(email_address=u'ed@bettyboop.com')])
+ eq_(
+ list(u.addresses.order_by(desc(Address.email_address))),
+ [Address(email_address=u'ed@wood.com'), Address(email_address=u'ed@lala.com'),
+ Address(email_address=u'ed@bettyboop.com')]
+ )
@testing.resolve_artifact_names
def test_configured_order_by(self):
assert o1 in i1.orders.all()
assert i1 in o1.items.all()
+ @testing.resolve_artifact_names
+ def test_association_nonaliased(self):
+ mapper(Order, orders, properties={
+ 'items':relation(Item, secondary=order_items,
+ lazy="dynamic",
+ order_by=order_items.c.item_id)
+ })
+ mapper(Item, items)
+
+ sess = create_session()
+ o = sess.query(Order).first()
+
+ self.assert_compile(
+ o.items,
+ "SELECT items.id AS items_id, items.description AS items_description FROM items,"
+ " order_items WHERE :param_1 = order_items.order_id AND items.id = order_items.item_id"
+ " ORDER BY order_items.item_id",
+ use_default_dialect=True
+ )
+
+ # filter criterion against the secondary table
+ # works
+ eq_(
+ o.items.filter(order_items.c.item_id==2).all(),
+ [Item(id=2)]
+ )
+
+
@testing.resolve_artifact_names
def test_transient_detached(self):
mapper(User, users, properties={
sess.delete(u)
sess.close()
-
-def _create_backref_test(autoflush, saveuser):
-
@testing.resolve_artifact_names
- def test_backref(self):
+ def _backref_test(self, autoflush, saveuser):
mapper(User, users, properties={
'addresses':dynamic_loader(mapper(Address, addresses), backref='user')
})
sess.flush()
eq_(list(u.addresses), [])
- test_backref = function_named(
- test_backref, "test%s%s" % ((autoflush and "_autoflush" or ""),
- (saveuser and "_saveuser" or "_savead")))
- setattr(SessionTest, test_backref.__name__, test_backref)
+ def test_backref_autoflush_saveuser(self):
+ self._backref_test(True, True)
+
+ def test_backref_autoflush_savead(self):
+ self._backref_test(True, False)
+
+ def test_backref_saveuser(self):
+ self._backref_test(False, True)
-for autoflush in (False, True):
- for saveuser in (False, True):
- _create_backref_test(autoflush, saveuser)
+ def test_backref_savead(self):
+ self._backref_test(False, False)
class DontDereferenceTest(_base.MappedTest):
@classmethod
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
