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,\
5
class_mapper, with_polymorphic
6
from sqlalchemy import exc as sa_exc
7
from sqlalchemy.engine import default
9
from sqlalchemy.testing import AssertsCompiledSQL, fixtures
10
from sqlalchemy import testing
11
from sqlalchemy.testing.schema import Table, Column
12
from sqlalchemy.testing import assert_raises, eq_
14
from _poly_fixtures import Company, Person, Engineer, Manager, Boss, \
15
Machine, Paperwork, _PolymorphicFixtureBase, _Polymorphic,\
16
_PolymorphicPolymorphic, _PolymorphicUnions, _PolymorphicJoins,\
17
_PolymorphicAliasedJoins
19
class _WithPolymorphicBase(_PolymorphicFixtureBase):
20
def test_join_base_to_sub(self):
21
sess = create_session()
22
pa = with_polymorphic(Person, [Engineer])
26
.filter(pa.Engineer.primary_language == 'java').all(),
27
self._emps_wo_relationships_fixture()[0:1])
28
self.assert_sql_count(testing.db, go, 1)
30
def test_col_expression_base_plus_two_subs(self):
31
sess = create_session()
32
pa = with_polymorphic(Person, [Engineer, Manager])
35
sess.query(pa.name, pa.Engineer.primary_language, pa.Manager.manager_name).\
36
filter(or_(pa.Engineer.primary_language=='java',
37
pa.Manager.manager_name=='dogbert')).\
38
order_by(pa.Engineer.type).all(),
40
(u'dilbert', u'java', None),
41
(u'dogbert', None, u'dogbert'),
46
def test_join_to_join_entities(self):
47
sess = create_session()
48
pa = with_polymorphic(Person, [Engineer])
49
pa_alias = with_polymorphic(Person, [Engineer], aliased=True)
52
[(p1.name, type(p1), p2.name, type(p2)) for (p1, p2) in sess.query(
56
pa.Engineer.primary_language==\
57
pa_alias.Engineer.primary_language,
59
pa.Engineer.primary_language == None,
60
pa_alias.Engineer.primary_language == None,
61
pa.person_id > pa_alias.person_id
64
).order_by(pa.name, pa_alias.name)],
66
(u'dilbert', Engineer, u'dilbert', Engineer),
67
(u'dogbert', Manager, u'pointy haired boss', Boss),
68
(u'vlad', Engineer, u'vlad', Engineer),
69
(u'wally', Engineer, u'wally', Engineer)
73
def test_join_to_join_columns(self):
74
sess = create_session()
75
pa = with_polymorphic(Person, [Engineer])
76
pa_alias = with_polymorphic(Person, [Engineer], aliased=True)
79
[row for row in sess.query(
80
pa.name, pa.Engineer.primary_language,
81
pa_alias.name, pa_alias.Engineer.primary_language
84
pa.Engineer.primary_language==\
85
pa_alias.Engineer.primary_language,
87
pa.Engineer.primary_language == None,
88
pa_alias.Engineer.primary_language == None,
89
pa.person_id > pa_alias.person_id
92
).order_by(pa.name, pa_alias.name)],
94
(u'dilbert', u'java', u'dilbert', u'java'),
95
(u'dogbert', None, u'pointy haired boss', None),
96
(u'vlad', u'cobol', u'vlad', u'cobol'),
97
(u'wally', u'c++', u'wally', u'c++')
101
class PolymorphicTest(_WithPolymorphicBase, _Polymorphic):
104
class PolymorphicPolymorphicTest(_WithPolymorphicBase, _PolymorphicPolymorphic):
107
class PolymorphicUnionsTest(_WithPolymorphicBase, _PolymorphicUnions):
110
class PolymorphicAliasedJoinsTest(_WithPolymorphicBase, _PolymorphicAliasedJoins):
113
class PolymorphicJoinsTest(_WithPolymorphicBase, _PolymorphicJoins):