def test_correlated_select(self):
s = select(['*'], t1.c.col1 == t2.c.col1,
from_obj=[t1, t2]).correlate(t2)
+
class Vis(CloningVisitor):
def visit_select(self, select):
select.append_whereclause(t1.c.col2 == 7)
- self.assert_compile(Vis().traverse(s),
- "SELECT * FROM table1 WHERE table1.col1 = table2.col1 "
- "AND table1.col2 = :col2_1")
+ self.assert_compile(
+ select([t2]).where(t2.c.col1 == Vis().traverse(s)),
+ "SELECT table2.col1, table2.col2, table2.col3 "
+ "FROM table2 WHERE table2.col1 = "
+ "(SELECT * FROM table1 WHERE table1.col1 = table2.col1 "
+ "AND table1.col2 = :col2_1)"
+ )
def test_this_thing(self):
s = select([t1]).where(t1.c.col1 == 'foo').alias()
'AS table1_1 WHERE table1_1.col1 = '
':col1_1) AS anon_1')
- def test_select_fromtwice(self):
+ def test_select_fromtwice_one(self):
t1a = t1.alias()
- s = select([1], t1.c.col1 == t1a.c.col1, from_obj=t1a).correlate(t1)
+ s = select([1], t1.c.col1 == t1a.c.col1, from_obj=t1a).correlate(t1a)
+ s = select([t1]).where(t1.c.col1 == s)
self.assert_compile(s,
- 'SELECT 1 FROM table1 AS table1_1 WHERE '
- 'table1.col1 = table1_1.col1')
-
+ "SELECT table1.col1, table1.col2, table1.col3 FROM table1 "
+ "WHERE table1.col1 = "
+ "(SELECT 1 FROM table1, table1 AS table1_1 "
+ "WHERE table1.col1 = table1_1.col1)"
+ )
s = CloningVisitor().traverse(s)
self.assert_compile(s,
- 'SELECT 1 FROM table1 AS table1_1 WHERE '
- 'table1.col1 = table1_1.col1')
+ "SELECT table1.col1, table1.col2, table1.col3 FROM table1 "
+ "WHERE table1.col1 = "
+ "(SELECT 1 FROM table1, table1 AS table1_1 "
+ "WHERE table1.col1 = table1_1.col1)")
+ def test_select_fromtwice_two(self):
s = select([t1]).where(t1.c.col1 == 'foo').alias()
s2 = select([1], t1.c.col1 == s.c.col1, from_obj=s).correlate(t1)
- self.assert_compile(s2,
- 'SELECT 1 FROM (SELECT table1.col1 AS '
- 'col1, table1.col2 AS col2, table1.col3 AS '
- 'col3 FROM table1 WHERE table1.col1 = '
- ':col1_1) AS anon_1 WHERE table1.col1 = '
- 'anon_1.col1')
- s2 = ReplacingCloningVisitor().traverse(s2)
- self.assert_compile(s2,
- 'SELECT 1 FROM (SELECT table1.col1 AS '
- 'col1, table1.col2 AS col2, table1.col3 AS '
- 'col3 FROM table1 WHERE table1.col1 = '
- ':col1_1) AS anon_1 WHERE table1.col1 = '
- 'anon_1.col1')
+ s3 = select([t1]).where(t1.c.col1 == s2)
+ self.assert_compile(s3,
+ "SELECT table1.col1, table1.col2, table1.col3 "
+ "FROM table1 WHERE table1.col1 = "
+ "(SELECT 1 FROM "
+ "(SELECT table1.col1 AS col1, table1.col2 AS col2, "
+ "table1.col3 AS col3 FROM table1 "
+ "WHERE table1.col1 = :col1_1) "
+ "AS anon_1 WHERE table1.col1 = anon_1.col1)"
+ )
+
+ s4 = ReplacingCloningVisitor().traverse(s3)
+ self.assert_compile(s4,
+ "SELECT table1.col1, table1.col2, table1.col3 "
+ "FROM table1 WHERE table1.col1 = "
+ "(SELECT 1 FROM "
+ "(SELECT table1.col1 AS col1, table1.col2 AS col2, "
+ "table1.col3 AS col3 FROM table1 "
+ "WHERE table1.col1 = :col1_1) "
+ "AS anon_1 WHERE table1.col1 = anon_1.col1)"
+ )
class ClauseAdapterTest(fixtures.TestBase, AssertsCompiledSQL):
__dialect__ = 'default'
def test_table_to_alias_6(self):
t1alias = t1.alias('t1alias')
vis = sql_util.ClauseAdapter(t1alias)
- self.assert_compile(select(t1).where(
- t1.c.col1 == vis.traverse(select(['*'], t1.c.col1
- == t2.c.col2, from_obj=[t1,
- t2]).correlate(t1))),
- 'SELECT * FROM table2 WHERE t1alias.col1 = '
- 'table2.col2')
+ self.assert_compile(
+ select([t1alias, t2]).where(t1alias.c.col1 ==
+ vis.traverse(select(['*'],
+ t1.c.col1 == t2.c.col2,
+ from_obj=[t1, t2]).correlate(t1))),
+ "SELECT t1alias.col1, t1alias.col2, t1alias.col3, "
+ "table2.col1, table2.col2, table2.col3 "
+ "FROM table1 AS t1alias, table2 WHERE t1alias.col1 = "
+ "(SELECT * FROM table2 WHERE t1alias.col1 = table2.col2)"
+ )
def test_table_to_alias_7(self):
t1alias = t1.alias('t1alias')
vis = sql_util.ClauseAdapter(t1alias)
- self.assert_compile(vis.traverse(select(['*'], t1.c.col1
- == t2.c.col2, from_obj=[t1,
- t2]).correlate(t2)),
- 'SELECT * FROM table1 AS t1alias WHERE '
- 't1alias.col1 = table2.col2')
+ self.assert_compile(
+ select([t1alias, t2]).where(t1alias.c.col1 ==
+ vis.traverse(select(['*'],
+ t1.c.col1 == t2.c.col2,
+ from_obj=[t1, t2]).correlate(t2))),
+ "SELECT t1alias.col1, t1alias.col2, t1alias.col3, "
+ "table2.col1, table2.col2, table2.col3 "
+ "FROM table1 AS t1alias, table2 "
+ "WHERE t1alias.col1 = "
+ "(SELECT * FROM table1 AS t1alias "
+ "WHERE t1alias.col1 = table2.col2)")
def test_table_to_alias_8(self):
t1alias = t1.alias('t1alias')
def test_table_to_alias_11(self):
s = select(['*'], from_obj=[t1]).alias('foo')
+ t1alias = t1.alias('t1alias')
vis = sql_util.ClauseAdapter(t1alias)
self.assert_compile(vis.traverse(s.select()),
'SELECT foo.* FROM (SELECT * FROM table1 '
'AS foo')
def test_table_to_alias_13(self):
- s = select(['*'], from_obj=[t1]).alias('foo')
+ t1alias = t1.alias('t1alias')
vis = sql_util.ClauseAdapter(t1alias)
ff = vis.traverse(func.count(t1.c.col1).label('foo'))
self.assert_compile(select([ff]),
vis = sql_util.ClauseAdapter(t1alias)
t2alias = t2.alias('t2alias')
vis.chain(sql_util.ClauseAdapter(t2alias))
- self.assert_compile(vis.traverse(select(['*'], t1.c.col1
- == t2.c.col2, from_obj=[t1,
- t2]).correlate(t1)),
- 'SELECT * FROM table2 AS t2alias WHERE '
- 't1alias.col1 = t2alias.col2')
+ self.assert_compile(
+ select([t1alias, t2alias]).where(
+ t1alias.c.col1 ==
+ vis.traverse(select(['*'],
+ t1.c.col1 == t2.c.col2,
+ from_obj=[t1, t2]).correlate(t1))
+ ),
+ "SELECT t1alias.col1, t1alias.col2, t1alias.col3, "
+ "t2alias.col1, t2alias.col2, t2alias.col3 "
+ "FROM table1 AS t1alias, table2 AS t2alias "
+ "WHERE t1alias.col1 = "
+ "(SELECT * FROM table2 AS t2alias "
+ "WHERE t1alias.col1 = t2alias.col2)"
+ )
def test_table_to_alias_17(self):
t1alias = t1.alias('t1alias')
'SELECT table1.col1, table1.col2, '
'table1.col3 FROM table1')
- def test_correlation(self):
- s = select([t2], t1.c.col1 == t2.c.col1)
- self.assert_compile(s,
- 'SELECT table2.col1, table2.col2, '
- '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, table1 WHERE '
- 'table1.col1 = table2.col1) WHERE '
- 'table1.col2 = col2')
- s3 = s.correlate(None)
- self.assert_compile(select([t1], t1.c.col2 == s3.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, 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, 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, '
- 'table1.col3 FROM table1, (SELECT '
- 'table2.col1 AS col1, table2.col2 AS 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 '
- 'table2.col1 AS col1, table2.col2 AS col2, '
- 'table2.col3 AS col3 FROM table2, table1 '
- '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()