← Back to branch summary

~ubuntu-branches/debian/jessie/sqlalchemy/jessie

« back to all changes in this revision

Viewing changes to test/orm/test_of_type.py

  • Committer: Package Import Robot
  • Author(s): Piotr Ożarowski, Jakub Wilk, Piotr Ożarowski
  • Date: 2013-07-06 20:53:52 UTC
  • mfrom: (1.4.23) (16.1.17 experimental)
  • Revision ID: package-import@ubuntu.com-20130706205352-ryppl1eto3illd79
Tags: 0.8.2-1
[ Jakub Wilk ]
* Use canonical URIs for Vcs-* fields.

[ Piotr Ożarowski ]
* New upstream release
* Upload to unstable
* Build depend on python3-all instead of -dev, extensions are not built for
  Python 3.X 

Show diffs side-by-side

added added

removed removed

Lines of Context:
test/orm/test_of_type.py
 
1
from sqlalchemy.orm import Session, aliased, with_polymorphic, \
 
2
    contains_eager, joinedload, subqueryload, relationship,\
 
3
    subqueryload_all, joinedload_all
 
4
from sqlalchemy import and_
 
5
from sqlalchemy import testing
 
6
from sqlalchemy.testing import fixtures
 
7
from sqlalchemy.testing import assert_raises, eq_
 
8
from sqlalchemy.testing.schema import Column
 
9
from sqlalchemy.engine import default
 
10
from sqlalchemy.testing.entities import ComparableEntity
 
11
from sqlalchemy import Integer, String, ForeignKey
 
12
from .inheritance._poly_fixtures import Company, Person, Engineer, Manager, Boss, \
 
13
    Machine, Paperwork, _PolymorphicFixtureBase, _Polymorphic,\
 
14
    _PolymorphicPolymorphic, _PolymorphicUnions, _PolymorphicJoins,\
 
15
    _PolymorphicAliasedJoins
 
16
 
 
17
class _PolymorphicTestBase(object):
 
18
    __dialect__ = 'default'
 
19
 
 
20
    def test_any_one(self):
 
21
        sess = Session()
 
22
        any_ = Company.employees.of_type(Engineer).any(
 
23
            Engineer.primary_language == 'cobol')
 
24
        eq_(sess.query(Company).filter(any_).one(), self.c2)
 
25
 
 
26
    def test_any_two(self):
 
27
        sess = Session()
 
28
        calias = aliased(Company)
 
29
        any_ = calias.employees.of_type(Engineer).any(
 
30
            Engineer.primary_language == 'cobol')
 
31
        eq_(sess.query(calias).filter(any_).one(), self.c2)
 
32
 
 
33
    def test_any_three(self):
 
34
        sess = Session()
 
35
        any_ = Company.employees.of_type(Boss).any(
 
36
            Boss.golf_swing == 'fore')
 
37
        eq_(sess.query(Company).filter(any_).one(), self.c1)
 
38
 
 
39
    def test_any_four(self):
 
40
        sess = Session()
 
41
        any_ = Company.employees.of_type(Boss).any(
 
42
            Manager.manager_name == 'pointy')
 
43
        eq_(sess.query(Company).filter(any_).one(), self.c1)
 
44
 
 
45
    def test_any_five(self):
 
46
        sess = Session()
 
47
        any_ = Company.employees.of_type(Engineer).any(
 
48
            and_(Engineer.primary_language == 'cobol'))
 
49
        eq_(sess.query(Company).filter(any_).one(), self.c2)
 
50
 
 
51
    def test_join_to_subclass_one(self):
 
52
        sess = Session()
 
53
        eq_(sess.query(Company)
 
54
                .join(Company.employees.of_type(Engineer))
 
55
                .filter(Engineer.primary_language == 'java').all(),
 
56
            [self.c1])
 
57
 
 
58
    def test_join_to_subclass_two(self):
 
59
        sess = Session()
 
60
        eq_(sess.query(Company)
 
61
                .join(Company.employees.of_type(Engineer), 'machines')
 
62
                .filter(Machine.name.ilike("%thinkpad%")).all(),
 
63
            [self.c1])
 
