← Back to branch summary

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

« back to all changes in this revision

Viewing changes to test/orm/test_froms.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_froms.py
1
 
from test.lib.testing import eq_, assert_raises, assert_raises_message
 
1
from sqlalchemy.testing import eq_, assert_raises, assert_raises_message
2
2
import operator
3
3
from sqlalchemy import *
4
4
from sqlalchemy import exc as sa_exc, util
7
7
from sqlalchemy.orm import *
8
8
from sqlalchemy.orm import attributes
9
9
 
10
 
from test.lib.testing import eq_
 
10
from sqlalchemy.testing import eq_
11
11
 
12
12
import sqlalchemy as sa
13
 
from test.lib import testing, AssertsCompiledSQL, Column, engines
 
13
from sqlalchemy import testing
 
14
from sqlalchemy.testing import AssertsCompiledSQL, engines
 
15
from sqlalchemy.testing.schema import Column
14
16
 
15
17
from test.orm import _fixtures
16
18
 
17
 
from test.lib import fixtures
 
19
from sqlalchemy.testing import fixtures
18
20
 
19
21
from sqlalchemy.orm.util import join, outerjoin, with_parent
20
22
 
65
67
 
66
68
        configure_mappers()
67
69
 
 
70
class QueryCorrelatesLikeSelect(QueryTest, AssertsCompiledSQL):
 
71
 
 
72
    query_correlated = "SELECT users.name AS users_name, " \
 
73
        "(SELECT count(addresses.id) AS count_1 FROM addresses " \
 
74
        "WHERE addresses.user_id = users.id) AS anon_1 FROM users"
 
75
 
 
76
    query_not_correlated = "SELECT users.name AS users_name, " \
 
77
        "(SELECT count(addresses.id) AS count_1 FROM addresses, users " \
 
78
        "WHERE addresses.user_id = users.id) AS anon_1 FROM users"
 
79
 
 
80
    def test_as_scalar_select_auto_correlate(self):
 
81
        addresses, users = self.tables.addresses, self.tables.users
 
82
        query = select(
 
83
            [func.count(addresses.c.id)],
 
84
            addresses.c.user_id==users.c.id
 
85
        ).as_scalar()
 
86
        query = select([users.c.name.label('users_name'), query])
 
87
        self.assert_compile(query, self.query_correlated,
 
88
            dialect=default.DefaultDialect()
 
89
        )
 
90
 
 
91
    def test_as_scalar_select_explicit_correlate(self):
 
92
        addresses, users = self.tables.addresses, self.tables.users
 
93
        query = select(
 
94
            [func.count(addresses.c.id)],
 
95
            addresses.c.user_id==users.c.id
 
96
        ).correlate(users).as_scalar()
 
97
        query = select([users.c.name.label('users_name'), query])
 
98
        self.assert_compile(query, self.query_correlated,
 
99
            dialect=default.DefaultDialect()
 
100
        )
 
101
 
 
102
    def test_as_scalar_select_correlate_off(self):
 
103
        addresses, users = self.tables.addresses, self.tables.users
 
104
        query = select(
 
105
            [func.count(addresses.c.id)],
 
106
            addresses.c.user_id==users.c.id
 
107
        ).correlate(None).as_scalar()
 
108
        query = select([ users.c.name.label('users_name'), query])
 
109
        self.assert_compile(query, self.query_not_correlated,
 
110
            dialect=default.DefaultDialect()
 
111
        )
 
112
 
 
113
    def test_as_scalar_query_auto_correlate(self):
 
114
        sess = create_session()
 
115
        Address, User = self.classes.Address, self.classes.User
 
116
        query = sess.query(func.count(Address.id))\
 
117
            .filter(Address.user_id==User.id)\
 
118
            .as_scalar()
 
119
        query = sess.query(User.name, query)
 
120
        self.assert_compile(query, self.query_correlated,
 
121
            dialect=default.DefaultDialect()
 
122
        )
 
123
 
 
124
    def test_as_scalar_query_explicit_correlate(self):
 
125
        sess = create_session()
 
126
        Address, User = self.classes.Address, self.classes.User
 
127
        query = sess.query(func.count(Address.id))\
 
128
            .filter(Address.user_id==User.id)\
 
129
            .correlate(self.tables.users)\
 
130
            .as_scalar()
 
