Here is a minimal script which shows what I'm trying to do and where things
are going wrong.
--
SQLAlchemy -
The Python SQL Toolkit and Object Relational Mapper
http://www.sqlalchemy.org/
To post example code, please provide an MCVE: Minimal, Complete, and Verifiable
Example. See http://stackoverflow.com/help/mcve for a full description.
---
You received this message because you are subscribed to the Google Groups
"sqlalchemy" group.
To unsubscribe from this group and stop receiving emails from it, send an email
to [email protected].
To post to this group, send email to [email protected].
Visit this group at https://groups.google.com/group/sqlalchemy.
For more options, visit https://groups.google.com/d/optout.
from sqlalchemy import *
from sqlalchemy import event
from sqlalchemy.engine import Engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import (backref, relationship, joinedload, subqueryload,
sessionmaker)
from sqlalchemy.orm.query import Query
from sqlalchemy.orm.strategy_options import Load
engine = create_engine('sqlite:///test.db')
Base = declarative_base(bind=engine)
Session = sessionmaker(bind=engine)
class User(Base):
__tablename__ = 'users'
id = Column(Integer, primary_key=True)
class Order(Base):
__tablename__ = 'orders'
id = Column(Integer, primary_key=True)
user_id = Column(Integer, ForeignKey(User.id))
user = relationship(User, backref=backref('orders'))
#user = relationship(User, backref=backref('orders', lazy='subquery'))
class Item(Base):
__tablename__ = 'items'
id = Column(Integer, primary_key=True)
order_id = Column(Integer, ForeignKey(Order.id))
order = relationship(Order, backref=backref('items'))
#order = relationship(Order, backref=backref('items', lazy='joined'))
class QueryCounter(object):
__slots__ = ['count']
def __enter__(self):
event.listen(Engine, 'before_cursor_execute', self.callback)
self.count = 0
return self
def __exit__(self, *exc):
event.remove(Engine, 'before_cursor_execute', self.callback)
return False
def callback(self, *args, **kwargs):
self.count += 1
counter = QueryCounter()
def populate():
session = Session()
user = session.query(User).first()
if not user:
user = User()
order = Order(user=user)
item = Item(order=order)
session.add(item)
session.commit()
session.close()
def run_test(name, query):
print 'Running test %r' % name
session = Session()
with counter:
user = query.with_session(session).first()
for order in user.orders:
for item in order.items:
pass
try:
assert counter.count == 2
print ' test passed'
except:
print ' test failed (expected 2 queries, %d were issued)' % counter.count
session.close()
if __name__ == '__main__':
Base.metadata.drop_all()
Base.metadata.create_all()
populate()
#engine.echo = True
# bound chained strats work
query = Query(User).options(
Load(User)
.subqueryload('orders')
.joinedload('items')
)
run_test('bound/chained', query)
# unbound chained strats also work
query = Query(User).options(
subqueryload('orders').joinedload('items')
)
run_test('unbound/chained', query)
# attempting to set an entity default does not work
query = Query(User).options(
Load(User).subqueryload('orders'),
Load(Order).joinedload('items'),
)
run_test('entity-default 1', query)
# this also doesn't work
query = (Query(User)
.options(Load(User).subqueryload('orders'))
.options(Load(Order).joinedload('items'))
)
run_test('entity-default 2', query)