64
 
 
65
    def test_join_to_subclass_three(self):
 
66
        sess = Session()
 
67
        eq_(sess.query(Company, Engineer)
 
68
                .join(Company.employees.of_type(Engineer))
 
69
                .filter(Engineer.primary_language == 'java').count(),
 
70
            1)
 
71
 
 
72
        # test [ticket:2093]
 
73
        eq_(sess.query(Company.company_id, Engineer)
 
74
                .join(Company.employees.of_type(Engineer))
 
75
                .filter(Engineer.primary_language == 'java').count(),
 
76
            1)
 
77
 
 
78
        eq_(sess.query(Company)
 
79
                .join(Company.employees.of_type(Engineer))
 
80
                .filter(Engineer.primary_language == 'java').count(),
 
81
            1)
 
82
 
 
83
    def test_with_polymorphic_join_compile_one(self):
 
84
        sess = Session()
 
85
 
 
86
        self.assert_compile(
 
87
            sess.query(Company).join(
 
88
                    Company.employees.of_type(
 
89
                        with_polymorphic(Person, [Engineer, Manager], aliased=True)
 
90
                    )
 
91
                ),
 
92
            "SELECT companies.company_id AS companies_company_id, "
 
93
            "companies.name AS companies_name FROM companies "
 
94
            "JOIN %s"
 
95
             % (
 
96
                self._polymorphic_join_target([Engineer, Manager])
 
97
            )
 
98
        )
 
99
 
 
100
    def test_with_polymorphic_join_exec_contains_eager_one(self):
 
101
        sess = Session()
 
102
        def go():
 
103
            wp = with_polymorphic(Person, [Engineer, Manager], aliased=True)
 
104
            eq_(
 
105
                sess.query(Company).join(
 
106
                    Company.employees.of_type(wp)
 
107
                ).order_by(Company.company_id, wp.person_id).\
 
108
                options(contains_eager(Company.employees.of_type(wp))).all(),
 
109
                [self.c1, self.c2]
 
110
            )
 
111
        self.assert_sql_count(testing.db, go, 1)
 
112
 
 
113
    def test_with_polymorphic_join_exec_contains_eager_two(self):
 
114
        sess = Session()
 
115
        def go():
 
116
            wp = with_polymorphic(Person, [Engineer, Manager], aliased=True)
 
117
            eq_(
 
118
                sess.query(Company).join(
 
119
                    Company.employees.of_type(wp)
 
120
                ).order_by(Company.company_id, wp.person_id).\
 
121
                options(contains_eager(Company.employees, alias=wp)).all(),
 
122
                [self.c1, self.c2]
 
123
            )
 
124
        self.assert_sql_count(testing.db, go, 1)
 
125
 
 
126
    def test_with_polymorphic_any(self):
 
127
        sess = Session()
 
128
        wp = with_polymorphic(Person, [Engineer], aliased=True)
 
129
        eq_(
 
130
            sess.query(Company.company_id).\
 
131
                filter(
 
132
                    Company.employees.of_type(wp).any(
 
133
                            wp.Engineer.primary_language == 'java')
 
134
                ).all(),
 
135
            [(1, )]
 
136
        )
 
137
 
 
138
    def test_subqueryload_implicit_withpoly(self):
 
139
        sess = Session()
 
140
        def go():
 
141
            eq_(
 
142
                sess.query(Company).\
 
143
                    filter_by(company_id=1).\
 
144
                    options(subqueryload(Company.employees.of_type(Engineer))).\
 
145
                    all(),
 
146
                [self._company_with_emps_fixture()[0]]
 
147
            )
 
148
        self.assert_sql_count(testing.db, go, 4)
 
149
 
 
150
    def test_joinedload_implicit_withpoly(self):
 
151
        sess = Session()
 
152
        def go():
 
153
            eq_(
 
154
                sess.query(Company).\
 
155
                    filter_by(company_id=1).\
 
156
                    options(joinedload(Company.employees.of_type(Engineer))).\
 
157
                    all(),
 
158
                [self._company_with_emps_fixture()[0]]
 
159
            )
 