131
        query = sess.query(User.name, query)
 
132
        self.assert_compile(query, self.query_correlated,
 
133
            dialect=default.DefaultDialect()
 
134
        )
 
135
 
 
136
    def test_as_scalar_query_correlate_off(self):
 
137
        sess = create_session()
 
138
        Address, User = self.classes.Address, self.classes.User
 
139
        query = sess.query(func.count(Address.id))\
 
140
            .filter(Address.user_id==User.id)\
 
141
            .correlate(None)\
 
142
            .as_scalar()
 
143
        query = sess.query(User.name, query)
 
144
        self.assert_compile(query, self.query_not_correlated,
 
145
            dialect=default.DefaultDialect()
 
146
        )
 
147
 
68
148
 
69
149
class RawSelectTest(QueryTest, AssertsCompiledSQL):
70
 
    """compare a bunch of select() tests with the equivalent Query using straight table/columns.
 
150
    """compare a bunch of select() tests with the equivalent Query using
 
151
    straight table/columns.
71
152
 
72
 
    Results should be the same as Query should act as a select() pass-thru for ClauseElement entities.
 
153
    Results should be the same as Query should act as a select() pass-
 
154
    thru for ClauseElement entities.
73
155
 
74
156
    """
 
157
    __dialect__ = 'default'
 
158
 
75
159
    def test_select(self):
76
160
        addresses, users = self.tables.addresses, self.tables.users
77
161
 
78
162
        sess = create_session()
79
163
 
80
 
        self.assert_compile(sess.query(users).select_from(users.select()).with_labels().statement,
 
164
        self.assert_compile(sess.query(users).select_from(
 
165
                    users.select()).with_labels().statement,
81
166
            "SELECT users.id AS users_id, users.name AS users_name FROM users, "
82
167
            "(SELECT users.id AS id, users.name AS name FROM users) AS anon_1",
83
 
            dialect=default.DefaultDialect()
84
168
            )
85
169
 
86
 
        self.assert_compile(sess.query(users, exists([1], from_obj=addresses)).with_labels().statement,
 
170
        self.assert_compile(sess.query(users, exists([1], from_obj=addresses)
 
171
                ).with_labels().statement,
87
172
            "SELECT users.id AS users_id, users.name AS users_name, EXISTS "
88
173
            "(SELECT 1 FROM addresses) AS anon_1 FROM users",
89
 
            dialect=default.DefaultDialect()
90
174
            )
91
175
 
92
 
        # a little tedious here, adding labels to work around Query's auto-labelling.
93
 
        # also correlate needed explicitly.  hmmm.....
94
 
        # TODO: can we detect only one table in the "froms" and then turn off use_labels ?
95
 
        s = sess.query(addresses.c.id.label('id'), addresses.c.email_address.label('email')).\
96
 
            filter(addresses.c.user_id==users.c.id).correlate(users).statement.alias()
 
176
        # a little tedious here, adding labels to work around Query's
 
177
        # auto-labelling.
 
178
        s = sess.query(addresses.c.id.label('id'),
 
179
                            addresses.c.email_address.label('email')).\
 
180
            filter(addresses.c.user_id == users.c.id).correlate(users).\
 
181
                        statement.alias()
97
182
 
98
 
        self.assert_compile(sess.query(users, s.c.email).select_from(users.join(s, s.c.id==users.c.id)).with_labels().statement,
99
 
                "SELECT users.id AS users_id, users.name AS users_name, anon_1.email AS anon_1_email "
100
 
                "FROM users JOIN (SELECT addresses.id AS id, addresses.email_address AS email FROM addresses "
101
 
                "WHERE addresses.user_id = users.id) AS anon_1 ON anon_1.id = users.id",
102
 
                dialect=default.DefaultDialect()
 
183
        self.assert_compile(sess.query(users, s.c.email).select_from(
 
184
                    users.join(s, s.c.id == users.c.id)
 
185
                ).with_labels().statement,
 
186
                "SELECT users.id AS users_id, users.name AS users_name, "
 
187
                "anon_1.email AS anon_1_email "
 
188
                "FROM users JOIN (SELECT addresses.id AS id, "
 
189
                "addresses.email_address AS email FROM addresses, users "
 
190
                "WHERE addresses.user_id = users.id) AS anon_1 "
 
191
                "ON anon_1.id = users.id",
103
192
            )
104
193
 
105
194
        x = func.lala(users.c.id).label('foo')
106
 
        self.assert_compile(sess.query(x).filter(x==5).statement,
107
 
            "SELECT lala(users.id) AS foo FROM users WHERE lala(users.id) = :param_1", dialect=default.DefaultDialect())
 
195
        self.assert_compile(sess.query(x).filter(x == 5).statement,
 
196
            "SELECT lala(users.id) AS foo FROM users WHERE "
 
197
            "lala(users.id) = :param_1")
108
198
 
109
199
        self.assert_compile(sess.query(func.sum(x).label('bar')).statement,
110
 
            "SELECT sum(lala(users.id)) AS bar FROM users", dialect=default.DefaultDialect())
 
200
            "SELECT sum(lala(users.id)) AS bar FROM users")
111
201
 
112
202
 
113
203
class FromSelfTest(QueryTest, AssertsCompiledSQL):
335
425
            "WHERE anon_1.anon_2_users_name = :name_1"
336
426
        )
337
427
 
 
428
    def test_select_entity_from(self):
 
429
        User = self.classes.User
 
430
        sess = create_session()
 
431
 
 
432
        q = sess.query(User)
 
433
        q = sess.query(User).select_entity_from(q.statement)
 
434
        self.assert_compile(
 
435
            q.filter(User.name=='ed'),
 
436
            "SELECT anon_1.id AS anon_1_id, anon_1.name AS anon_1_name "
 
437
            "FROM (SELECT users.id AS id, users.name AS name FROM "
 
438
            "users) AS anon_1 WHERE anon_1.name = :name_1"
 
439
        )
 
440
 
 
441
    def test_select_entity_from_no_entities(self):
 
442
        User = self.classes.User
 
443
        sess = create_session()
 
444
 
 
445
        q = sess.query(User)
 
446
        assert_raises_message(
 
447
            sa.exc.ArgumentError,
 
448
            r"A selectable \(FromClause\) instance is "
 
449
            "expected when the base alias is being set.",
 
450
            sess.query(User).select_entity_from, User
 
451
        )
 
452
 
 
453
 
338
454
    def test_select_from(self):
339
455
        User = self.classes.User
340
456
        sess = create_session()
1048
1164
    @testing.fails_on('postgresql+zxjdbc',
1049
1165
                      "zxjdbc parses the SQL itself before passing on "
1050
1166
                      "to PG, doesn't parse this")
 
1167
    @testing.fails_on("firebird", "unknown")
1051
1168
    def test_values_with_boolean_selects(self):
1052
1169
        """Tests a values clause that works with select boolean
