:version: 0.8.0b1
:released:
+ .. change::
+ :tags: sql
+ :tickets: 2595
+
+ The auto-correlation feature of :func:`.select`, and
+ by proxy that of :class:`.orm.Query`, will not
+ take effect for a SELECT statement that is being
+ rendered directly in the FROM list of the enclosing
+ SELECT. Correlation in SQL only applies to column
+ expressions such as those in the WHERE, ORDER BY,
+ columns clause.
+
.. change::
:tags: sqlite
:pullreq: 23
positional_names=None, **kwargs):
entry = self.stack and self.stack[-1] or {}
- existingfroms = entry.get('from', None)
+ if not asfrom:
+ existingfroms = entry.get('from', None)
+ else:
+ # don't render correlations if we're rendering a FROM list
+ # entry
+ existingfroms = []
froms = select._get_display_froms(existingfroms)
class RawSelectTest(QueryTest, AssertsCompiledSQL):
- """compare a bunch of select() tests with the equivalent Query using straight table/columns.
+ """compare a bunch of select() tests with the equivalent Query using
+ straight table/columns.
- Results should be the same as Query should act as a select() pass-thru for ClauseElement entities.
+ Results should be the same as Query should act as a select() pass-
+ thru for ClauseElement entities.
"""
+ __dialect__ = 'default'
+
def test_select(self):
addresses, users = self.tables.addresses, self.tables.users
sess = create_session()
- self.assert_compile(sess.query(users).select_from(users.select()).with_labels().statement,
+ self.assert_compile(sess.query(users).select_from(
+ users.select()).with_labels().statement,
"SELECT users.id AS users_id, users.name AS users_name FROM users, "
"(SELECT users.id AS id, users.name AS name FROM users) AS anon_1",
- dialect=default.DefaultDialect()
)
- self.assert_compile(sess.query(users, exists([1], from_obj=addresses)).with_labels().statement,
+ self.assert_compile(sess.query(users, exists([1], from_obj=addresses)
+ ).with_labels().statement,
"SELECT users.id AS users_id, users.name AS users_name, EXISTS "
"(SELECT 1 FROM addresses) AS anon_1 FROM users",
- dialect=default.DefaultDialect()
)
- # a little tedious here, adding labels to work around Query's auto-labelling.
- # TODO: can we detect only one table in the "froms" and then turn off use_labels ?
- s = sess.query(addresses.c.id.label('id'), addresses.c.email_address.label('email')).\
- filter(addresses.c.user_id==users.c.id).statement.alias()
-
- self.assert_compile(sess.query(users, s.c.email).select_from(users.join(s, s.c.id==users.c.id)).with_labels().statement,
- "SELECT users.id AS users_id, users.name AS users_name, anon_1.email AS anon_1_email "
- "FROM users JOIN (SELECT addresses.id AS id, addresses.email_address AS email FROM addresses "
- "WHERE addresses.user_id = users.id) AS anon_1 ON anon_1.id = users.id",
- dialect=default.DefaultDialect()
+ # a little tedious here, adding labels to work around Query's
+ # auto-labelling. TODO: can we detect only one table in the
+ # "froms" and then turn off use_labels ? note: this query is
+ # incorrect SQL with the correlate of users in the FROM list.
+ s = sess.query(addresses.c.id.label('id'),
+ addresses.c.email_address.label('email')).\
+ filter(addresses.c.user_id == users.c.id).correlate(users).\
+ statement.alias()
+
+ self.assert_compile(sess.query(users, s.c.email).select_from(
+ users.join(s, s.c.id == users.c.id)
+ ).with_labels().statement,
+ "SELECT users.id AS users_id, users.name AS users_name, "
+ "anon_1.email AS anon_1_email "
+ "FROM users JOIN (SELECT addresses.id AS id, "
+ "addresses.email_address AS email FROM addresses "
+ "WHERE addresses.user_id = users.id) AS anon_1 "
+ "ON anon_1.id = users.id",
)
x = func.lala(users.c.id).label('foo')
- self.assert_compile(sess.query(x).filter(x==5).statement,
- "SELECT lala(users.id) AS foo FROM users WHERE lala(users.id) = :param_1", dialect=default.DefaultDialect())
+ self.assert_compile(sess.query(x).filter(x == 5).statement,
+ "SELECT lala(users.id) AS foo FROM users WHERE "
+ "lala(users.id) = :param_1")
self.assert_compile(sess.query(func.sum(x).label('bar')).statement,
- "SELECT sum(lala(users.id)) AS bar FROM users", dialect=default.DefaultDialect())
+ "SELECT sum(lala(users.id)) AS bar FROM users")
class FromSelfTest(QueryTest, AssertsCompiledSQL):
def test_where_subquery(self):
s = select([addresses.c.street], addresses.c.user_id
== users.c.user_id, correlate=True).alias('s')
+
+ # don't correlate in a FROM list
self.assert_compile(select([users, s.c.street], from_obj=s),
"SELECT users.user_id, users.user_name, "
"users.password, s.street FROM users, "
"(SELECT addresses.street AS street FROM "
- "addresses WHERE addresses.user_id = "
+ "addresses, users WHERE addresses.user_id = "
"users.user_id) AS s")
self.assert_compile(table1.select(table1.c.myid
== select([table1.c.myid], table1.c.name
'mytable AS ta WHERE EXISTS (SELECT 1 FROM '
'myothertable WHERE myothertable.otherid = '
'ta.myid)) AS sq2, mytable')
- s = select([addresses.c.street], addresses.c.user_id
- == users.c.user_id, correlate=True).alias('s')
- self.assert_compile(select([users, s.c.street], from_obj=s),
- "SELECT users.user_id, users.user_name, "
- "users.password, s.street FROM users, "
- "(SELECT addresses.street AS street FROM "
- "addresses WHERE addresses.user_id = "
- "users.user_id) AS s")
+
# test constructing the outer query via append_column(), which
# occurs in the ORM's Query object
column("col3"),
)
- def test_select(self):
- self.assert_compile(t1.select().where(t1.c.col1
- == 5).order_by(t1.c.col3),
- 'SELECT table1.col1, table1.col2, '
- 'table1.col3 FROM table1 WHERE table1.col1 '
- '= :col1_1 ORDER BY table1.col3')
- self.assert_compile(t1.select().select_from(select([t2],
- t2.c.col1
- == t1.c.col1)).order_by(t1.c.col3),
- 'SELECT table1.col1, table1.col2, '
- 'table1.col3 FROM table1, (SELECT '
- 'table2.col1 AS col1, table2.col2 AS col2, '
- 'table2.col3 AS col3 FROM table2 WHERE '
- 'table2.col1 = table1.col1) ORDER BY '
- 'table1.col3')
- s = select([t2], t2.c.col1 == t1.c.col1, correlate=False)
- s = s.correlate(t1).order_by(t2.c.col3)
- self.assert_compile(t1.select().select_from(s).order_by(t1.c.col3),
- 'SELECT table1.col1, table1.col2, '
- 'table1.col3 FROM table1, (SELECT '
- 'table2.col1 AS col1, table2.col2 AS col2, '
- 'table2.col3 AS col3 FROM table2 WHERE '
- 'table2.col1 = table1.col1 ORDER BY '
- 'table2.col3) ORDER BY table1.col3')
def test_columns(self):
s = t1.select()
'table2.col3 FROM table2, table1 WHERE '
'table1.col1 = table2.col1')
s2 = select([t1], t1.c.col2 == s.c.col2)
+ # dont correlate in a FROM entry
self.assert_compile(s2,
'SELECT table1.col1, table1.col2, '
'table1.col3 FROM table1, (SELECT '
'table2.col1 AS col1, table2.col2 AS col2, '
- 'table2.col3 AS col3 FROM table2 WHERE '
+ 'table2.col3 AS col3 FROM table2, table1 WHERE '
'table1.col1 = table2.col1) WHERE '
'table1.col2 = col2')
s3 = s.correlate(None)
'table2.col3 AS col3 FROM table2, table1 '
'WHERE table1.col1 = table2.col1) WHERE '
'table1.col2 = col2')
+ # dont correlate in a FROM entry
self.assert_compile(select([t1], t1.c.col2 == s.c.col2),
'SELECT table1.col1, table1.col2, '
'table1.col3 FROM table1, (SELECT '
'table2.col1 AS col1, table2.col2 AS col2, '
- 'table2.col3 AS col3 FROM table2 WHERE '
+ 'table2.col3 AS col3 FROM table2, table1 WHERE '
'table1.col1 = table2.col1) WHERE '
'table1.col2 = col2')
+
+ # but correlate in a WHERE entry
+ s_w = select([t2.c.col1]).where(t1.c.col1 == t2.c.col1)
+ self.assert_compile(select([t1], t1.c.col2 == s_w),
+ 'SELECT table1.col1, table1.col2, table1.col3 '
+ 'FROM table1 WHERE table1.col2 = '
+ '(SELECT table2.col1 FROM table2 '
+ 'WHERE table1.col1 = table2.col1)'
+ )
+
+
s4 = s3.correlate(t1)
self.assert_compile(select([t1], t1.c.col2 == s4.c.col2),
'SELECT table1.col1, table1.col2, '
'table2.col3 AS col3 FROM table2 WHERE '
'table1.col1 = table2.col1) WHERE '
'table1.col2 = col2')
+
self.assert_compile(select([t1], t1.c.col2 == s3.c.col2),
'SELECT table1.col1, table1.col2, '
'table1.col3 FROM table1, (SELECT '
'WHERE table1.col1 = table2.col1) WHERE '
'table1.col2 = col2')
+ self.assert_compile(t1.select().where(t1.c.col1
+ == 5).order_by(t1.c.col3),
+ 'SELECT table1.col1, table1.col2, '
+ 'table1.col3 FROM table1 WHERE table1.col1 '
+ '= :col1_1 ORDER BY table1.col3')
+
+ # dont correlate in FROM
+ self.assert_compile(t1.select().select_from(select([t2],
+ t2.c.col1
+ == t1.c.col1)).order_by(t1.c.col3),
+ 'SELECT table1.col1, table1.col2, '
+ 'table1.col3 FROM table1, (SELECT '
+ 'table2.col1 AS col1, table2.col2 AS col2, '
+ 'table2.col3 AS col3 FROM table2, table1 WHERE '
+ 'table2.col1 = table1.col1) ORDER BY '
+ 'table1.col3')
+
+ # still works if you actually add that table to correlate()
+ s = select([t2], t2.c.col1 == t1.c.col1)
+ s = s.correlate(t1).order_by(t2.c.col3)
+
+ self.assert_compile(t1.select().select_from(s).order_by(t1.c.col3),
+ 'SELECT table1.col1, table1.col2, '
+ 'table1.col3 FROM table1, (SELECT '
+ 'table2.col1 AS col1, table2.col2 AS col2, '
+ 'table2.col3 AS col3 FROM table2 WHERE '
+ 'table2.col1 = table1.col1 ORDER BY '
+ 'table2.col3) ORDER BY table1.col3')
+
def test_prefixes(self):
s = t1.select()
self.assert_compile(s,
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