160
        self.assert_sql_count(testing.db, go, 3)
 
161
 
 
162
    def test_subqueryload_explicit_withpoly(self):
 
163
        sess = Session()
 
164
        def go():
 
165
            target = with_polymorphic(Person, Engineer, aliased=True)
 
166
            eq_(
 
167
                sess.query(Company).\
 
168
                    filter_by(company_id=1).\
 
169
                    options(subqueryload(Company.employees.of_type(target))).\
 
170
                    all(),
 
171
                [self._company_with_emps_fixture()[0]]
 
172
            )
 
173
        self.assert_sql_count(testing.db, go, 4)
 
174
 
 
175
    def test_joinedload_explicit_withpoly(self):
 
176
        sess = Session()
 
177
        def go():
 
178
            target = with_polymorphic(Person, Engineer, aliased=True)
 
179
            eq_(
 
180
                sess.query(Company).\
 
181
                    filter_by(company_id=1).\
 
182
                    options(joinedload(Company.employees.of_type(target))).\
 
183
                    all(),
 
184
                [self._company_with_emps_fixture()[0]]
 
185
            )
 
186
        self.assert_sql_count(testing.db, go, 3)
 
187
 
 
188
class PolymorphicPolymorphicTest(_PolymorphicTestBase, _PolymorphicPolymorphic):
 
189
    def _polymorphic_join_target(self, cls):
 
190
        from sqlalchemy.orm import class_mapper
 
191
 
 
192
        m, sel = class_mapper(Person)._with_polymorphic_args(cls)
 
193
        sel = sel.alias()
 
194
        comp_sel = sel.compile(dialect=default.DefaultDialect())
 
195
 
 
196
        return \
 
197
            comp_sel.process(sel, asfrom=True).replace("\n", "") + \
 
198
            " ON companies.company_id = anon_1.people_company_id"
 
199
 
 
200
class PolymorphicUnionsTest(_PolymorphicTestBase, _PolymorphicUnions):
 
201
 
 
202
    def _polymorphic_join_target(self, cls):
 
203
        from sqlalchemy.orm import class_mapper
 
204
 
 
205
        sel = class_mapper(Person)._with_polymorphic_selectable.element
 
206
        comp_sel = sel.compile(dialect=default.DefaultDialect())
 
207
 
 
208
        return \
 
209
            comp_sel.process(sel, asfrom=True).replace("\n", "") + \
 
210
            " AS anon_1 ON companies.company_id = anon_1.company_id"
 
211
 
 
212
class PolymorphicAliasedJoinsTest(_PolymorphicTestBase, _PolymorphicAliasedJoins):
 
213
    def _polymorphic_join_target(self, cls):
 
214
        from sqlalchemy.orm import class_mapper
 
215
 
 
216
        sel = class_mapper(Person)._with_polymorphic_selectable.element
 
217
        comp_sel = sel.compile(dialect=default.DefaultDialect())
 
218
 
 
219
        return \
 
220
            comp_sel.process(sel, asfrom=True).replace("\n", "") + \
 
221
            " AS anon_1 ON companies.company_id = anon_1.people_company_id"
 
222
 
 
223
class PolymorphicJoinsTest(_PolymorphicTestBase, _PolymorphicJoins):
 
224
    def _polymorphic_join_target(self, cls):
 
225
        from sqlalchemy.orm import class_mapper
 
226
 
 
227
        sel = class_mapper(Person)._with_polymorphic_selectable.alias()
 
228
        comp_sel = sel.compile(dialect=default.DefaultDialect())
 
229
 
 
230
        return \
 
231
            comp_sel.process(sel, asfrom=True).replace("\n", "") + \
 
232
            " ON companies.company_id = anon_1.people_company_id"
 
233
 
 
234
 
 
235
class SubclassRelationshipTest(testing.AssertsCompiledSQL, fixtures.DeclarativeMappedTest):
 
236
    """There's overlap here vs. the ones above."""
 