1053
1170
        evaluations"""
1237
1354
            eq_(results, [(User(name='jack'), 'jack')])
1238
1355
        self.assert_sql_count(testing.db, go, 1)
1239
1356
 
 
1357
    @testing.fails_on("firebird", "unknown")
1240
1358
    @testing.fails_on('postgresql+pg8000', "'type oid 705 not mapped to py type' (due to literal)")
1241
1359
    def test_self_referential(self):
1242
1360
        Order = self.classes.Order
1572
1690
        ]:
1573
1691
            q = s.query(crit)
1574
1692
            mzero = q._mapper_zero()
1575
 
            assert mzero._AliasedClass__alias is q._entity_zero().selectable
 
1693
            assert inspect(mzero).selectable is q._entity_zero().selectable
1576
1694
            q = q.join(j)
1577
1695
            self.assert_compile(q, exp)
1578
1696
 
1721
1839
        )
1722
1840
 
1723
1841
 
 
1842
    def test_aliased_class_vs_nonaliased(self):
 
1843
        User, users = self.classes.User, self.tables.users
 
1844
        mapper(User, users)
 
1845
 
 
1846
        ua = aliased(User)
 
1847
 
 
1848
        sess = create_session()
 
1849
        self.assert_compile(
 
1850
            sess.query(User).select_from(ua).join(User, ua.name > User.name),
 
1851
            "SELECT users.id AS users_id, users.name AS users_name "
 
1852
            "FROM users AS users_1 JOIN users ON users.name < users_1.name"
 
1853
        )
 
1854
 
 
1855
        self.assert_compile(
 
1856
            sess.query(User.name).select_from(ua).join(User, ua.name > User.name),
 
1857
            "SELECT users.name AS users_name FROM users AS users_1 "
 
1858
            "JOIN users ON users.name < users_1.name"
 
1859
        )
 
1860
 
 
1861
        self.assert_compile(
 
1862
            sess.query(ua.name).select_from(ua).join(User, ua.name > User.name),
 
1863
            "SELECT users_1.name AS users_1_name FROM users AS users_1 "
 
1864
            "JOIN users ON users.name < users_1.name"
 
1865
        )
 
1866
 
 
1867
        self.assert_compile(
 
1868
            sess.query(ua).select_from(User).join(ua, ua.name > User.name),
 
1869
            "SELECT users_1.id AS users_1_id, users_1.name AS users_1_name "
 
1870
            "FROM users JOIN users AS users_1 ON users.name < users_1.name"
 
1871
        )
 
1872
 
 
1873
        # this is tested in many other places here, just adding it
 
1874
        # here for comparison
 
1875
        self.assert_compile(
 
1876
            sess.query(User.name).\
 
1877
                    select_from(users.select().where(users.c.id > 5)),
 
1878
            "SELECT anon_1.name AS anon_1_name FROM (SELECT users.id AS id, "
 
1879
            "users.name AS name FROM users WHERE users.id > :id_1) AS anon_1"
 
1880
        )
1724
1881
 
1725
1882
    def test_join_no_order_by(self):
1726
1883
        User, users = self.classes.User, self.tables.users
2189
2346
                filter(Sub1.id==1).one(),
2190
2347
                b1
2191
2348
        )
 
2349
 
 
2350
class LabelCollideTest(fixtures.MappedTest):
 
2351
    """Test handling for a label collision.  This collision
 
