from sqlalchemy.test.testing import eq_, assert_raises, assert_raises_message from sqlalchemy import * from sqlalchemy.orm import * from sqlalchemy.orm import interfaces from sqlalchemy import exc as sa_exc from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.engine import default from sqlalchemy.test import AssertsCompiledSQL, testing from test.orm import _base, _fixtures from sqlalchemy.test.testing import eq_ from sqlalchemy.test.schema import Table, Column class Company(_fixtures.Base): pass class Person(_fixtures.Base): pass class Engineer(Person): pass class Manager(Person): pass class Boss(Manager): pass class Machine(_fixtures.Base): pass class Paperwork(_fixtures.Base): pass def _produce_test(select_type): class PolymorphicQueryTest(_base.MappedTest, AssertsCompiledSQL): run_inserts = 'once' run_setup_mappers = 'once' run_deletes = None @classmethod def define_tables(cls, metadata): global companies, people, engineers, managers, boss, paperwork, machines companies = Table('companies', metadata, Column('company_id', Integer, primary_key=True, test_needs_autoincrement=True), Column('name', String(50))) people = Table('people', metadata, Column('person_id', Integer, primary_key=True, test_needs_autoincrement=True), Column('company_id', Integer, ForeignKey('companies.company_id')), Column('name', String(50)), Column('type', String(30))) engineers = Table('engineers', metadata, Column('person_id', Integer, ForeignKey('people.person_id'), primary_key=True), Column('status', String(30)), Column('engineer_name', String(50)), Column('primary_language', String(50)), ) machines = Table('machines', metadata, Column('machine_id', Integer, primary_key=True, test_needs_autoincrement=True), Column('name', String(50)), Column('engineer_id', Integer, ForeignKey('engineers.person_id'))) managers = Table('managers', metadata, Column('person_id', Integer, ForeignKey('people.person_id'), primary_key=True), Column('status', String(30)), Column('manager_name', String(50)) ) boss = Table('boss', metadata, Column('boss_id', Integer, ForeignKey('managers.person_id'), primary_key=True), Column('golf_swing', String(30)), ) paperwork = Table('paperwork', metadata, Column('paperwork_id', Integer, primary_key=True, test_needs_autoincrement=True), Column('description', String(50)), Column('person_id', Integer, ForeignKey('people.person_id'))) clear_mappers() mapper(Company, companies, properties={ 'employees':relationship(Person, order_by=people.c.person_id) }) mapper(Machine, machines) if select_type == '': person_join = manager_join = None person_with_polymorphic = None manager_with_polymorphic = None elif select_type == 'Polymorphic': person_join = manager_join = None person_with_polymorphic = '*' manager_with_polymorphic = '*' elif select_type == 'Unions': person_join = polymorphic_union( { 'engineer':people.join(engineers), 'manager':people.join(managers), }, None, 'pjoin') manager_join = people.join(managers).outerjoin(boss) person_with_polymorphic = ([Person, Manager, Engineer], person_join) manager_with_polymorphic = ('*', manager_join) elif select_type == 'AliasedJoins': person_join = people.outerjoin(engineers).outerjoin(managers).select(use_labels=True).alias('pjoin') manager_join = people.join(managers).outerjoin(boss).select(use_labels=True).alias('mjoin') person_with_polymorphic = ([Person, Manager, Engineer], person_join) manager_with_polymorphic = ('*', manager_join) elif select_type == 'Joins': person_join = people.outerjoin(engineers).outerjoin(managers) manager_join = people.join(managers).outerjoin(boss) person_with_polymorphic = ([Person, Manager, Engineer], person_join) manager_with_polymorphic = ('*', manager_join) # testing a order_by here as well; the surrogate mapper has to adapt it mapper(Person, people, with_polymorphic=person_with_polymorphic, polymorphic_on=people.c.type, polymorphic_identity='person', order_by=people.c.person_id, properties={ 'paperwork':relationship(Paperwork, order_by=paperwork.c.paperwork_id) }) mapper(Engineer, engineers, inherits=Person, polymorphic_identity='engineer', properties={ 'machines':relationship(Machine, order_by=machines.c.machine_id) }) mapper(Manager, managers, with_polymorphic=manager_with_polymorphic, inherits=Person, polymorphic_identity='manager') mapper(Boss, boss, inherits=Manager, polymorphic_identity='boss') mapper(Paperwork, paperwork) @classmethod def insert_data(cls): global all_employees, c1_employees, c2_employees, e1, e2, b1, m1, e3, c1, c2 c1 = Company(name="MegaCorp, Inc.") c2 = Company(name="Elbonia, Inc.") e1 = Engineer(name="dilbert", engineer_name="dilbert", primary_language="java", status="regular engineer", paperwork=[ Paperwork(description="tps report #1"), Paperwork(description="tps report #2") ], machines=[ Machine(name='IBM ThinkPad'), Machine(name='IPhone'), ]) e2 = Engineer(name="wally", engineer_name="wally", primary_language="c++", status="regular engineer", paperwork=[ Paperwork(description="tps report #3"), Paperwork(description="tps report #4") ], machines=[ Machine(name="Commodore 64") ]) b1 = Boss(name="pointy haired boss", golf_swing="fore", manager_name="pointy", status="da boss", paperwork=[ Paperwork(description="review #1"), ]) m1 = Manager(name="dogbert", manager_name="dogbert", status="regular manager", paperwork=[ Paperwork(description="review #2"), Paperwork(description="review #3") ]) c1.employees = [e1, e2, b1, m1] e3 = Engineer(name="vlad", engineer_name="vlad", primary_language="cobol", status="elbonian engineer", paperwork=[ Paperwork(description='elbonian missive #3') ], machines=[ Machine(name="Commodore 64"), Machine(name="IBM 3270") ]) c2.employees = [e3] sess = create_session() sess.add(c1) sess.add(c2) sess.flush() sess.expunge_all() all_employees = [e1, e2, b1, m1, e3] c1_employees = [e1, e2, b1, m1] c2_employees = [e3] def test_loads_at_once(self): """test that all objects load from the full query, when with_polymorphic is used""" sess = create_session() def go(): eq_(sess.query(Person).all(), all_employees) self.assert_sql_count(testing.db, go, {'':14, 'Polymorphic':9}.get(select_type, 10)) def test_foo(self): sess = create_session() def go(): eq_(sess.query(Person).options(subqueryload(Engineer.machines)).all(), all_employees) self.assert_sql_count(testing.db, go, {'':14, 'Unions':8, 'Polymorphic':7}.get(select_type, 8)) def test_primary_eager_aliasing(self): sess = create_session() # for both joinedload() and subqueryload(), if the original q is not loading # the subclass table, the joinedload doesn't happen. def go(): eq_(sess.query(Person).options(joinedload(Engineer.machines))[1:3], all_employees[1:3]) self.assert_sql_count(testing.db, go, {'':6, 'Polymorphic':3}.get(select_type, 4)) sess = create_session() def go(): eq_(sess.query(Person).options(subqueryload(Engineer.machines)).all(), all_employees) self.assert_sql_count(testing.db, go, {'':14, 'Unions':8, 'Polymorphic':7}.get(select_type, 8)) sess = create_session() # assert the JOINs dont over JOIN assert sess.query(Person).with_polymorphic('*').options(joinedload(Engineer.machines)).\ limit(2).offset(1).with_labels().subquery().count().scalar() == 2 def go(): eq_( sess.query(Person).with_polymorphic('*'). options(joinedload(Engineer.machines))[1:3], all_employees[1:3]) self.assert_sql_count(testing.db, go, 3) def test_get(self): sess = create_session() # for all mappers, ensure the primary key has been calculated as just the "person_id" # column eq_(sess.query(Person).get(e1.person_id), Engineer(name="dilbert", primary_language="java")) eq_(sess.query(Engineer).get(e1.person_id), Engineer(name="dilbert", primary_language="java")) eq_(sess.query(Manager).get(b1.person_id), Boss(name="pointy haired boss", golf_swing="fore")) def test_multi_join(self): sess = create_session() e = aliased(Person) c = aliased(Company) q = sess.query(Company, Person, c, e).join((Person, Company.employees)).join((e, c.employees)).\ filter(Person.name=='dilbert').filter(e.name=='wally') eq_(q.count(), 1) eq_(q.all(), [ ( Company(company_id=1,name=u'MegaCorp, Inc.'), Engineer(status=u'regular engineer',engineer_name=u'dilbert',name=u'dilbert',company_id=1,primary_language=u'java',person_id=1,type=u'engineer'), Company(company_id=1,name=u'MegaCorp, Inc.'), Engineer(status=u'regular engineer',engineer_name=u'wally',name=u'wally',company_id=1,primary_language=u'c++',person_id=2,type=u'engineer') ) ]) def test_filter_on_subclass(self): sess = create_session() eq_(sess.query(Engineer).all()[0], Engineer(name="dilbert")) eq_(sess.query(Engineer).first(), Engineer(name="dilbert")) eq_(sess.query(Engineer).filter(Engineer.person_id==e1.person_id).first(), Engineer(name="dilbert")) eq_(sess.query(Manager).filter(Manager.person_id==m1.person_id).one(), Manager(name="dogbert")) eq_(sess.query(Manager).filter(Manager.person_id==b1.person_id).one(), Boss(name="pointy haired boss")) eq_(sess.query(Boss).filter(Boss.person_id==b1.person_id).one(), Boss(name="pointy haired boss")) def test_join_from_polymorphic(self): sess = create_session() for aliased in (True, False): eq_(sess.query(Person).join('paperwork', aliased=aliased).filter(Paperwork.description.like('%review%')).all(), [b1, m1]) eq_(sess.query(Person).join('paperwork', aliased=aliased).filter(Paperwork.description.like('%#2%')).all(), [e1, m1]) eq_(sess.query(Engineer).join('paperwork', aliased=aliased).filter(Paperwork.description.like('%#2%')).all(), [e1]) eq_(sess.query(Person).join('paperwork', aliased=aliased).filter(Person.name.like('%dog%')).filter(Paperwork.description.like('%#2%')).all(), [m1]) def test_join_from_with_polymorphic(self): sess = create_session() for aliased in (True, False): sess.expunge_all() eq_(sess.query(Person).with_polymorphic(Manager).join('paperwork', aliased=aliased).filter(Paperwork.description.like('%review%')).all(), [b1, m1]) sess.expunge_all() eq_(sess.query(Person).with_polymorphic([Manager, Engineer]).join('paperwork', aliased=aliased).filter(Paperwork.description.like('%#2%')).all(), [e1, m1]) sess.expunge_all() eq_(sess.query(Person).with_polymorphic([Manager, Engineer]).join('paperwork', aliased=aliased).filter(Person.name.like('%dog%')).filter(Paperwork.description.like('%#2%')).all(), [m1]) def test_join_to_polymorphic(self): sess = create_session() eq_(sess.query(Company).join('employees').filter(Person.name=='vlad').one(), c2) eq_(sess.query(Company).join('employees', aliased=True).filter(Person.name=='vlad').one(), c2) def test_polymorphic_any(self): sess = create_session() eq_( sess.query(Company).\ filter(Company.employees.any(Person.name=='vlad')).all(), [c2] ) # test that the aliasing on "Person" does not bleed into the # EXISTS clause generated by any() eq_( sess.query(Company).join(Company.employees, aliased=True).filter(Person.name=='dilbert').\ filter(Company.employees.any(Person.name=='wally')).all(), [c1] ) eq_( sess.query(Company).join(Company.employees, aliased=True).filter(Person.name=='dilbert').\ filter(Company.employees.any(Person.name=='vlad')).all(), [] ) eq_( sess.query(Company).filter(Company.employees.of_type(Engineer).any(Engineer.primary_language=='cobol')).one(), c2 ) calias = aliased(Company) eq_( sess.query(calias).filter(calias.employees.of_type(Engineer).any(Engineer.primary_language=='cobol')).one(), c2 ) eq_( sess.query(Company).filter(Company.employees.of_type(Boss).any(Boss.golf_swing=='fore')).one(), c1 ) eq_( sess.query(Company).filter(Company.employees.of_type(Boss).any(Manager.manager_name=='pointy')).one(), c1 ) if select_type != '': eq_( sess.query(Person).filter(Engineer.machines.any(Machine.name=="Commodore 64")).all(), [e2, e3] ) eq_( sess.query(Person).filter(Person.paperwork.any(Paperwork.description=="review #2")).all(), [m1] ) eq_( sess.query(Company).filter(Company.employees.of_type(Engineer).any(and_(Engineer.primary_language=='cobol'))).one(), c2 ) def test_join_from_columns_or_subclass(self): sess = create_session() eq_( sess.query(Manager.name).order_by(Manager.name).all(), [(u'dogbert',), (u'pointy haired boss',)] ) eq_( sess.query(Manager.name).join((Paperwork, Manager.paperwork)).order_by(Manager.name).all(), [(u'dogbert',), (u'dogbert',), (u'pointy haired boss',)] ) eq_( sess.query(Person.name).join((Paperwork, Person.paperwork)).order_by(Person.name).all(), [(u'dilbert',), (u'dilbert',), (u'dogbert',), (u'dogbert',), (u'pointy haired boss',), (u'vlad',), (u'wally',), (u'wally',)] ) eq_( sess.query(Person.name).join((paperwork, Manager.person_id==paperwork.c.person_id)).order_by(Person.name).all(), [(u'dilbert',), (u'dilbert',), (u'dogbert',), (u'dogbert',), (u'pointy haired boss',), (u'vlad',), (u'wally',), (u'wally',)] ) eq_( sess.query(Manager).join((Paperwork, Manager.paperwork)).order_by(Manager.name).all(), [m1, b1] ) eq_( sess.query(Manager.name).join((paperwork, Manager.person_id==paperwork.c.person_id)).order_by(Manager.name).all(), [(u'dogbert',), (u'dogbert',), (u'pointy haired boss',)] ) eq_( sess.query(Manager.person_id).join((paperwork, Manager.person_id==paperwork.c.person_id)).order_by(Manager.name).all(), [(4,), (4,), (3,)] ) eq_( sess.query(Manager.name, Paperwork.description). join((Paperwork, Manager.person_id==Paperwork.person_id)). order_by(Paperwork.paperwork_id). all(), [(u'pointy haired boss', u'review #1'), (u'dogbert', u'review #2'), (u'dogbert', u'review #3')] ) malias = aliased(Manager) eq_( sess.query(malias.name).join((paperwork, malias.person_id==paperwork.c.person_id)).all(), [(u'pointy haired boss',), (u'dogbert',), (u'dogbert',)] ) def test_polymorphic_option(self): """test that polymorphic loading sets state.load_path with its actual mapper on a subclass, and not the superclass mapper. """ paths = [] class MyOption(interfaces.MapperOption): propagate_to_loaders = True def process_query_conditionally(self, query): paths.append(query._current_path) sess = create_session() dilbert, boss = sess.query(Person).\ options(MyOption()).\ filter(Person.name.in_(['dilbert', 'pointy haired boss'])).\ order_by(Person.name).\ all() dilbert.machines boss.paperwork eq_(paths, [(class_mapper(Engineer), 'machines'), (class_mapper(Boss), 'paperwork')]) def test_expire(self): """test that individual column refresh doesn't get tripped up by the select_table mapper""" sess = create_session() m1 = sess.query(Manager).filter(Manager.name=='dogbert').one() sess.expire(m1) assert m1.status == 'regular manager' m2 = sess.query(Manager).filter(Manager.name=='pointy haired boss').one() sess.expire(m2, ['manager_name', 'golf_swing']) assert m2.golf_swing=='fore' def test_with_polymorphic(self): sess = create_session() assert_raises(sa_exc.InvalidRequestError, sess.query(Person).with_polymorphic, Paperwork) assert_raises(sa_exc.InvalidRequestError, sess.query(Engineer).with_polymorphic, Boss) assert_raises(sa_exc.InvalidRequestError, sess.query(Engineer).with_polymorphic, Person) # compare to entities without related collections to prevent additional lazy SQL from firing on # loaded entities emps_without_relationships = [ Engineer(name="dilbert", engineer_name="dilbert", primary_language="java", status="regular engineer"), Engineer(name="wally", engineer_name="wally", primary_language="c++", status="regular engineer"), Boss(name="pointy haired boss", golf_swing="fore", manager_name="pointy", status="da boss"), Manager(name="dogbert", manager_name="dogbert", status="regular manager"), Engineer(name="vlad", engineer_name="vlad", primary_language="cobol", status="elbonian engineer") ] eq_(sess.query(Person).with_polymorphic('*').all(), emps_without_relationships) def go(): eq_(sess.query(Person).with_polymorphic(Engineer).filter(Engineer.primary_language=='java').all(), emps_without_relationships[0:1]) self.assert_sql_count(testing.db, go, 1) sess.expunge_all() def go(): eq_(sess.query(Person).with_polymorphic('*').all(), emps_without_relationships) self.assert_sql_count(testing.db, go, 1) sess.expunge_all() def go(): eq_(sess.query(Person).with_polymorphic(Engineer).all(), emps_without_relationships) self.assert_sql_count(testing.db, go, 3) sess.expunge_all() def go(): eq_(sess.query(Person).with_polymorphic(Engineer, people.outerjoin(engineers)).all(), emps_without_relationships) self.assert_sql_count(testing.db, go, 3) sess.expunge_all() def go(): # limit the polymorphic join down to just "Person", overriding select_table eq_(sess.query(Person).with_polymorphic(Person).all(), emps_without_relationships) self.assert_sql_count(testing.db, go, 6) def test_relationship_to_polymorphic(self): assert_result = [ Company(name="MegaCorp, Inc.", employees=[ Engineer(name="dilbert", engineer_name="dilbert", primary_language="java", status="regular engineer", machines=[Machine(name="IBM ThinkPad"), Machine(name="IPhone")]), Engineer(name="wally", engineer_name="wally", primary_language="c++", status="regular engineer"), Boss(name="pointy haired boss", golf_swing="fore", manager_name="pointy", status="da boss"), Manager(name="dogbert", manager_name="dogbert", status="regular manager"), ]), Company(name="Elbonia, Inc.", employees=[ Engineer(name="vlad", engineer_name="vlad", primary_language="cobol", status="elbonian engineer") ]) ] sess = create_session() def go(): # test load Companies with lazy load to 'employees' eq_(sess.query(Company).all(), assert_result) self.assert_sql_count(testing.db, go, {'':9, 'Polymorphic':4}.get(select_type, 5)) sess = create_session() def go(): # currently, it doesn't matter if we say Company.employees, # or Company.employees.of_type(Engineer). joinedloader doesn't # pick up on the "of_type()" as of yet. eq_( sess.query(Company).options( joinedload_all(Company.employees.of_type(Engineer), Engineer.machines )).all(), assert_result) # in the case of select_type='', the joinedload # doesn't take in this case; it joinedloads company->people, # then a load for each of 5 rows, then lazyload of "machines" self.assert_sql_count(testing.db, go, {'':7, 'Polymorphic':1}.get(select_type, 2) ) sess = create_session() def go(): eq_( sess.query(Company).options( subqueryload_all(Company.employees.of_type(Engineer), Engineer.machines )).all(), assert_result) self.assert_sql_count( testing.db, go, {'':8, 'Joins':4, 'Unions':4, 'Polymorphic':3, 'AliasedJoins':4}[select_type] ) def test_joinedload_on_subclass(self): sess = create_session() def go(): # test load People with joinedload to engineers + machines eq_(sess.query(Person).with_polymorphic('*').options(joinedload(Engineer.machines)).filter(Person.name=='dilbert').all(), [Engineer(name="dilbert", engineer_name="dilbert", primary_language="java", status="regular engineer", machines=[Machine(name="IBM ThinkPad"), Machine(name="IPhone")])] ) self.assert_sql_count(testing.db, go, 1) sess = create_session() def go(): # test load People with subqueryload to engineers + machines eq_(sess.query(Person).with_polymorphic('*').options(subqueryload(Engineer.machines)).filter(Person.name=='dilbert').all(), [Engineer(name="dilbert", engineer_name="dilbert", primary_language="java", status="regular engineer", machines=[Machine(name="IBM ThinkPad"), Machine(name="IPhone")])] ) self.assert_sql_count(testing.db, go, 2) def test_query_subclass_join_to_base_relationship(self): sess = create_session() # non-polymorphic eq_(sess.query(Engineer).join(Person.paperwork).all(), [e1, e2, e3]) def test_join_to_subclass(self): sess = create_session() eq_(sess.query(Company).join(('employees', people.join(engineers))).filter(Engineer.primary_language=='java').all(), [c1]) if select_type == '': eq_(sess.query(Company).select_from(companies.join(people).join(engineers)).filter(Engineer.primary_language=='java').all(), [c1]) eq_(sess.query(Company).join(('employees', people.join(engineers))).filter(Engineer.primary_language=='java').all(), [c1]) ealias = aliased(Engineer) eq_(sess.query(Company).join(('employees', ealias)).filter(ealias.primary_language=='java').all(), [c1]) eq_(sess.query(Person).select_from(people.join(engineers)).join(Engineer.machines).all(), [e1, e2, e3]) eq_(sess.query(Person).select_from(people.join(engineers)).join(Engineer.machines).filter(Machine.name.ilike("%ibm%")).all(), [e1, e3]) eq_(sess.query(Company).join(('employees', people.join(engineers)), Engineer.machines).all(), [c1, c2]) eq_(sess.query(Company).join(('employees', people.join(engineers)), Engineer.machines).filter(Machine.name.ilike("%thinkpad%")).all(), [c1]) else: eq_(sess.query(Company).select_from(companies.join(people).join(engineers)).filter(Engineer.primary_language=='java').all(), [c1]) eq_(sess.query(Company).join('employees').filter(Engineer.primary_language=='java').all(), [c1]) eq_(sess.query(Person).join(Engineer.machines).all(), [e1, e2, e3]) eq_(sess.query(Person).join(Engineer.machines).filter(Machine.name.ilike("%ibm%")).all(), [e1, e3]) eq_(sess.query(Company).join('employees', Engineer.machines).all(), [c1, c2]) eq_(sess.query(Company).join('employees', Engineer.machines).filter(Machine.name.ilike("%thinkpad%")).all(), [c1]) # non-polymorphic eq_(sess.query(Engineer).join(Engineer.machines).all(), [e1, e2, e3]) eq_(sess.query(Engineer).join(Engineer.machines).filter(Machine.name.ilike("%ibm%")).all(), [e1, e3]) # here's the new way eq_(sess.query(Company).join(Company.employees.of_type(Engineer)).filter(Engineer.primary_language=='java').all(), [c1]) eq_(sess.query(Company).join(Company.employees.of_type(Engineer), 'machines').filter(Machine.name.ilike("%thinkpad%")).all(), [c1]) def test_join_through_polymorphic(self): sess = create_session() for aliased in (True, False): eq_( sess.query(Company).\ join('employees', 'paperwork', aliased=aliased).filter(Paperwork.description.like('%#2%')).all(), [c1] ) eq_( sess.query(Company).\ join('employees', 'paperwork', aliased=aliased).filter(Paperwork.description.like('%#%')).all(), [c1, c2] ) eq_( sess.query(Company).\ join('employees', 'paperwork', aliased=aliased).filter(Person.name.in_(['dilbert', 'vlad'])).filter(Paperwork.description.like('%#2%')).all(), [c1] ) eq_( sess.query(Company).\ join('employees', 'paperwork', aliased=aliased).filter(Person.name.in_(['dilbert', 'vlad'])).filter(Paperwork.description.like('%#%')).all(), [c1, c2] ) eq_( sess.query(Company).join('employees', aliased=aliased).filter(Person.name.in_(['dilbert', 'vlad'])).\ join('paperwork', from_joinpoint=True, aliased=aliased).filter(Paperwork.description.like('%#2%')).all(), [c1] ) eq_( sess.query(Company).join('employees', aliased=aliased).filter(Person.name.in_(['dilbert', 'vlad'])).\ join('paperwork', from_joinpoint=True, aliased=aliased).filter(Paperwork.description.like('%#%')).all(), [c1, c2] ) def test_explicit_polymorphic_join(self): sess = create_session() # join from Company to Engineer; join condition formulated by # ORMJoin using regular table foreign key connections. Engineer # is expressed as "(select * people join engineers) as anon_1" # so the join is contained. eq_( sess.query(Company).join(Engineer).filter(Engineer.engineer_name=='vlad').one(), c2 ) # same, using explicit join condition. Query.join() must adapt the on clause # here to match the subquery wrapped around "people join engineers". eq_( sess.query(Company).join((Engineer, Company.company_id==Engineer.company_id)).filter(Engineer.engineer_name=='vlad').one(), c2 ) def test_filter_on_baseclass(self): sess = create_session() eq_(sess.query(Person).all(), all_employees) eq_(sess.query(Person).first(), all_employees[0]) eq_(sess.query(Person).filter(Person.person_id==e2.person_id).one(), e2) def test_from_alias(self): sess = create_session() palias = aliased(Person) eq_( sess.query(palias).filter(palias.name.in_(['dilbert', 'wally'])).all(), [e1, e2] ) def test_self_referential(self): sess = create_session() c1_employees = [e1, e2, b1, m1] palias = aliased(Person) eq_( sess.query(Person, palias).filter(Person.company_id==palias.company_id).filter(Person.name=='dogbert').\ filter(Person.person_id>palias.person_id).order_by(Person.person_id, palias.person_id).all(), [ (m1, e1), (m1, e2), (m1, b1), ] ) eq_( sess.query(Person, palias).filter(Person.company_id==palias.company_id).filter(Person.name=='dogbert').\ filter(Person.person_id>palias.person_id).from_self().order_by(Person.person_id, palias.person_id).all(), [ (m1, e1), (m1, e2), (m1, b1), ] ) def test_nesting_queries(self): sess = create_session() # query.statement places a flag "no_adapt" on the returned statement. This prevents # the polymorphic adaptation in the second "filter" from hitting it, which would pollute # the subquery and usually results in recursion overflow errors within the adaption. subq = sess.query(engineers.c.person_id).filter(Engineer.primary_language=='java').statement.as_scalar() eq_(sess.query(Person).filter(Person.person_id==subq).one(), e1) def test_mixed_entities(self): sess = create_session() eq_( sess.query(Company.name, Person).join(Company.employees).filter(Company.name=='Elbonia, Inc.').all(), [(u'Elbonia, Inc.', Engineer(status=u'elbonian engineer',engineer_name=u'vlad',name=u'vlad',primary_language=u'cobol'))] ) eq_( sess.query(Person, Company.name).join(Company.employees).filter(Company.name=='Elbonia, Inc.').all(), [(Engineer(status=u'elbonian engineer',engineer_name=u'vlad',name=u'vlad',primary_language=u'cobol'), u'Elbonia, Inc.')] ) eq_( sess.query(Manager.name).all(), [('pointy haired boss', ), ('dogbert',)] ) eq_( sess.query(Manager.name + " foo").all(), [('pointy haired boss foo', ), ('dogbert foo',)] ) row = sess.query(Engineer.name, Engineer.primary_language).filter(Engineer.name=='dilbert').first() assert row.name == 'dilbert' assert row.primary_language == 'java' eq_( sess.query(Engineer.name, Engineer.primary_language).all(), [(u'dilbert', u'java'), (u'wally', u'c++'), (u'vlad', u'cobol')] ) eq_( sess.query(Boss.name, Boss.golf_swing).all(), [(u'pointy haired boss', u'fore')] ) # TODO: I think raise error on these for now. different inheritance/loading schemes have different # results here, all incorrect # # self.assertEquals( # sess.query(Person.name, Engineer.primary_language).all(), # [] # ) # self.assertEquals( # sess.query(Person.name, Engineer.primary_language, Manager.manager_name).all(), # [] # ) eq_( sess.query(Person.name, Company.name).join(Company.employees).filter(Company.name=='Elbonia, Inc.').all(), [(u'vlad',u'Elbonia, Inc.')] ) eq_( sess.query(Engineer.primary_language).filter(Person.type=='engineer').all(), [(u'java',), (u'c++',), (u'cobol',)] ) if select_type != '': eq_( sess.query(Engineer, Company.name).join(Company.employees).filter(Person.type=='engineer').all(), [ (Engineer(status=u'regular engineer',engineer_name=u'dilbert',name=u'dilbert',company_id=1,primary_language=u'java',person_id=1,type=u'engineer'), u'MegaCorp, Inc.'), (Engineer(status=u'regular engineer',engineer_name=u'wally',name=u'wally',company_id=1,primary_language=u'c++',person_id=2,type=u'engineer'), u'MegaCorp, Inc.'), (Engineer(status=u'elbonian engineer',engineer_name=u'vlad',name=u'vlad',company_id=2,primary_language=u'cobol',person_id=5,type=u'engineer'), u'Elbonia, Inc.') ] ) eq_( sess.query(Engineer.primary_language, Company.name).join(Company.employees).filter(Person.type=='engineer').order_by(desc(Engineer.primary_language)).all(), [(u'java', u'MegaCorp, Inc.'), (u'cobol', u'Elbonia, Inc.'), (u'c++', u'MegaCorp, Inc.')] ) palias = aliased(Person) eq_( sess.query(Person, Company.name, palias).join(Company.employees).filter(Company.name=='Elbonia, Inc.').filter(palias.name=='dilbert').all(), [(Engineer(status=u'elbonian engineer',engineer_name=u'vlad',name=u'vlad',primary_language=u'cobol'), u'Elbonia, Inc.', Engineer(status=u'regular engineer',engineer_name=u'dilbert',name=u'dilbert',company_id=1,primary_language=u'java',person_id=1,type=u'engineer'))] ) eq_( sess.query(palias, Company.name, Person).join(Company.employees).filter(Company.name=='Elbonia, Inc.').filter(palias.name=='dilbert').all(), [(Engineer(status=u'regular engineer',engineer_name=u'dilbert',name=u'dilbert',company_id=1,primary_language=u'java',person_id=1,type=u'engineer'), u'Elbonia, Inc.', Engineer(status=u'elbonian engineer',engineer_name=u'vlad',name=u'vlad',primary_language=u'cobol'),) ] ) eq_( sess.query(Person.name, Company.name, palias.name).join(Company.employees).filter(Company.name=='Elbonia, Inc.').filter(palias.name=='dilbert').all(), [(u'vlad', u'Elbonia, Inc.', u'dilbert')] ) palias = aliased(Person) eq_( sess.query(Person.type, Person.name, palias.type, palias.name).filter(Person.company_id==palias.company_id).filter(Person.name=='dogbert').\ filter(Person.person_id>palias.person_id).order_by(Person.person_id, palias.person_id).all(), [(u'manager', u'dogbert', u'engineer', u'dilbert'), (u'manager', u'dogbert', u'engineer', u'wally'), (u'manager', u'dogbert', u'boss', u'pointy haired boss')] ) eq_( sess.query(Person.name, Paperwork.description).filter(Person.person_id==Paperwork.person_id).order_by(Person.name, Paperwork.description).all(), [(u'dilbert', u'tps report #1'), (u'dilbert', u'tps report #2'), (u'dogbert', u'review #2'), (u'dogbert', u'review #3'), (u'pointy haired boss', u'review #1'), (u'vlad', u'elbonian missive #3'), (u'wally', u'tps report #3'), (u'wally', u'tps report #4'), ] ) if select_type != '': eq_( sess.query(func.count(Person.person_id)).filter(Engineer.primary_language=='java').all(), [(1, )] ) eq_( sess.query(Company.name, func.count(Person.person_id)).filter(Company.company_id==Person.company_id).group_by(Company.name).order_by(Company.name).all(), [(u'Elbonia, Inc.', 1), (u'MegaCorp, Inc.', 4)] ) eq_( sess.query(Company.name, func.count(Person.person_id)).join(Company.employees).group_by(Company.name).order_by(Company.name).all(), [(u'Elbonia, Inc.', 1), (u'MegaCorp, Inc.', 4)] ) PolymorphicQueryTest.__name__ = "Polymorphic%sTest" % select_type return PolymorphicQueryTest for select_type in ('', 'Polymorphic', 'Unions', 'AliasedJoins', 'Joins'): testclass = _produce_test(select_type) exec("%s = testclass" % testclass.__name__) del testclass class SelfReferentialTestJoinedToBase(_base.MappedTest): run_setup_mappers = 'once' @classmethod def define_tables(cls, metadata): global people, engineers people = Table('people', metadata, Column('person_id', Integer, primary_key=True, test_needs_autoincrement=True), Column('name', String(50)), Column('type', String(30))) engineers = Table('engineers', metadata, Column('person_id', Integer, ForeignKey('people.person_id'), primary_key=True), Column('primary_language', String(50)), Column('reports_to_id', Integer, ForeignKey('people.person_id')) ) @classmethod def setup_mappers(cls): mapper(Person, people, polymorphic_on=people.c.type, polymorphic_identity='person') mapper(Engineer, engineers, inherits=Person, inherit_condition=engineers.c.person_id==people.c.person_id, polymorphic_identity='engineer', properties={ 'reports_to':relationship(Person, primaryjoin=people.c.person_id==engineers.c.reports_to_id) }) def test_has(self): p1 = Person(name='dogbert') e1 = Engineer(name='dilbert', primary_language='java', reports_to=p1) sess = create_session() sess.add(p1) sess.add(e1) sess.flush() sess.expunge_all() eq_(sess.query(Engineer).filter(Engineer.reports_to.has(Person.name=='dogbert')).first(), Engineer(name='dilbert')) def test_oftype_aliases_in_exists(self): e1 = Engineer(name='dilbert', primary_language='java') e2 = Engineer(name='wally', primary_language='c++', reports_to=e1) sess = create_session() sess.add_all([e1, e2]) sess.flush() eq_(sess.query(Engineer).filter(Engineer.reports_to.of_type(Engineer).has(Engineer.name=='dilbert')).first(), e2) def test_join(self): p1 = Person(name='dogbert') e1 = Engineer(name='dilbert', primary_language='java', reports_to=p1) sess = create_session() sess.add(p1) sess.add(e1) sess.flush() sess.expunge_all() eq_( sess.query(Engineer).join('reports_to', aliased=True).filter(Person.name=='dogbert').first(), Engineer(name='dilbert')) class SelfReferentialJ2JTest(_base.MappedTest): run_setup_mappers = 'once' @classmethod def define_tables(cls, metadata): global people, engineers, managers people = Table('people', metadata, Column('person_id', Integer, primary_key=True, test_needs_autoincrement=True), Column('name', String(50)), Column('type', String(30))) engineers = Table('engineers', metadata, Column('person_id', Integer, ForeignKey('people.person_id'), primary_key=True), Column('primary_language', String(50)), Column('reports_to_id', Integer, ForeignKey('managers.person_id')) ) managers = Table('managers', metadata, Column('person_id', Integer, ForeignKey('people.person_id'), primary_key=True), ) @classmethod def setup_mappers(cls): mapper(Person, people, polymorphic_on=people.c.type, polymorphic_identity='person') mapper(Manager, managers, inherits=Person, polymorphic_identity='manager') mapper(Engineer, engineers, inherits=Person, polymorphic_identity='engineer', properties={ 'reports_to':relationship(Manager, primaryjoin=managers.c.person_id==engineers.c.reports_to_id, backref='engineers') }) def test_has(self): m1 = Manager(name='dogbert') e1 = Engineer(name='dilbert', primary_language='java', reports_to=m1) sess = create_session() sess.add(m1) sess.add(e1) sess.flush() sess.expunge_all() eq_(sess.query(Engineer).filter(Engineer.reports_to.has(Manager.name=='dogbert')).first(), Engineer(name='dilbert')) def test_join(self): m1 = Manager(name='dogbert') e1 = Engineer(name='dilbert', primary_language='java', reports_to=m1) sess = create_session() sess.add(m1) sess.add(e1) sess.flush() sess.expunge_all() eq_( sess.query(Engineer).join('reports_to', aliased=True).filter(Manager.name=='dogbert').first(), Engineer(name='dilbert')) def test_filter_aliasing(self): m1 = Manager(name='dogbert') m2 = Manager(name='foo') e1 = Engineer(name='wally', primary_language='java', reports_to=m1) e2 = Engineer(name='dilbert', primary_language='c++', reports_to=m2) e3 = Engineer(name='etc', primary_language='c++') sess = create_session() sess.add_all([m1, m2, e1, e2, e3]) sess.flush() sess.expunge_all() # filter aliasing applied to Engineer doesn't whack Manager eq_( sess.query(Manager).join(Manager.engineers).filter(Manager.name=='dogbert').all(), [m1] ) eq_( sess.query(Manager).join(Manager.engineers).filter(Engineer.name=='dilbert').all(), [m2] ) eq_( sess.query(Manager, Engineer).join(Manager.engineers).order_by(Manager.name.desc()).all(), [ (m2, e2), (m1, e1), ] ) def test_relationship_compare(self): m1 = Manager(name='dogbert') m2 = Manager(name='foo') e1 = Engineer(name='dilbert', primary_language='java', reports_to=m1) e2 = Engineer(name='wally', primary_language='c++', reports_to=m2) e3 = Engineer(name='etc', primary_language='c++') sess = create_session() sess.add(m1) sess.add(m2) sess.add(e1) sess.add(e2) sess.add(e3) sess.flush() sess.expunge_all() eq_( sess.query(Manager).join(Manager.engineers).filter(Engineer.reports_to==None).all(), [] ) eq_( sess.query(Manager).join(Manager.engineers).filter(Engineer.reports_to==m1).all(), [m1] ) class M2MFilterTest(_base.MappedTest): run_setup_mappers = 'once' run_inserts = 'once' run_deletes = None @classmethod def define_tables(cls, metadata): global people, engineers, organizations, engineers_to_org organizations = Table('organizations', metadata, Column('id', Integer, primary_key=True, test_needs_autoincrement=True), Column('name', String(50)), ) engineers_to_org = Table('engineers_org', metadata, Column('org_id', Integer, ForeignKey('organizations.id')), Column('engineer_id', Integer, ForeignKey('engineers.person_id')), ) people = Table('people', metadata, Column('person_id', Integer, primary_key=True, test_needs_autoincrement=True), Column('name', String(50)), Column('type', String(30))) engineers = Table('engineers', metadata, Column('person_id', Integer, ForeignKey('people.person_id'), primary_key=True), Column('primary_language', String(50)), ) @classmethod def setup_mappers(cls): global Organization class Organization(_fixtures.Base): pass mapper(Organization, organizations, properties={ 'engineers':relationship(Engineer, secondary=engineers_to_org, backref='organizations') }) mapper(Person, people, polymorphic_on=people.c.type, polymorphic_identity='person') mapper(Engineer, engineers, inherits=Person, polymorphic_identity='engineer') @classmethod def insert_data(cls): e1 = Engineer(name='e1') e2 = Engineer(name='e2') e3 = Engineer(name='e3') e4 = Engineer(name='e4') org1 = Organization(name='org1', engineers=[e1, e2]) org2 = Organization(name='org2', engineers=[e3, e4]) sess = create_session() sess.add(org1) sess.add(org2) sess.flush() def test_not_contains(self): sess = create_session() e1 = sess.query(Person).filter(Engineer.name=='e1').one() # this works eq_(sess.query(Organization).filter(~Organization.engineers.of_type(Engineer).contains(e1)).all(), [Organization(name='org2')]) # this had a bug eq_(sess.query(Organization).filter(~Organization.engineers.contains(e1)).all(), [Organization(name='org2')]) def test_any(self): sess = create_session() eq_(sess.query(Organization).filter(Organization.engineers.of_type(Engineer).any(Engineer.name=='e1')).all(), [Organization(name='org1')]) eq_(sess.query(Organization).filter(Organization.engineers.any(Engineer.name=='e1')).all(), [Organization(name='org1')]) class SelfReferentialM2MTest(_base.MappedTest, AssertsCompiledSQL): run_setup_mappers = 'once' @classmethod def define_tables(cls, metadata): global Parent, Child1, Child2 Base = declarative_base(metadata=metadata) secondary_table = Table('secondary', Base.metadata, Column('left_id', Integer, ForeignKey('parent.id'), nullable=False), Column('right_id', Integer, ForeignKey('parent.id'), nullable=False)) class Parent(Base): __tablename__ = 'parent' id = Column(Integer, primary_key=True, test_needs_autoincrement=True) cls = Column(String(50)) __mapper_args__ = dict(polymorphic_on = cls ) class Child1(Parent): __tablename__ = 'child1' id = Column(Integer, ForeignKey('parent.id'), primary_key=True) __mapper_args__ = dict(polymorphic_identity = 'child1') class Child2(Parent): __tablename__ = 'child2' id = Column(Integer, ForeignKey('parent.id'), primary_key=True) __mapper_args__ = dict(polymorphic_identity = 'child2') Child1.left_child2 = relationship(Child2, secondary = secondary_table, primaryjoin = Parent.id == secondary_table.c.right_id, secondaryjoin = Parent.id == secondary_table.c.left_id, uselist = False, backref="right_children" ) def test_query_crit(self): session = create_session() c11, c12, c13 = Child1(), Child1(), Child1() c21, c22, c23 = Child2(), Child2(), Child2() c11.left_child2 = c22 c12.left_child2 = c22 c13.left_child2 = c23 session.add_all([c11, c12, c13, c21, c22, c23]) session.flush() # test that the join to Child2 doesn't alias Child1 in the select eq_( set(session.query(Child1).join(Child1.left_child2)), set([c11, c12, c13]) ) eq_( set(session.query(Child1, Child2).join(Child1.left_child2)), set([(c11, c22), (c12, c22), (c13, c23)]) ) # test __eq__() on property is annotating correctly eq_( set(session.query(Child2).join(Child2.right_children).filter(Child1.left_child2==c22)), set([c22]) ) # test the same again self.assert_compile( session.query(Child2).join(Child2.right_children).filter(Child1.left_child2==c22).with_labels().statement, "SELECT parent.id AS parent_id, child2.id AS child2_id, parent.cls AS parent_cls FROM " "secondary AS secondary_1, parent JOIN child2 ON parent.id = child2.id JOIN secondary AS secondary_2 " "ON parent.id = secondary_2.left_id JOIN (SELECT parent.id AS parent_id, parent.cls AS parent_cls, " "child1.id AS child1_id FROM parent JOIN child1 ON parent.id = child1.id) AS anon_1 ON " "anon_1.parent_id = secondary_2.right_id WHERE anon_1.parent_id = secondary_1.right_id AND :param_1 = secondary_1.left_id", dialect=default.DefaultDialect() ) def test_eager_join(self): session = create_session() c1 = Child1() c1.left_child2 = Child2() session.add(c1) session.flush() q = session.query(Child1).options(joinedload('left_child2')) # test that the splicing of the join works here, doesnt break in the middle of "parent join child1" self.assert_compile(q.limit(1).with_labels().statement, "SELECT anon_1.parent_id AS anon_1_parent_id, anon_1.child1_id AS anon_1_child1_id, "\ "anon_1.parent_cls AS anon_1_parent_cls, anon_2.parent_id AS anon_2_parent_id, "\ "anon_2.child2_id AS anon_2_child2_id, anon_2.parent_cls AS anon_2_parent_cls FROM "\ "(SELECT parent.id AS parent_id, child1.id AS child1_id, parent.cls AS parent_cls FROM parent "\ "JOIN child1 ON parent.id = child1.id LIMIT 1) AS anon_1 LEFT OUTER JOIN secondary AS secondary_1 "\ "ON anon_1.parent_id = secondary_1.right_id LEFT OUTER JOIN (SELECT parent.id AS parent_id, "\ "parent.cls AS parent_cls, child2.id AS child2_id FROM parent JOIN child2 ON parent.id = child2.id) "\ "AS anon_2 ON anon_2.parent_id = secondary_1.left_id" , dialect=default.DefaultDialect()) # another way to check assert q.limit(1).with_labels().subquery().count().scalar() == 1 assert q.first() is c1 def test_subquery_load(self): session = create_session() c1 = Child1() c1.left_child2 = Child2() session.add(c1) session.flush() session.expunge_all() for row in session.query(Child1).options(subqueryload('left_child2')).all(): assert row.left_child2 class EagerToSubclassTest(_base.MappedTest): """Test joinedloads to subclass mappers""" run_setup_classes = 'once' run_setup_mappers = 'once' run_inserts = 'once' run_deletes = None @classmethod def define_tables(cls, metadata): Table('parent', metadata, Column('id', Integer, primary_key=True, test_needs_autoincrement=True), Column('data', String(10)), ) Table('base', metadata, Column('id', Integer, primary_key=True, test_needs_autoincrement=True), Column('type', String(10)), ) Table('sub', metadata, Column('id', Integer, ForeignKey('base.id'), primary_key=True), Column('data', String(10)), Column('parent_id', Integer, ForeignKey('parent.id'), nullable=False) ) @classmethod @testing.resolve_artifact_names def setup_classes(cls): class Parent(_base.ComparableEntity): pass class Base(_base.ComparableEntity): pass class Sub(Base): pass @classmethod @testing.resolve_artifact_names def setup_mappers(cls): mapper(Parent, parent, properties={ 'children':relationship(Sub) }) mapper(Base, base, polymorphic_on=base.c.type, polymorphic_identity='b') mapper(Sub, sub, inherits=Base, polymorphic_identity='s') @classmethod @testing.resolve_artifact_names def insert_data(cls): sess = create_session() p1 = Parent(data='p1', children=[Sub(data='s1'), Sub(data='s2'), Sub(data='s3')]) p2 = Parent(data='p2', children=[Sub(data='s4'), Sub(data='s5')]) sess.add(p1) sess.add(p2) sess.flush() @testing.resolve_artifact_names def test_joinedload(self): sess = create_session() def go(): eq_( sess.query(Parent).options(joinedload(Parent.children)).all(), [ Parent(data='p1', children=[Sub(data='s1'), Sub(data='s2'), Sub(data='s3')]), Parent(data='p2', children=[Sub(data='s4'), Sub(data='s5')]) ] ) self.assert_sql_count(testing.db, go, 1) @testing.resolve_artifact_names def test_contains_eager(self): sess = create_session() def go(): eq_( sess.query(Parent).join(Parent.children).options(contains_eager(Parent.children)).\ order_by(Parent.data, Sub.data).all(), [ Parent(data='p1', children=[Sub(data='s1'), Sub(data='s2'), Sub(data='s3')]), Parent(data='p2', children=[Sub(data='s4'), Sub(data='s5')]) ] ) self.assert_sql_count(testing.db, go, 1) class SubClassEagerToSubClassTest(_base.MappedTest): """Test joinedloads from subclass to subclass mappers""" run_setup_classes = 'once' run_setup_mappers = 'once' run_inserts = 'once' run_deletes = None @classmethod def define_tables(cls, metadata): Table('parent', metadata, Column('id', Integer, primary_key=True, test_needs_autoincrement=True), Column('type', String(10)), ) Table('subparent', metadata, Column('id', Integer, ForeignKey('parent.id'), primary_key=True), Column('data', String(10)), ) Table('base', metadata, Column('id', Integer, primary_key=True, test_needs_autoincrement=True), Column('type', String(10)), ) Table('sub', metadata, Column('id', Integer, ForeignKey('base.id'), primary_key=True), Column('data', String(10)), Column('subparent_id', Integer, ForeignKey('subparent.id'), nullable=False) ) @classmethod @testing.resolve_artifact_names def setup_classes(cls): class Parent(_base.ComparableEntity): pass class Subparent(Parent): pass class Base(_base.ComparableEntity): pass class Sub(Base): pass @classmethod @testing.resolve_artifact_names def setup_mappers(cls): mapper(Parent, parent, polymorphic_on=parent.c.type, polymorphic_identity='b') mapper(Subparent, subparent, inherits=Parent, polymorphic_identity='s', properties={ 'children':relationship(Sub, order_by=base.c.id) }) mapper(Base, base, polymorphic_on=base.c.type, polymorphic_identity='b') mapper(Sub, sub, inherits=Base, polymorphic_identity='s') @classmethod @testing.resolve_artifact_names def insert_data(cls): sess = create_session() p1 = Subparent(data='p1', children=[Sub(data='s1'), Sub(data='s2'), Sub(data='s3')]) p2 = Subparent(data='p2', children=[Sub(data='s4'), Sub(data='s5')]) sess.add(p1) sess.add(p2) sess.flush() @testing.resolve_artifact_names def test_joinedload(self): sess = create_session() def go(): eq_( sess.query(Subparent).options(joinedload(Subparent.children)).all(), [ Subparent(data='p1', children=[Sub(data='s1'), Sub(data='s2'), Sub(data='s3')]), Subparent(data='p2', children=[Sub(data='s4'), Sub(data='s5')]) ] ) self.assert_sql_count(testing.db, go, 1) sess.expunge_all() def go(): eq_( sess.query(Subparent).options(joinedload("children")).all(), [ Subparent(data='p1', children=[Sub(data='s1'), Sub(data='s2'), Sub(data='s3')]), Subparent(data='p2', children=[Sub(data='s4'), Sub(data='s5')]) ] ) self.assert_sql_count(testing.db, go, 1) @testing.resolve_artifact_names def test_contains_eager(self): sess = create_session() def go(): eq_( sess.query(Subparent).join(Subparent.children).options(contains_eager(Subparent.children)).all(), [ Subparent(data='p1', children=[Sub(data='s1'), Sub(data='s2'), Sub(data='s3')]), Subparent(data='p2', children=[Sub(data='s4'), Sub(data='s5')]) ] ) self.assert_sql_count(testing.db, go, 1) sess.expunge_all() def go(): eq_( sess.query(Subparent).join(Subparent.children).options(contains_eager("children")).all(), [ Subparent(data='p1', children=[Sub(data='s1'), Sub(data='s2'), Sub(data='s3')]), Subparent(data='p2', children=[Sub(data='s4'), Sub(data='s5')]) ] ) self.assert_sql_count(testing.db, go, 1) @testing.resolve_artifact_names def test_subqueryload(self): sess = create_session() def go(): eq_( sess.query(Subparent).options(subqueryload(Subparent.children)).all(), [ Subparent(data='p1', children=[Sub(data='s1'), Sub(data='s2'), Sub(data='s3')]), Subparent(data='p2', children=[Sub(data='s4'), Sub(data='s5')]) ] ) self.assert_sql_count(testing.db, go, 2) sess.expunge_all() def go(): eq_( sess.query(Subparent).options(subqueryload("children")).all(), [ Subparent(data='p1', children=[Sub(data='s1'), Sub(data='s2'), Sub(data='s3')]), Subparent(data='p2', children=[Sub(data='s4'), Sub(data='s5')]) ] ) self.assert_sql_count(testing.db, go, 2)