237
 
 
238
    run_setup_classes = 'once'
 
239
    run_setup_mappers = 'once'
 
240
    run_inserts = 'once'
 
241
    run_deletes = None
 
242
    __dialect__ = 'default'
 
243
 
 
244
    @classmethod
 
245
    def setup_classes(cls):
 
246
        Base = cls.DeclarativeBasic
 
247
        class Job(ComparableEntity, Base):
 
248
            __tablename__ = "job"
 
249
 
 
250
            id = Column(Integer, primary_key=True,
 
251
                                        test_needs_autoincrement=True)
 
252
            type = Column(String(10))
 
253
            container_id = Column(Integer, ForeignKey('data_container.id'))
 
254
            __mapper_args__ = {"polymorphic_on": type}
 
255
 
 
256
        class SubJob(Job):
 
257
            __tablename__ = 'subjob'
 
258
            id = Column(Integer, ForeignKey('job.id'), primary_key=True)
 
259
            attr = Column(String(10))
 
260
            __mapper_args__ = {"polymorphic_identity": "sub"}
 
261
 
 
262
        class ParentThing(ComparableEntity, Base):
 
263
            __tablename__ = 'parent'
 
264
            id = Column(Integer, primary_key=True,
 
265
                                            test_needs_autoincrement=True)
 
266
            container_id = Column(Integer, ForeignKey('data_container.id'))
 
267
            container = relationship("DataContainer")
 
268
 
 
269
        class DataContainer(ComparableEntity, Base):
 
270
            __tablename__ = "data_container"
 
271
 
 
272
            id = Column(Integer, primary_key=True,
 
273
                                            test_needs_autoincrement=True)
 
274
            name = Column(String(10))
 
275
            jobs = relationship(Job, order_by=Job.id)
 
276
 
 
277
    @classmethod
 
278
    def insert_data(cls):
 
279
        s = Session(testing.db)
 
280
 
 
281
        s.add_all(cls._fixture())
 
282
        s.commit()
 
283
 
 
284
    @classmethod
 
285
    def _fixture(cls):
 
286
        ParentThing, DataContainer, SubJob = \
 
287
            cls.classes.ParentThing,\
 
288
            cls.classes.DataContainer,\
 
289
            cls.classes.SubJob
 
290
        return [
 
291
            ParentThing(
 
292
                container=DataContainer(name="d1",
 
293
                    jobs=[
 
294
                        SubJob(attr="s1"),
 
295
                        SubJob(attr="s2")
 
296
                    ])
 
297
            ),
 
298
            ParentThing(
 
299
                container=DataContainer(name="d2",
 
300
                    jobs=[
 
301
                        SubJob(attr="s3"),
 
302
                        SubJob(attr="s4")
 
303
                    ])
 
304
            ),
 
305
        ]
 
306
 
 
307
    @classmethod
 
308
    def _dc_fixture(cls):
 
309
        return [p.container for p in cls._fixture()]
 
310
 
 
311
    def test_contains_eager_wpoly(self):
 
312
        ParentThing, DataContainer, Job, SubJob = \
 
313
            self.classes.ParentThing,\
 
314
            self.classes.DataContainer,\
 
315
            self.classes.Job,\
 
316
            self.classes.SubJob
 
317
 
 
318
        Job_P = with_polymorphic(Job, SubJob, aliased=True)
 
319
 
 
320
        s = Session(testing.db)
 
321
        q = s.query(DataContainer).\
 
322
                    join(DataContainer.jobs.of_type(Job_P)).\
 
323
                        options(contains_eager(DataContainer.jobs.of_type(Job_P)))
 
324
        def go():
 
325
            eq_(
 
326
                q.all(),
 
327
                self._dc_fixture()
 
328
            )
 
329
        self.assert_sql_count(testing.db, go, 1)
 
330
 
 
331
    def test_joinedload_wpoly(self):
 
332
        ParentThing, DataContainer, Job, SubJob = \
 
333
            self.classes.ParentThing,\
 
334
            self.classes.DataContainer,\
 