2352
    is handled by core, see ticket:2702 as well as
 
2353
    test/sql/test_selectable->WithLabelsTest.  here we want
 
2354
    to make sure the end result is as we expect.
 
2355
 
 
2356
    """
 
2357
 
 
2358
    @classmethod
 
2359
    def define_tables(cls, metadata):
 
2360
        Table('foo', metadata,
 
2361
                Column('id', Integer, primary_key=True),
 
2362
                Column('bar_id', Integer)
 
2363
        )
 
2364
        Table('foo_bar', metadata,
 
2365
                Column('id', Integer, primary_key=True),
 
2366
                )
 
2367
 
 
2368
    @classmethod
 
2369
    def setup_classes(cls):
 
2370
        class Foo(cls.Basic):
 
2371
            pass
 
2372
        class Bar(cls.Basic):
 
2373
            pass
 
2374
 
 
2375
    @classmethod
 
2376
    def setup_mappers(cls):
 
2377
        mapper(cls.classes.Foo, cls.tables.foo)
 
2378
        mapper(cls.classes.Bar, cls.tables.foo_bar)
 
2379
 
 
2380
    @classmethod
 
2381
    def insert_data(cls):
 
2382
        s = Session()
 
2383
        s.add_all([
 
2384
            cls.classes.Foo(id=1, bar_id=2),
 
2385
            cls.classes.Bar(id=3)
 
2386
        ])
 
2387
        s.commit()
 
2388
 
 
2389
    def test_overlap_plain(self):
 
2390
        s = Session()
 
2391
        row = s.query(self.classes.Foo, self.classes.Bar).all()[0]
 
2392
        def go():
 
2393
            eq_(row.Foo.id, 1)
 
2394
            eq_(row.Foo.bar_id, 2)
 
2395
            eq_(row.Bar.id, 3)
 
2396
        # all three columns are loaded independently without
 
2397
        # overlap, no additional SQL to load all attributes
 
2398
        self.assert_sql_count(testing.db, go, 0)
 
2399
 
 
2400
    def test_overlap_subquery(self):
 
2401
        s = Session()
 
2402
        row = s.query(self.classes.Foo, self.classes.Bar).from_self().all()[0]
 
2403
        def go():
 
2404
            eq_(row.Foo.id, 1)
 
2405
            eq_(row.Foo.bar_id, 2)
 
2406
            eq_(row.Bar.id, 3)
 
2407
        # all three columns are loaded independently without
 
2408
        # overlap, no additional SQL to load all attributes
 
2409
        self.assert_sql_count(testing.db, go, 0)
 
 
b'\\ No newline at end of file'