1
from sqlalchemy import Integer, String, ForeignKey, func, desc, and_, or_
2
from sqlalchemy.orm import interfaces, relationship, mapper, \
3
clear_mappers, create_session, joinedload, joinedload_all, \
4
subqueryload, subqueryload_all, polymorphic_union, aliased,\
1
from sqlalchemy import func, desc
2
from sqlalchemy.orm import interfaces, create_session, joinedload, joinedload_all, \
3
subqueryload, subqueryload_all, aliased,\
6
5
from sqlalchemy import exc as sa_exc
7
from sqlalchemy.engine import default
9
from test.lib import AssertsCompiledSQL, fixtures, testing
10
from test.lib.schema import Table, Column
11
from test.lib.testing import assert_raises, eq_
7
from sqlalchemy import testing
8
from sqlalchemy.testing import assert_raises, eq_
13
10
from _poly_fixtures import Company, Person, Engineer, Manager, Boss, \
14
Machine, Paperwork, _PolymorphicFixtureBase, _Polymorphic,\
11
Machine, Paperwork, _Polymorphic,\
15
12
_PolymorphicPolymorphic, _PolymorphicUnions, _PolymorphicJoins,\
16
13
_PolymorphicAliasedJoins
323
320
.filter(any_).all(),
326
def test_polymorphic_any_four(self):
327
sess = create_session()
328
any_ = Company.employees.of_type(Engineer).any(
329
Engineer.primary_language == 'cobol')
330
eq_(sess.query(Company).filter(any_).one(), c2)
332
def test_polymorphic_any_five(self):
333
sess = create_session()
334
calias = aliased(Company)
335
any_ = calias.employees.of_type(Engineer).any(
336
Engineer.primary_language == 'cobol')
337
eq_(sess.query(calias).filter(any_).one(), c2)
339
def test_polymorphic_any_six(self):
340
sess = create_session()
341
any_ = Company.employees.of_type(Boss).any(
342
Boss.golf_swing == 'fore')
343
eq_(sess.query(Company).filter(any_).one(), c1)
345
def test_polymorphic_any_seven(self):
346
sess = create_session()
347
any_ = Company.employees.of_type(Boss).any(
348
Manager.manager_name == 'pointy')
349
eq_(sess.query(Company).filter(any_).one(), c1)
351
323
def test_polymorphic_any_eight(self):
352
324
sess = create_session()
353
325
any_ = Engineer.machines.any(
360
332
Paperwork.description == "review #2")
361
333
eq_(sess.query(Person).filter(any_).all(), [m1])
363
def test_polymorphic_any_ten(self):
364
sess = create_session()
365
any_ = Company.employees.of_type(Engineer).any(
366
and_(Engineer.primary_language == 'cobol'))
367
eq_(sess.query(Company).filter(any_).one(), c2)
369
336
def test_join_from_columns_or_subclass_one(self):
370
337
sess = create_session()
532
def test_polymorphic_option(self):
534
Test that polymorphic loading sets state.load_path with its
535
actual mapper on a subclass, and not the superclass mapper.
539
class MyOption(interfaces.MapperOption):
540
propagate_to_loaders = True
541
def process_query_conditionally(self, query):
542
paths.append(query._current_path)
500
def test_subclass_option_pathing(self):
501
from sqlalchemy.orm import defer
544
502
sess = create_session()
545
names = ['dilbert', 'pointy haired boss']
549
.filter(Person.name.in_(names))
550
.order_by(Person.name).all())
556
[(class_mapper(Engineer), 'machines'),
557
(class_mapper(Boss), 'paperwork')])
503
dilbert = sess.query(Person).\
504
options(defer(Engineer.machines, Machine.name)).\
505
filter(Person.name == 'dilbert').first()
506
m = dilbert.machines[0]
507
assert 'name' not in m.__dict__
508
eq_(m.name, 'IBM ThinkPad')
559
510
def test_expire(self):
639
590
self._emps_wo_relationships_fixture())
642
def test_relationship_to_polymorphic(self):
645
name="MegaCorp, Inc.",
649
engineer_name="dilbert",
650
primary_language="java",
651
status="regular engineer",
653
Machine(name="IBM ThinkPad"),
654
Machine(name="IPhone")]),
657
engineer_name="wally",
658
primary_language="c++",
659
status="regular engineer"),
661
name="pointy haired boss",
663
manager_name="pointy",
667
manager_name="dogbert",
668
status="regular manager"),
671
name="Elbonia, Inc.",
675
engineer_name="vlad",
676
primary_language="cobol",
677
status="elbonian engineer")
593
def test_relationship_to_polymorphic_one(self):
594
expected = self._company_with_emps_machines_fixture()
681
595
sess = create_session()
683
597
# test load Companies with lazy load to 'employees'
684
598
eq_(sess.query(Company).all(), expected)
685
count = {'':9, 'Polymorphic':4}.get(self.select_type, 5)
599
count = {'':10, 'Polymorphic':5}.get(self.select_type, 6)
686
600
self.assert_sql_count(testing.db, go, count)
602
def test_relationship_to_polymorphic_two(self):
603
expected = self._company_with_emps_machines_fixture()
688
604
sess = create_session()
690
# currently, it doesn't matter if we say Company.employees,
691
# or Company.employees.of_type(Engineer). joinedloader
692
# doesn't pick up on the "of_type()" as of yet.
606
# with #2438, of_type() is recognized. This
607
# overrides the with_polymorphic of the mapper
608
# and we get a consistent 3 queries now.
693
609
eq_(sess.query(Company)
694
610
.options(joinedload_all(
695
611
Company.employees.of_type(Engineer),
696
612
Engineer.machines))
699
# in the case of select_type='', the joinedload
700
# doesn't take in this case; it joinedloads company->people,
701
# then a load for each of 5 rows, then lazyload of "machines"
702
count = {'':7, 'Polymorphic':1}.get(self.select_type, 2)
616
# in the old case, we would get this
617
#count = {'':7, 'Polymorphic':1}.get(self.select_type, 2)
619
# query one is company->Person/Engineer->Machines
620
# query two is managers + boss for row #3
621
# query three is managers for row #4
703
623
self.assert_sql_count(testing.db, go, count)
625
def test_relationship_to_polymorphic_three(self):
626
expected = self._company_with_emps_machines_fixture()
627
sess = create_session()
705
629
sess = create_session()
707
631
eq_(sess.query(Company)
710
634
Engineer.machines))
718
'AliasedJoins':4}[self.select_type]
638
# the old case where subqueryload_all
639
# didn't work with of_tyoe
640
#count = { '':8, 'Joins':4, 'Unions':4, 'Polymorphic':3,
641
# 'AliasedJoins':4}[self.select_type]
643
# query one is company->Person/Engineer->Machines
644
# query two is Person/Engineer subq
645
# query three is Machines subq
646
# (however this test can't tell if the Q was a
647
# lazyload or subqload ...)
648
# query four is managers + boss for row #3
649
# query five is managers for row #4
719
651
self.assert_sql_count(testing.db, go, count)
721
654
def test_joinedload_on_subclass(self):
722
655
sess = create_session()
869
802
.filter(Machine.name.ilike("%ibm%")).all(),
872
def test_join_to_subclass_eightteen(self):
873
sess = create_session()
875
eq_(sess.query(Company)
876
.join(Company.employees.of_type(Engineer))
877
.filter(Engineer.primary_language == 'java').all(),
880
def test_join_to_subclass_nineteen(self):
881
sess = create_session()
882
eq_(sess.query(Company)
883
.join(Company.employees.of_type(Engineer), 'machines')
884
.filter(Machine.name.ilike("%thinkpad%")).all(),
887
def test_join_to_subclass_count(self):
888
sess = create_session()
890
eq_(sess.query(Company, Engineer)
891
.join(Company.employees.of_type(Engineer))
892
.filter(Engineer.primary_language == 'java').count(),
896
eq_(sess.query(Company.company_id, Engineer)
897
.join(Company.employees.of_type(Engineer))
898
.filter(Engineer.primary_language == 'java').count(),
901
eq_(sess.query(Company)
902
.join(Company.employees.of_type(Engineer))
903
.filter(Engineer.primary_language == 'java').count(),
906
805
def test_join_through_polymorphic_nonaliased_one(self):
907
806
sess = create_session()
908
807
eq_(sess.query(Company)
1034
933
.filter(palias.name.in_(['dilbert', 'wally'])).all(),
1037
def test_self_referential(self):
936
def test_self_referential_one(self):
1038
937
sess = create_session()
1039
938
palias = aliased(Person)
1040
939
expected = [(m1, e1), (m1, e2), (m1, b1)]
1046
945
.order_by(Person.person_id, palias.person_id).all(),
948
def test_self_referential_two(self):
949
sess = create_session()
950
palias = aliased(Person)
951
expected = [(m1, e1), (m1, e2), (m1, b1)]
1049
953
eq_(sess.query(Person, palias)
1050
954
.filter(Person.company_id == palias.company_id)
1051
955
.filter(Person.name == 'dogbert')
added
removed
test/orm/inheritance/test_polymorphic_rel.py