335
            self.classes.Job,\
 
336
            self.classes.SubJob
 
337
 
 
338
        Job_P = with_polymorphic(Job, SubJob, aliased=True)
 
339
 
 
340
        s = Session(testing.db)
 
341
        q = s.query(DataContainer).\
 
342
                        options(joinedload(DataContainer.jobs.of_type(Job_P)))
 
343
        def go():
 
344
            eq_(
 
345
                q.all(),
 
346
                self._dc_fixture()
 
347
            )
 
348
        self.assert_sql_count(testing.db, go, 1)
 
349
 
 
350
    def test_joinedload_wsubclass(self):
 
351
        ParentThing, DataContainer, Job, SubJob = \
 
352
            self.classes.ParentThing,\
 
353
            self.classes.DataContainer,\
 
354
            self.classes.Job,\
 
355
            self.classes.SubJob
 
356
        s = Session(testing.db)
 
357
        q = s.query(DataContainer).\
 
358
                        options(joinedload(DataContainer.jobs.of_type(SubJob)))
 
359
        def go():
 
360
            eq_(
 
361
                q.all(),
 
362
                self._dc_fixture()
 
363
            )
 
364
        self.assert_sql_count(testing.db, go, 1)
 
365
 
 
366
    def test_lazyload(self):
 
367
        DataContainer = self.classes.DataContainer
 
368
        s = Session(testing.db)
 
369
        q = s.query(DataContainer)
 
370
        def go():
 
371
            eq_(
 
372
                q.all(),
 
373
                self._dc_fixture()
 
374
            )
 
375
        # SELECT data container
 
376
        # SELECT job * 2 container rows
 
377
        # SELECT subjob * 4 rows
 
378
        self.assert_sql_count(testing.db, go, 7)
 
379
 
 
380
    def test_subquery_wsubclass(self):
 
381
        ParentThing, DataContainer, Job, SubJob = \
 
382
            self.classes.ParentThing,\
 
383
            self.classes.DataContainer,\
 
384
            self.classes.Job,\
 
385
            self.classes.SubJob
 
386
        s = Session(testing.db)
 
387
        q = s.query(DataContainer).\
 
388
                        options(subqueryload(DataContainer.jobs.of_type(SubJob)))
 
389
        def go():
 
390
            eq_(
 
391
                q.all(),
 
392
                self._dc_fixture()
 
393
            )
 
394
        self.assert_sql_count(testing.db, go, 2)
 
395
 
 
396
    def test_twolevel_subqueryload_wsubclass(self):
 
397
        ParentThing, DataContainer, Job, SubJob = \
 
398
            self.classes.ParentThing,\
 
399
            self.classes.DataContainer,\
 
400
            self.classes.Job,\
 
401
            self.classes.SubJob
 
402
        s = Session(testing.db)
 
403
        q = s.query(ParentThing).\
 
404
                        options(
 
405
                            subqueryload_all(
 
406
                                ParentThing.container,
 
407
                                DataContainer.jobs.of_type(SubJob)
 
408
                        ))
 
409
        def go():
 
410
            eq_(
 
411
                q.all(),
 
412
                self._fixture()
 
413
            )
 
414
        self.assert_sql_count(testing.db, go, 3)
 
415
 
 
416
    def test_twolevel_joinedload_wsubclass(self):
 
417
        ParentThing, DataContainer, Job, SubJob = \
 
418
            self.classes.ParentThing,\
 
419
            self.classes.DataContainer,\
 
420
            self.classes.Job,\
 
421
            self.classes.SubJob
 
422
        s = Session(testing.db)
 
423
        q = s.query(ParentThing).\
 
424
                        options(
 
425
                            joinedload_all(
 
426
                                ParentThing.container,
 
427
                                DataContainer.jobs.of_type(SubJob)
 
428
                        ))
 
429
        def go():
 
430
            eq_(
 
431
                q.all(),
 
432
                self._fixture()
 
433
            )
 
434
        self.assert_sql_count(testing.db, go, 1)
 
435
 
 
436
    def test_any_wpoly(self):
 
