def _has_key(self, row, key):
try:
+ # _key_cache uses __missing__ in 2.5, so not much alternative
+ # to catching KeyError
self._key_cache[key]
return True
except KeyError:
# locate all embedded Column clauses so they can be added to the
# "inner" select statement where they'll be available to the enclosing
# statement's "order by"
+
+ cf = sql_util.ColumnFinder()
+
if order_by:
order_by = [expression._literal_as_text(o) for o in util.to_list(order_by) or []]
- cf = sql_util.ColumnFinder()
for o in order_by:
cf.traverse(o)
- else:
- cf = []
-
+
s2 = sql.select(context.primary_columns + list(cf), whereclause, from_obj=context.from_clauses, use_labels=True, correlate=False, **self._select_args())
if order_by:
"""Deferred column loader, a per-column or per-column-group lazy loader."""
def create_row_processor(self, selectcontext, mapper, row):
- if self.group is not None and selectcontext.attributes.get(('undefer', self.group), False):
+ if (self.group is not None and selectcontext.attributes.get(('undefer', self.group), False)) or self.columns[0] in row:
return self.parent_property._get_strategy(ColumnLoader).create_row_processor(selectcontext, mapper, row)
elif not self.is_class_level or len(selectcontext.options):
def new_execute(instance, row, **flags):
if self.secondaryjoin is not None:
context.eager_joins = sql.outerjoin(towrap, clauses.secondary, clauses.primaryjoin).outerjoin(clauses.alias, clauses.secondaryjoin)
+
+ # TODO: check for "deferred" cols on parent/child tables here ? this would only be
+ # useful if the primary/secondaryjoin are against non-PK columns on the tables (and therefore might be deferred)
+
if self.order_by is False and self.secondary.default_order_by() is not None:
context.eager_order_by += clauses.secondary.default_order_by()
else:
context.eager_joins = towrap.outerjoin(clauses.alias, clauses.primaryjoin)
+
+ # ensure all the cols on the parent side are actually in the
+ # columns clause (i.e. are not deferred), so that aliasing applied by the Query propagates
+ # those columns outward. This has the effect of "undefering" those columns.
+ for col in sql_util.find_columns(clauses.primaryjoin):
+ if localparent.mapped_table.c.contains_column(col):
+ context.primary_columns.append(col)
+ else:
+ context.secondary_columns.append(col)
+
if self.order_by is False and clauses.alias.default_order_by() is not None:
context.eager_order_by += clauses.alias.default_order_by()
def __init__(self, row):
self.row = row
def __contains__(self, key):
- return key in map or key in self.row
+ return key in self.row or (key in map and map[key] in self.row)
def has_key(self, key):
return key in self
def __getitem__(self, key):
def __iter__(self):
return iter(self.columns)
+def find_columns(selectable):
+ cf = ColumnFinder()
+ cf.traverse(selectable)
+ return iter(cf)
+
class ColumnsInClause(visitors.ClauseVisitor):
"""Given a selectable, visit clauses and determine if any columns
from the clause are in the selectable.
User(id=10, addresses=[])
] == sess.query(User).all()
+ def test_deferred_fk_col(self):
+ mapper(Address, addresses, properties={
+ 'user_id':deferred(addresses.c.user_id),
+ 'user':relation(User, lazy=False)
+ })
+ mapper(User, users)
+
+ assert [Address(id=1, user=User(id=7)), Address(id=4, user=User(id=8)), Address(id=5, user=User(id=9))] == create_session().query(Address).filter(Address.id.in_([1, 4, 5])).all()
+
+ assert [Address(id=1, user=User(id=7)), Address(id=4, user=User(id=8)), Address(id=5, user=User(id=9))] == create_session().query(Address).filter(Address.id.in_([1, 4, 5])).limit(3).all()
+
+ sess = create_session()
+ a = sess.query(Address).get(1)
+ def go():
+ assert a.user_id==7
+ # assert that the eager loader added 'user_id' to the row
+ # and deferred loading of that col was disabled
+ self.assert_sql_count(testbase.db, go, 0)
+
+ # do the mapping in reverse
+ # (we would have just used an "addresses" backref but the test fixtures then require the whole
+ # backref to be set up, lazy loaders trigger, etc.)
+ clear_mappers()
+
+ mapper(Address, addresses, properties={
+ 'user_id':deferred(addresses.c.user_id),
+ })
+ mapper(User, users, properties={'addresses':relation(Address, lazy=False)})
+
+ assert [User(id=7, addresses=[Address(id=1)])] == create_session().query(User).filter(User.id==7).options(eagerload('addresses')).all()
+
+ assert [User(id=7, addresses=[Address(id=1)])] == create_session().query(User).limit(1).filter(User.id==7).options(eagerload('addresses')).all()
+
def test_many_to_many(self):
mapper(Keyword, keywords)
self.assert_sql(testbase.db, go, [
("SELECT orders.user_id AS orders_user_id, orders.description AS orders_description, orders.isopen AS orders_isopen, orders.order_id AS orders_order_id FROM orders ORDER BY %s" % orderby, {}),
])
+
+ def test_locates_col(self):
+ """test that manually adding a col to the result undefers the column"""
+ mapper(Order, orders, properties={
+ 'description':deferred(orders.c.description)
+ })
-
+ sess = create_session()
+ o1 = sess.query(Order).first()
+ def go():
+ assert o1.description == 'order 1'
+ self.assert_sql_count(testbase.db, go, 1)
+
+ sess = create_session()
+ o1 = sess.query(Order).add_column(orders.c.description).first()[0]
+ def go():
+ assert o1.description == 'order 1'
+ self.assert_sql_count(testbase.db, go, 0)
+
def test_deepoptions(self):
m = mapper(User, users, properties={
'orders':relation(mapper(Order, orders, properties={
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