437
        ParentThing, DataContainer, Job, SubJob = \
 
438
            self.classes.ParentThing,\
 
439
            self.classes.DataContainer,\
 
440
            self.classes.Job,\
 
441
            self.classes.SubJob
 
442
 
 
443
        Job_P = with_polymorphic(Job, SubJob, aliased=True)
 
444
 
 
445
        s = Session()
 
446
        q = s.query(Job).join(DataContainer.jobs).\
 
447
                        filter(
 
448
                            DataContainer.jobs.of_type(Job_P).\
 
449
                                any(Job_P.id < Job.id)
 
450
                        )
 
451
        self.assert_compile(q,
 
452
            "SELECT job.id AS job_id, job.type AS job_type, "
 
453
            "job.container_id "
 
454
            "AS job_container_id "
 
455
            "FROM data_container "
 
456
            "JOIN job ON data_container.id = job.container_id "
 
457
            "WHERE EXISTS (SELECT 1 "
 
458
            "FROM (SELECT job.id AS job_id, job.type AS job_type, "
 
459
            "job.container_id AS job_container_id, "
 
460
            "subjob.id AS subjob_id, subjob.attr AS subjob_attr "
 
461
            "FROM job LEFT OUTER JOIN subjob ON job.id = subjob.id) AS anon_1 "
 
462
            "WHERE data_container.id = anon_1.job_container_id AND job.id > anon_1.job_id)"
 
463
        )
 
464
 
 
465
    def test_any_walias(self):
 
466
        ParentThing, DataContainer, Job, SubJob = \
 
467
            self.classes.ParentThing,\
 
468
            self.classes.DataContainer,\
 
469
            self.classes.Job,\
 
470
            self.classes.SubJob
 
471
 
 
472
        Job_A = aliased(Job)
 
473
 
 
474
        s = Session()
 
475
        q = s.query(Job).join(DataContainer.jobs).\
 
476
                        filter(
 
477
                            DataContainer.jobs.of_type(Job_A).\
 
478
                                any(and_(Job_A.id < Job.id, Job_A.type=='fred'))
 
479
                        )
 
480
        self.assert_compile(q,
 
481
            "SELECT job.id AS job_id, job.type AS job_type, "
 
482
            "job.container_id AS job_container_id "
 
483
            "FROM data_container JOIN job ON data_container.id = job.container_id "
 
484
            "WHERE EXISTS (SELECT 1 "
 
485
            "FROM job AS job_1 "
 
486
            "WHERE data_container.id = job_1.container_id "
 
487
            "AND job.id > job_1.id AND job_1.type = :type_1)"
 
488
        )
 
489
 
 
490
    def test_join_wpoly(self):
 
491
        ParentThing, DataContainer, Job, SubJob = \
 
492
            self.classes.ParentThing,\
 
493
            self.classes.DataContainer,\
 
494
            self.classes.Job,\
 
495
            self.classes.SubJob
 
496
 
 
497
        Job_P = with_polymorphic(Job, SubJob, aliased=True)
 
498
 
 
499
        s = Session()
 
500
        q = s.query(DataContainer).join(DataContainer.jobs.of_type(Job_P))
 
501
        self.assert_compile(q,
 
502
            "SELECT data_container.id AS data_container_id, "
 
503
            "data_container.name AS data_container_name "
 
504
            "FROM data_container JOIN (SELECT job.id AS job_id, "
 
505
            "job.type AS job_type, job.container_id AS job_container_id, "
 
506
            "subjob.id AS subjob_id, subjob.attr AS subjob_attr "
 
507
            "FROM job LEFT OUTER JOIN subjob ON job.id = subjob.id) "
 
508
            "AS anon_1 ON data_container.id = anon_1.job_container_id")
 
509
 
 
510
    def test_join_wsubclass(self):
 
511
        ParentThing, DataContainer, Job, SubJob = \
 
512
            self.classes.ParentThing,\
 
513
            self.classes.DataContainer,\
 
514
            self.classes.Job,\
 
515
            self.classes.SubJob
 
516
 
 
517
        s = Session()
 
518
        q = s.query(DataContainer).join(DataContainer.jobs.of_type(SubJob))
 
519
        # note the of_type() here renders JOIN for the Job->SubJob.
 
520
        # this is because it's using the SubJob mapper directly within
 
521
        # query.join().  When we do joinedload() etc., we're instead
 
522
        # doing a with_polymorphic(), and there we need the join to be
 
523
        # outer by default.
 
524
        self.assert_compile(q,
 
525
            "SELECT data_container.id AS data_container_id, "
 
526
            "data_container.name AS data_container_name "
 
527
            "FROM data_container JOIN (SELECT job.id AS job_id, "
 
528
            "job.type AS job_type, job.container_id AS job_container_id, "
 
529
            "subjob.id AS subjob_id, subjob.attr AS subjob_attr "
 
530
            "FROM job JOIN subjob ON job.id = subjob.id) AS anon_1 "
 
531
            "ON data_container.id = anon_1.job_container_id"
 
532
        )
 
533
 
 
534
    def test_join_wpoly_innerjoin(self):
 
535
        ParentThing, DataContainer, Job, SubJob = \
 
536
            self.classes.ParentThing,\
 
537
            self.classes.DataContainer,\
 
538
            self.classes.Job,\
 
539
            self.classes.SubJob
 
540
 
 
541
        Job_P = with_polymorphic(Job, SubJob, aliased=True, innerjoin=True)
 
542
 
 
543
        s = Session()
 
544
        q = s.query(DataContainer).join(DataContainer.jobs.of_type(Job_P))
 
545
        self.assert_compile(q,
 
546
            "SELECT data_container.id AS data_container_id, "
 
547
            "data_container.name AS data_container_name "
 
548
            "FROM data_container JOIN (SELECT job.id AS job_id, "
 
549
            "job.type AS job_type, job.container_id AS job_container_id, "
 
550
            "subjob.id AS subjob_id, subjob.attr AS subjob_attr "
 
551
            "FROM job JOIN subjob ON job.id = subjob.id) "
 
552
            "AS anon_1 ON data_container.id = anon_1.job_container_id")
 
553
 
 
554
    def test_join_walias(self):
 
555
        ParentThing, DataContainer, Job, SubJob = \
 
556
            self.classes.ParentThing,\
 
557
            self.classes.DataContainer,\
 
558
            self.classes.Job,\
 
559
            self.classes.SubJob
 
560
 
 
561
        Job_A = aliased(Job)
 
562
 
 
563
        s = Session()
 
564
        q = s.query(DataContainer).join(DataContainer.jobs.of_type(Job_A))
 
565
        self.assert_compile(q,
 
566
            "SELECT data_container.id AS data_container_id, "
 
567
            "data_container.name AS data_container_name "
 
568
            "FROM data_container JOIN job AS job_1 "
 
569
            "ON data_container.id = job_1.container_id")
 
570
 
 
571
    def test_join_explicit_wpoly(self):
 
572
        ParentThing, DataContainer, Job, SubJob = \
 
573
            self.classes.ParentThing,\
 
574
            self.classes.DataContainer,\
 
575
            self.classes.Job,\
 
576
            self.classes.SubJob
 
577
 
 
578
        Job_P = with_polymorphic(Job, SubJob, aliased=True)
 
579
 
 
580
        s = Session()
 
581
        q = s.query(DataContainer).join(Job_P, DataContainer.jobs)
 
582
        self.assert_compile(q,
 
583
            "SELECT data_container.id AS data_container_id, "
 
584
            "data_container.name AS data_container_name "
 
585
            "FROM data_container JOIN (SELECT job.id AS job_id, "
 
586
            "job.type AS job_type, job.container_id AS job_container_id, "
 
587
            "subjob.id AS subjob_id, subjob.attr AS subjob_attr "
 
588
            "FROM job LEFT OUTER JOIN subjob ON job.id = subjob.id) "
 
589
            "AS anon_1 ON data_container.id = anon_1.job_container_id")
 
590