From: Mike Bayer Date: Fri, 11 Feb 2011 03:09:26 +0000 (-0500) Subject: fix some really egregious long lines X-Git-Tag: rel_0_7b1~13 X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=3cbab1ee1ecc4dc2ae74eaccbfd6de2fcdeed22c;p=thirdparty%2Fsqlalchemy%2Fsqlalchemy.git fix some really egregious long lines --- diff --git a/test/sql/test_generative.py b/test/sql/test_generative.py index f132cbb2d7..c6f5dc05ef 100644 --- a/test/sql/test_generative.py +++ b/test/sql/test_generative.py @@ -8,15 +8,16 @@ from sqlalchemy.sql import util as sql_util from test.lib.testing import eq_ class TraversalTest(TestBase, AssertsExecutionResults): - """test ClauseVisitor's traversal, particularly its ability to copy and modify - a ClauseElement in place.""" + """test ClauseVisitor's traversal, particularly its + ability to copy and modify a ClauseElement in place.""" @classmethod def setup_class(cls): global A, B # establish two ficticious ClauseElements. - # define deep equality semantics as well as deep identity semantics. + # define deep equality semantics as well as deep + # identity semantics. class A(ClauseElement): __visit_name__ = 'a' @@ -78,7 +79,8 @@ class TraversalTest(TestBase, AssertsExecutionResults): a1 = A("expr1") struct = B(a1, A("expr2"), B(A("expr1b"), A("expr2b")), A("expr3")) struct2 = B(a1, A("expr2"), B(A("expr1b"), A("expr2b")), A("expr3")) - struct3 = B(a1, A("expr2"), B(A("expr1b"), A("expr2bmodified")), A("expr3")) + struct3 = B(a1, A("expr2"), B(A("expr1b"), + A("expr2bmodified")), A("expr3")) assert a1.is_other(a1) assert struct.is_other(struct) @@ -88,7 +90,8 @@ class TraversalTest(TestBase, AssertsExecutionResults): assert not struct.is_other(struct3) def test_clone(self): - struct = B(A("expr1"), A("expr2"), B(A("expr1b"), A("expr2b")), A("expr3")) + struct = B(A("expr1"), A("expr2"), B(A("expr1b"), + A("expr2b")), A("expr3")) class Vis(CloningVisitor): def visit_a(self, a): @@ -102,7 +105,8 @@ class TraversalTest(TestBase, AssertsExecutionResults): assert not struct.is_other(s2) def test_no_clone(self): - struct = B(A("expr1"), A("expr2"), B(A("expr1b"), A("expr2b")), A("expr3")) + struct = B(A("expr1"), A("expr2"), B(A("expr1b"), + A("expr2b")), A("expr3")) class Vis(ClauseVisitor): def visit_a(self, a): @@ -116,9 +120,12 @@ class TraversalTest(TestBase, AssertsExecutionResults): assert struct.is_other(s2) def test_change_in_place(self): - struct = B(A("expr1"), A("expr2"), B(A("expr1b"), A("expr2b")), A("expr3")) - struct2 = B(A("expr1"), A("expr2modified"), B(A("expr1b"), A("expr2b")), A("expr3")) - struct3 = B(A("expr1"), A("expr2"), B(A("expr1b"), A("expr2bmodified")), A("expr3")) + struct = B(A("expr1"), A("expr2"), B(A("expr1b"), + A("expr2b")), A("expr3")) + struct2 = B(A("expr1"), A("expr2modified"), B(A("expr1b"), + A("expr2b")), A("expr3")) + struct3 = B(A("expr1"), A("expr2"), B(A("expr1b"), + A("expr2bmodified")), A("expr3")) class Vis(CloningVisitor): def visit_a(self, a): @@ -185,14 +192,16 @@ class ClauseTest(TestBase, AssertsCompiledSQL): f = t.c.col1 * 5 - self.assert_compile(select([f]), "SELECT t1.col1 * :col1_1 AS anon_1 FROM t1") + self.assert_compile(select([f]), + "SELECT t1.col1 * :col1_1 AS anon_1 FROM t1") f.anon_label a = t.alias() f = sql_util.ClauseAdapter(a).traverse(f) - self.assert_compile(select([f]), "SELECT t1_1.col1 * :col1_1 AS anon_1 FROM t1 AS t1_1") + self.assert_compile(select([f]), + "SELECT t1_1.col1 * :col1_1 AS anon_1 FROM t1 AS t1_1") def test_join(self): clause = t1.join(t2, t1.c.col2==t2.c.col2) @@ -233,7 +242,9 @@ class ClauseTest(TestBase, AssertsCompiledSQL): def test_text(self): - clause = text("select * from table where foo=:bar", bindparams=[bindparam('bar')]) + clause = text( + "select * from table where foo=:bar", + bindparams=[bindparam('bar')]) c1 = str(clause) class Vis(CloningVisitor): def visit_textclause(self, text): @@ -322,44 +333,63 @@ class ClauseTest(TestBase, AssertsCompiledSQL): assert str(expr) == str(expr2) def test_adapt_union(self): - u = union(t1.select().where(t1.c.col1==4), t1.select().where(t1.c.col1==5)).alias() + u = union( + t1.select().where(t1.c.col1==4), + t1.select().where(t1.c.col1==5) + ).alias() assert sql_util.ClauseAdapter(u).traverse(t1) is u def test_binds(self): - """test that unique bindparams change their name upon clone() to prevent conflicts""" + """test that unique bindparams change their name upon clone() + to prevent conflicts""" s = select([t1], t1.c.col1==bindparam(None, unique=True)).alias() s2 = CloningVisitor().traverse(s).alias() s3 = select([s], s.c.col2==s2.c.col2) - self.assert_compile(s3, "SELECT anon_1.col1, anon_1.col2, anon_1.col3 FROM (SELECT table1.col1 AS col1, table1.col2 AS col2, "\ - "table1.col3 AS col3 FROM table1 WHERE table1.col1 = :param_1) AS anon_1, "\ - "(SELECT table1.col1 AS col1, table1.col2 AS col2, table1.col3 AS col3 FROM table1 WHERE table1.col1 = :param_2) AS anon_2 "\ - "WHERE anon_1.col2 = anon_2.col2") + self.assert_compile(s3, + "SELECT anon_1.col1, anon_1.col2, anon_1.col3 FROM " + "(SELECT table1.col1 AS col1, table1.col2 AS col2, " + "table1.col3 AS col3 FROM table1 WHERE table1.col1 = :param_1) " + "AS anon_1, " + "(SELECT table1.col1 AS col1, table1.col2 AS col2, table1.col3 " + "AS col3 FROM table1 WHERE table1.col1 = :param_2) AS anon_2 " + "WHERE anon_1.col2 = anon_2.col2") s = select([t1], t1.c.col1==4).alias() s2 = CloningVisitor().traverse(s).alias() s3 = select([s], s.c.col2==s2.c.col2) - self.assert_compile(s3, "SELECT anon_1.col1, anon_1.col2, anon_1.col3 FROM (SELECT table1.col1 AS col1, table1.col2 AS col2, "\ - "table1.col3 AS col3 FROM table1 WHERE table1.col1 = :col1_1) AS anon_1, "\ - "(SELECT table1.col1 AS col1, table1.col2 AS col2, table1.col3 AS col3 FROM table1 WHERE table1.col1 = :col1_2) AS anon_2 "\ - "WHERE anon_1.col2 = anon_2.col2") + self.assert_compile(s3, + "SELECT anon_1.col1, anon_1.col2, anon_1.col3 FROM " + "(SELECT table1.col1 AS col1, table1.col2 AS col2, " + "table1.col3 AS col3 FROM table1 WHERE table1.col1 = :col1_1) " + "AS anon_1, " + "(SELECT table1.col1 AS col1, table1.col2 AS col2, table1.col3 " + "AS col3 FROM table1 WHERE table1.col1 = :col1_2) AS anon_2 " + "WHERE anon_1.col2 = anon_2.col2") def test_extract(self): s = select([extract('foo', t1.c.col1).label('col1')]) - self.assert_compile(s, "SELECT EXTRACT(foo FROM table1.col1) AS col1 FROM table1") + self.assert_compile(s, + "SELECT EXTRACT(foo FROM table1.col1) AS col1 FROM table1") s2 = CloningVisitor().traverse(s).alias() s3 = select([s2.c.col1]) - self.assert_compile(s, "SELECT EXTRACT(foo FROM table1.col1) AS col1 FROM table1") - self.assert_compile(s3, "SELECT anon_1.col1 FROM (SELECT EXTRACT(foo FROM table1.col1) AS col1 FROM table1) AS anon_1") + self.assert_compile(s, + "SELECT EXTRACT(foo FROM table1.col1) AS col1 FROM table1") + self.assert_compile(s3, + "SELECT anon_1.col1 FROM (SELECT EXTRACT(foo FROM " + "table1.col1) AS col1 FROM table1) AS anon_1") @testing.emits_warning('.*replaced by another column with the same key') def test_alias(self): subq = t2.select().alias('subq') - s = select([t1.c.col1, subq.c.col1], from_obj=[t1, subq, t1.join(subq, t1.c.col1==subq.c.col2)]) + s = select([t1.c.col1, subq.c.col1], + from_obj=[t1, subq, + t1.join(subq, t1.c.col1==subq.c.col2)] + ) orig = str(s) s2 = CloningVisitor().traverse(s) assert orig == str(s) == str(s2) @@ -374,42 +404,71 @@ class ClauseTest(TestBase, AssertsCompiledSQL): assert orig == str(s) == str(s3) == str(s4) subq = subq.alias('subq') - s = select([t1.c.col1, subq.c.col1], from_obj=[t1, subq, t1.join(subq, t1.c.col1==subq.c.col2)]) + s = select([t1.c.col1, subq.c.col1], + from_obj=[t1, subq, + t1.join(subq, t1.c.col1==subq.c.col2)] + ) s5 = CloningVisitor().traverse(s) assert orig == str(s) == str(s5) def test_correlated_select(self): - s = select(['*'], t1.c.col1==t2.c.col1, from_obj=[t1, t2]).correlate(t2) + 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(Vis().traverse(s), + "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() s2 = select([s.c.col1]) - self.assert_compile(s2, "SELECT anon_1.col1 FROM (SELECT table1.col1 AS col1, table1.col2 AS col2, table1.col3 AS col3 FROM table1 WHERE table1.col1 = :col1_1) AS anon_1") + self.assert_compile(s2, + 'SELECT anon_1.col1 FROM (SELECT ' + 'table1.col1 AS col1, table1.col2 AS col2, ' + 'table1.col3 AS col3 FROM table1 WHERE ' + 'table1.col1 = :col1_1) AS anon_1') t1a = t1.alias() s2 = sql_util.ClauseAdapter(t1a).traverse(s2) - self.assert_compile(s2, "SELECT anon_1.col1 FROM (SELECT table1_1.col1 AS col1, table1_1.col2 AS col2, table1_1.col3 AS col3 FROM table1 AS table1_1 WHERE table1_1.col1 = :col1_1) AS anon_1") + self.assert_compile(s2, + 'SELECT anon_1.col1 FROM (SELECT ' + 'table1_1.col1 AS col1, table1_1.col2 AS ' + 'col2, table1_1.col3 AS col3 FROM table1 ' + 'AS table1_1 WHERE table1_1.col1 = ' + ':col1_1) AS anon_1') def test_select_fromtwice(self): t1a = t1.alias() s = select([1], t1.c.col1==t1a.c.col1, from_obj=t1a).correlate(t1) - self.assert_compile(s, "SELECT 1 FROM table1 AS table1_1 WHERE table1.col1 = table1_1.col1") + self.assert_compile(s, + 'SELECT 1 FROM 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") + self.assert_compile(s, + 'SELECT 1 FROM table1 AS table1_1 WHERE ' + 'table1.col1 = table1_1.col1') 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") + 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") + 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') class ClauseAdapterTest(TestBase, AssertsCompiledSQL): @classmethod @@ -435,54 +494,96 @@ class ClauseAdapterTest(TestBase, AssertsCompiledSQL): assert t2alias in s._froms assert t1alias in s._froms - self.assert_compile(select(['*'], t2alias.c.col1==s), "SELECT * FROM table2 AS t2alias WHERE t2alias.col1 = (SELECT * FROM table1 AS t1alias)") + self.assert_compile(select(['*'], t2alias.c.col1 == s), + 'SELECT * FROM table2 AS t2alias WHERE ' + 't2alias.col1 = (SELECT * FROM table1 AS ' + 't1alias)') s = vis.traverse(s) - assert t2alias not in s._froms # not present because it's been cloned - - assert t1alias in s._froms # present because the adapter placed it there - - # correlate list on "s" needs to take into account the full _cloned_set for each element in _froms when correlating - self.assert_compile(select(['*'], t2alias.c.col1==s), "SELECT * FROM table2 AS t2alias WHERE t2alias.col1 = (SELECT * FROM table1 AS t1alias)") - - s = select(['*'], from_obj=[t1alias, t2alias]).correlate(t2alias).as_scalar() - self.assert_compile(select(['*'], t2alias.c.col1==s), "SELECT * FROM table2 AS t2alias WHERE t2alias.col1 = (SELECT * FROM table1 AS t1alias)") + assert t2alias not in s._froms # not present because it's been + # cloned + assert t1alias in s._froms # present because the adapter placed + # it there + + # correlate list on "s" needs to take into account the full + # _cloned_set for each element in _froms when correlating + + self.assert_compile(select(['*'], t2alias.c.col1 == s), + 'SELECT * FROM table2 AS t2alias WHERE ' + 't2alias.col1 = (SELECT * FROM table1 AS ' + 't1alias)') + s = select(['*'], from_obj=[t1alias, + t2alias]).correlate(t2alias).as_scalar() + self.assert_compile(select(['*'], t2alias.c.col1 == s), + 'SELECT * FROM table2 AS t2alias WHERE ' + 't2alias.col1 = (SELECT * FROM table1 AS ' + 't1alias)') s = vis.traverse(s) - self.assert_compile(select(['*'], t2alias.c.col1==s), "SELECT * FROM table2 AS t2alias WHERE t2alias.col1 = (SELECT * FROM table1 AS t1alias)") + self.assert_compile(select(['*'], t2alias.c.col1 == s), + 'SELECT * FROM table2 AS t2alias WHERE ' + 't2alias.col1 = (SELECT * FROM table1 AS ' + 't1alias)') s = CloningVisitor().traverse(s) - self.assert_compile(select(['*'], t2alias.c.col1==s), "SELECT * FROM table2 AS t2alias WHERE t2alias.col1 = (SELECT * FROM table1 AS t1alias)") - - s = select(['*']).where(t1.c.col1==t2.c.col1).as_scalar() - self.assert_compile(select([t1.c.col1, s]), "SELECT table1.col1, (SELECT * FROM table2 WHERE table1.col1 = table2.col1) AS anon_1 FROM table1") + self.assert_compile(select(['*'], t2alias.c.col1 == s), + 'SELECT * FROM table2 AS t2alias WHERE ' + 't2alias.col1 = (SELECT * FROM table1 AS ' + 't1alias)') + + s = select(['*']).where(t1.c.col1 == t2.c.col1).as_scalar() + self.assert_compile(select([t1.c.col1, s]), + 'SELECT table1.col1, (SELECT * FROM table2 ' + 'WHERE table1.col1 = table2.col1) AS ' + 'anon_1 FROM table1') vis = sql_util.ClauseAdapter(t1alias) s = vis.traverse(s) - self.assert_compile(select([t1alias.c.col1, s]), "SELECT t1alias.col1, (SELECT * FROM table2 WHERE t1alias.col1 = table2.col1) AS anon_1 FROM table1 AS t1alias") + self.assert_compile(select([t1alias.c.col1, s]), + 'SELECT t1alias.col1, (SELECT * FROM ' + 'table2 WHERE t1alias.col1 = table2.col1) ' + 'AS anon_1 FROM table1 AS t1alias') s = CloningVisitor().traverse(s) - self.assert_compile(select([t1alias.c.col1, s]), "SELECT t1alias.col1, (SELECT * FROM table2 WHERE t1alias.col1 = table2.col1) AS anon_1 FROM table1 AS t1alias") - - s = select(['*']).where(t1.c.col1==t2.c.col1).correlate(t1).as_scalar() - self.assert_compile(select([t1.c.col1, s]), "SELECT table1.col1, (SELECT * FROM table2 WHERE table1.col1 = table2.col1) AS anon_1 FROM table1") + self.assert_compile(select([t1alias.c.col1, s]), + 'SELECT t1alias.col1, (SELECT * FROM ' + 'table2 WHERE t1alias.col1 = table2.col1) ' + 'AS anon_1 FROM table1 AS t1alias') + s = select(['*']).where(t1.c.col1 + == t2.c.col1).correlate(t1).as_scalar() + self.assert_compile(select([t1.c.col1, s]), + 'SELECT table1.col1, (SELECT * FROM table2 ' + 'WHERE table1.col1 = table2.col1) AS ' + 'anon_1 FROM table1') vis = sql_util.ClauseAdapter(t1alias) s = vis.traverse(s) - self.assert_compile(select([t1alias.c.col1, s]), "SELECT t1alias.col1, (SELECT * FROM table2 WHERE t1alias.col1 = table2.col1) AS anon_1 FROM table1 AS t1alias") + self.assert_compile(select([t1alias.c.col1, s]), + 'SELECT t1alias.col1, (SELECT * FROM ' + 'table2 WHERE t1alias.col1 = table2.col1) ' + 'AS anon_1 FROM table1 AS t1alias') s = CloningVisitor().traverse(s) - self.assert_compile(select([t1alias.c.col1, s]), "SELECT t1alias.col1, (SELECT * FROM table2 WHERE t1alias.col1 = table2.col1) AS anon_1 FROM table1 AS t1alias") + self.assert_compile(select([t1alias.c.col1, s]), + 'SELECT t1alias.col1, (SELECT * FROM ' + 'table2 WHERE t1alias.col1 = table2.col1) ' + 'AS anon_1 FROM table1 AS t1alias') @testing.fails_on_everything_except() def test_joins_dont_adapt(self): - # adapting to a join, i.e. ClauseAdapter(t1.join(t2)), doesn't make much sense. - # ClauseAdapter doesn't make any changes if it's against a straight join. + # adapting to a join, i.e. ClauseAdapter(t1.join(t2)), doesn't + # make much sense. ClauseAdapter doesn't make any changes if + # it's against a straight join. + users = table('users', column('id')) addresses = table('addresses', column('id'), column('user_id')) ualias = users.alias() - s = select([func.count(addresses.c.id)], users.c.id==addresses.c.user_id).correlate(users) #.as_scalar().label(None) - s= sql_util.ClauseAdapter(ualias).traverse(s) + s = select([func.count(addresses.c.id)], users.c.id + == addresses.c.user_id).correlate(users) + s = sql_util.ClauseAdapter(ualias).traverse(s) j1 = addresses.join(ualias, addresses.c.user_id==ualias.c.id) - self.assert_compile(sql_util.ClauseAdapter(j1).traverse(s), "SELECT count(addresses.id) AS count_1 FROM addresses WHERE users_1.id = addresses.user_id") + self.assert_compile(sql_util.ClauseAdapter(j1).traverse(s), + 'SELECT count(addresses.id) AS count_1 ' + 'FROM addresses WHERE users_1.id = ' + 'addresses.user_id') def test_table_to_alias(self): @@ -492,45 +593,89 @@ class ClauseAdapterTest(TestBase, AssertsCompiledSQL): ff = vis.traverse(func.count(t1.c.col1).label('foo')) assert list(_from_objects(ff)) == [t1alias] - self.assert_compile(vis.traverse(select(['*'], from_obj=[t1])), "SELECT * FROM table1 AS t1alias") - self.assert_compile(select(['*'], t1.c.col1==t2.c.col2), "SELECT * FROM table1, table2 WHERE table1.col1 = table2.col2") - self.assert_compile(vis.traverse(select(['*'], t1.c.col1==t2.c.col2)), "SELECT * FROM table1 AS t1alias, table2 WHERE t1alias.col1 = table2.col2") - self.assert_compile(vis.traverse(select(['*'], t1.c.col1==t2.c.col2, from_obj=[t1, t2])), "SELECT * FROM table1 AS t1alias, table2 WHERE t1alias.col1 = table2.col2") - self.assert_compile(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(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(vis.traverse(case([(t1.c.col1==5, t1.c.col2)], else_=t1.c.col1)), - "CASE WHEN (t1alias.col1 = :col1_1) THEN t1alias.col2 ELSE t1alias.col1 END" - ) - self.assert_compile(vis.traverse(case([(5, t1.c.col2)], value=t1.c.col1, else_=t1.c.col1)), - "CASE t1alias.col1 WHEN :param_1 THEN t1alias.col2 ELSE t1alias.col1 END" - ) + self.assert_compile(vis.traverse(select(['*'], from_obj=[t1])), + 'SELECT * FROM table1 AS t1alias') + self.assert_compile(select(['*'], t1.c.col1 == t2.c.col2), + 'SELECT * FROM table1, table2 WHERE ' + 'table1.col1 = table2.col2') + self.assert_compile(vis.traverse(select(['*'], t1.c.col1 + == t2.c.col2)), + 'SELECT * FROM table1 AS t1alias, table2 ' + 'WHERE t1alias.col1 = table2.col2') + self.assert_compile(vis.traverse(select(['*'], t1.c.col1 + == t2.c.col2, from_obj=[t1, t2])), + 'SELECT * FROM table1 AS t1alias, table2 ' + 'WHERE t1alias.col1 = table2.col2') + self.assert_compile(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(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(vis.traverse(case([(t1.c.col1 == 5, + t1.c.col2)], else_=t1.c.col1)), + 'CASE WHEN (t1alias.col1 = :col1_1) THEN ' + 't1alias.col2 ELSE t1alias.col1 END') + self.assert_compile(vis.traverse(case([(5, t1.c.col2)], + value=t1.c.col1, else_=t1.c.col1)), + 'CASE t1alias.col1 WHEN :param_1 THEN ' + 't1alias.col2 ELSE t1alias.col1 END') s = select(['*'], from_obj=[t1]).alias('foo') - self.assert_compile(s.select(), "SELECT foo.* FROM (SELECT * FROM table1) AS foo") - self.assert_compile(vis.traverse(s.select()), "SELECT foo.* FROM (SELECT * FROM table1 AS t1alias) AS foo") - self.assert_compile(s.select(), "SELECT foo.* FROM (SELECT * FROM table1) AS foo") - + self.assert_compile(s.select(), + 'SELECT foo.* FROM (SELECT * FROM table1) ' + 'AS foo') + self.assert_compile(vis.traverse(s.select()), + 'SELECT foo.* FROM (SELECT * FROM table1 ' + 'AS t1alias) AS foo') + self.assert_compile(s.select(), + 'SELECT foo.* FROM (SELECT * FROM table1) ' + 'AS foo') ff = vis.traverse(func.count(t1.c.col1).label('foo')) - self.assert_compile(select([ff]), "SELECT count(t1alias.col1) AS foo FROM table1 AS t1alias") + self.assert_compile(select([ff]), + 'SELECT count(t1alias.col1) AS foo FROM ' + 'table1 AS t1alias') assert list(_from_objects(ff)) == [t1alias] -# TODO: - # self.assert_compile(vis.traverse(select([func.count(t1.c.col1).label('foo')]), clone=True), "SELECT count(t1alias.col1) AS foo FROM table1 AS t1alias") + # TODO: self.assert_compile(vis.traverse(select([func.count(t1.c + # .col1).l abel('foo')]), clone=True), "SELECT + # count(t1alias.col1) AS foo FROM table1 AS t1alias") t2alias = t2.alias('t2alias') vis.chain(sql_util.ClauseAdapter(t2alias)) - self.assert_compile(vis.traverse(select(['*'], t1.c.col1==t2.c.col2)), "SELECT * FROM table1 AS t1alias, table2 AS t2alias WHERE t1alias.col1 = t2alias.col2") - self.assert_compile(vis.traverse(select(['*'], t1.c.col1==t2.c.col2, from_obj=[t1, t2])), "SELECT * FROM table1 AS t1alias, table2 AS t2alias WHERE t1alias.col1 = t2alias.col2") - 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(vis.traverse(select(['*'], t1.c.col1==t2.c.col2, from_obj=[t1, t2]).correlate(t2)), "SELECT * FROM table1 AS t1alias WHERE t1alias.col1 = t2alias.col2") + self.assert_compile(vis.traverse(select(['*'], t1.c.col1 + == t2.c.col2)), + 'SELECT * FROM table1 AS t1alias, table2 ' + 'AS t2alias WHERE t1alias.col1 = ' + 't2alias.col2') + self.assert_compile(vis.traverse(select(['*'], t1.c.col1 + == t2.c.col2, from_obj=[t1, t2])), + 'SELECT * FROM table1 AS t1alias, table2 ' + 'AS t2alias WHERE t1alias.col1 = ' + 't2alias.col2') + 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(vis.traverse(select(['*'], t1.c.col1 + == t2.c.col2, from_obj=[t1, + t2]).correlate(t2)), + 'SELECT * FROM table1 AS t1alias WHERE ' + 't1alias.col1 = t2alias.col2') def test_include_exclude(self): m = MetaData() a=Table( 'a',m, Column( 'id', Integer, primary_key=True), - Column( 'xxx_id', Integer, ForeignKey( 'a.id', name='adf',use_alter=True ) ) + Column( 'xxx_id', Integer, + ForeignKey( 'a.id', name='adf',use_alter=True ) + ) ) e = (a.c.id == a.c.xxx_id) @@ -552,7 +697,8 @@ class ClauseAdapterTest(TestBase, AssertsCompiledSQL): # force a recursion overflow, by linking a.c.x<->c.c.x, and # asking for a nonexistent col. corresponding_column should prevent # endless depth. - adapt = sql_util.ClauseAdapter( b, equivalents= {a.c.x: set([ c.c.x]), c.c.x:set([a.c.x])}) + adapt = sql_util.ClauseAdapter(b, + equivalents={a.c.x: set([c.c.x]), c.c.x: set([a.c.x])}) assert adapt._corresponding_column(a.c.x, False) is None def test_multilevel_equivalents(self): @@ -565,7 +711,8 @@ class ClauseAdapterTest(TestBase, AssertsCompiledSQL): # two levels of indirection from c.x->b.x->a.x, requires recursive # corresponding_column call - adapt = sql_util.ClauseAdapter(alias, equivalents= {b.c.x: set([ a.c.x]), c.c.x:set([b.c.x])}) + adapt = sql_util.ClauseAdapter(alias, + equivalents={b.c.x: set([a.c.x]), c.c.x: set([b.c.x])}) assert adapt._corresponding_column(a.c.x, False) is alias.c.x assert adapt._corresponding_column(c.c.x, False) is alias.c.x @@ -593,19 +740,27 @@ class ClauseAdapterTest(TestBase, AssertsCompiledSQL): j3 = c.join(j2, j2.c.b_id==c.c.bid) j4 = j3.outerjoin(d) - self.assert_compile(j4, "c JOIN (SELECT a.id AS a_id, b.id AS b_id, b.aid AS b_aid FROM a LEFT OUTER JOIN b ON a.id = b.aid) " - "ON b_id = c.bid" - " LEFT OUTER JOIN d ON a_id = d.aid") + self.assert_compile(j4, + 'c JOIN (SELECT a.id AS a_id, b.id AS ' + 'b_id, b.aid AS b_aid FROM a LEFT OUTER ' + 'JOIN b ON a.id = b.aid) ON b_id = c.bid ' + 'LEFT OUTER JOIN d ON a_id = d.aid') j5 = j3.alias('foo') j6 = sql_util.ClauseAdapter(j5).copy_and_process([j4])[0] - # this statement takes c join(a join b), wraps it inside an aliased "select * from c join(a join b) AS foo". - # the outermost right side "left outer join d" stays the same, except "d" joins against foo.a_id instead - # of plain "a_id" - self.assert_compile(j6, "(SELECT c.id AS c_id, c.bid AS c_bid, a_id AS a_id, b_id AS b_id, b_aid AS b_aid FROM " - "c JOIN (SELECT a.id AS a_id, b.id AS b_id, b.aid AS b_aid FROM a LEFT OUTER JOIN b ON a.id = b.aid) " - "ON b_id = c.bid) AS foo" - " LEFT OUTER JOIN d ON foo.a_id = d.aid") + # this statement takes c join(a join b), wraps it inside an + # aliased "select * from c join(a join b) AS foo". the outermost + # right side "left outer join d" stays the same, except "d" + # joins against foo.a_id instead of plain "a_id" + + self.assert_compile(j6, + '(SELECT c.id AS c_id, c.bid AS c_bid, ' + 'a_id AS a_id, b_id AS b_id, b_aid AS ' + 'b_aid FROM c JOIN (SELECT a.id AS a_id, ' + 'b.id AS b_id, b.aid AS b_aid FROM a LEFT ' + 'OUTER JOIN b ON a.id = b.aid) ON b_id = ' + 'c.bid) AS foo LEFT OUTER JOIN d ON ' + 'foo.a_id = d.aid') def test_derived_from(self): assert select([t1]).is_derived_from(t1) @@ -621,38 +776,52 @@ class ClauseAdapterTest(TestBase, AssertsCompiledSQL): assert s2.is_derived_from(s1) def test_aliasedselect_to_aliasedselect(self): + # original issue from ticket #904 + s1 = select([t1]).alias('foo') s2 = select([s1]).limit(5).offset(10).alias() - self.assert_compile(sql_util.ClauseAdapter(s2).traverse(s1), - "SELECT foo.col1, foo.col2, foo.col3 FROM (SELECT table1.col1 AS col1, table1.col2 AS col2, table1.col3 AS col3 FROM table1) AS foo LIMIT :param_1 OFFSET :param_2", - {'param_1':5, 'param_2':10} - ) - - j = s1.outerjoin(t2, s1.c.col1==t2.c.col1) + 'SELECT foo.col1, foo.col2, foo.col3 FROM ' + '(SELECT table1.col1 AS col1, table1.col2 ' + 'AS col2, table1.col3 AS col3 FROM table1) ' + 'AS foo LIMIT :param_1 OFFSET :param_2', + {'param_1': 5, 'param_2': 10}) + j = s1.outerjoin(t2, s1.c.col1 == t2.c.col1) self.assert_compile(sql_util.ClauseAdapter(s2).traverse(j).select(), - "SELECT anon_1.col1, anon_1.col2, anon_1.col3, table2.col1, table2.col2, table2.col3 FROM "\ - "(SELECT foo.col1 AS col1, foo.col2 AS col2, foo.col3 AS col3 FROM "\ - "(SELECT table1.col1 AS col1, table1.col2 AS col2, table1.col3 AS col3 FROM table1) AS foo LIMIT :param_1 OFFSET :param_2) AS anon_1 "\ - "LEFT OUTER JOIN table2 ON anon_1.col1 = table2.col1", - {'param_1':5, 'param_2':10} - ) - + 'SELECT anon_1.col1, anon_1.col2, ' + 'anon_1.col3, table2.col1, table2.col2, ' + 'table2.col3 FROM (SELECT foo.col1 AS ' + 'col1, foo.col2 AS col2, foo.col3 AS col3 ' + 'FROM (SELECT table1.col1 AS col1, ' + 'table1.col2 AS col2, table1.col3 AS col3 ' + 'FROM table1) AS foo LIMIT :param_1 OFFSET ' + ':param_2) AS anon_1 LEFT OUTER JOIN ' + 'table2 ON anon_1.col1 = table2.col1', + {'param_1': 5, 'param_2': 10}) talias = t1.alias('bar') - j = s1.outerjoin(talias, s1.c.col1==talias.c.col1) + j = s1.outerjoin(talias, s1.c.col1 == talias.c.col1) self.assert_compile(sql_util.ClauseAdapter(s2).traverse(j).select(), - "SELECT anon_1.col1, anon_1.col2, anon_1.col3, bar.col1, bar.col2, bar.col3 FROM "\ - "(SELECT foo.col1 AS col1, foo.col2 AS col2, foo.col3 AS col3 FROM "\ - "(SELECT table1.col1 AS col1, table1.col2 AS col2, table1.col3 AS col3 FROM table1) AS foo LIMIT :param_1 OFFSET :param_2) AS anon_1 "\ - "LEFT OUTER JOIN table1 AS bar ON anon_1.col1 = bar.col1", - {'param_1':5, 'param_2':10}) + 'SELECT anon_1.col1, anon_1.col2, ' + 'anon_1.col3, bar.col1, bar.col2, bar.col3 ' + 'FROM (SELECT foo.col1 AS col1, foo.col2 ' + 'AS col2, foo.col3 AS col3 FROM (SELECT ' + 'table1.col1 AS col1, table1.col2 AS col2, ' + 'table1.col3 AS col3 FROM table1) AS foo ' + 'LIMIT :param_1 OFFSET :param_2) AS anon_1 ' + 'LEFT OUTER JOIN table1 AS bar ON ' + 'anon_1.col1 = bar.col1', {'param_1': 5, + 'param_2': 10}) def test_functions(self): - self.assert_compile(sql_util.ClauseAdapter(t1.alias()).traverse(func.count(t1.c.col1)), "count(table1_1.col1)") - + self.assert_compile( + sql_util.ClauseAdapter(t1.alias()).\ + traverse(func.count(t1.c.col1)), + 'count(table1_1.col1)') s = select([func.count(t1.c.col1)]) - self.assert_compile(sql_util.ClauseAdapter(t1.alias()).traverse(s), "SELECT count(table1_1.col1) AS count_1 FROM table1 AS table1_1") + self.assert_compile(sql_util.ClauseAdapter(t1.alias()).traverse(s), + 'SELECT count(table1_1.col1) AS count_1 ' + 'FROM table1 AS table1_1') def test_recursive(self): metadata = MetaData() @@ -678,11 +847,13 @@ class ClauseAdapterTest(TestBase, AssertsCompiledSQL): ).alias() self.assert_compile( - sql_util.ClauseAdapter(u).traverse(select([c.c.bid]).where(c.c.bid==u.c.b_aid)), + sql_util.ClauseAdapter(u).\ + traverse(select([c.c.bid]).where(c.c.bid==u.c.b_aid)), "SELECT c.bid "\ - "FROM c, (SELECT a.id AS a_id, b.id AS b_id, b.aid AS b_aid "\ - "FROM a JOIN b ON a.id = b.aid UNION SELECT a.id AS a_id, d.id AS d_id, d.aid AS d_aid "\ - "FROM a JOIN d ON a.id = d.aid) AS anon_1 "\ + "FROM c, (SELECT a.id AS a_id, b.id AS b_id, b.aid AS b_aid " + "FROM a JOIN b ON a.id = b.aid UNION SELECT a.id AS a_id, d.id " + "AS d_id, d.aid AS d_aid " + "FROM a JOIN d ON a.id = d.aid) AS anon_1 " "WHERE c.bid = anon_1.b_aid" ) @@ -690,60 +861,69 @@ class SpliceJoinsTest(TestBase, AssertsCompiledSQL): @classmethod def setup_class(cls): global table1, table2, table3, table4 + def _table(name): - return table(name, column("col1"), column("col2"),column("col3")) + return table(name, column('col1'), column('col2'), + column('col3')) - table1, table2, table3, table4 = [_table(name) for name in ("table1", "table2", "table3", "table4")] + table1, table2, table3, table4 = [_table(name) for name in + ('table1', 'table2', 'table3', 'table4')] def test_splice(self): - (t1, t2, t3, t4) = (table1, table2, table1.alias(), table2.alias()) - - j = t1.join(t2, t1.c.col1==t2.c.col1).join(t3, t2.c.col1==t3.c.col1).join(t4, t4.c.col1==t1.c.col1) - - s = select([t1]).where(t1.c.col2<5).alias() - - self.assert_compile(sql_util.splice_joins(s, j), - "(SELECT table1.col1 AS col1, table1.col2 AS col2, "\ - "table1.col3 AS col3 FROM table1 WHERE table1.col2 < :col2_1) AS anon_1 "\ - "JOIN table2 ON anon_1.col1 = table2.col1 JOIN table1 AS table1_1 ON table2.col1 = table1_1.col1 "\ - "JOIN table2 AS table2_1 ON table2_1.col1 = anon_1.col1") + t1, t2, t3, t4 = table1, table2, table1.alias(), table2.alias() + j = t1.join(t2, t1.c.col1 == t2.c.col1).join(t3, t2.c.col1 + == t3.c.col1).join(t4, t4.c.col1 == t1.c.col1) + s = select([t1]).where(t1.c.col2 < 5).alias() + self.assert_compile(sql_util.splice_joins(s, j), + '(SELECT table1.col1 AS col1, table1.col2 ' + 'AS col2, table1.col3 AS col3 FROM table1 ' + 'WHERE table1.col2 < :col2_1) AS anon_1 ' + 'JOIN table2 ON anon_1.col1 = table2.col1 ' + 'JOIN table1 AS table1_1 ON table2.col1 = ' + 'table1_1.col1 JOIN table2 AS table2_1 ON ' + 'table2_1.col1 = anon_1.col1') def test_stop_on(self): - (t1, t2, t3) = (table1, table2, table3) - - j1= t1.join(t2, t1.c.col1==t2.c.col1) - j2 = j1.join(t3, t2.c.col1==t3.c.col1) - + t1, t2, t3 = table1, table2, table3 + j1 = t1.join(t2, t1.c.col1 == t2.c.col1) + j2 = j1.join(t3, t2.c.col1 == t3.c.col1) s = select([t1]).select_from(j1).alias() - - self.assert_compile(sql_util.splice_joins(s, j2), - "(SELECT table1.col1 AS col1, table1.col2 AS col2, table1.col3 AS col3 FROM table1 JOIN table2 "\ - "ON table1.col1 = table2.col1) AS anon_1 JOIN table2 ON anon_1.col1 = table2.col1 JOIN table3 "\ - "ON table2.col1 = table3.col1" - ) - - self.assert_compile(sql_util.splice_joins(s, j2, j1), - "(SELECT table1.col1 AS col1, table1.col2 AS col2, table1.col3 AS col3 FROM table1 "\ - "JOIN table2 ON table1.col1 = table2.col1) AS anon_1 JOIN table3 ON table2.col1 = table3.col1") + self.assert_compile(sql_util.splice_joins(s, j2), + '(SELECT table1.col1 AS col1, table1.col2 ' + 'AS col2, table1.col3 AS col3 FROM table1 ' + 'JOIN table2 ON table1.col1 = table2.col1) ' + 'AS anon_1 JOIN table2 ON anon_1.col1 = ' + 'table2.col1 JOIN table3 ON table2.col1 = ' + 'table3.col1') + self.assert_compile(sql_util.splice_joins(s, j2, j1), + '(SELECT table1.col1 AS col1, table1.col2 ' + 'AS col2, table1.col3 AS col3 FROM table1 ' + 'JOIN table2 ON table1.col1 = table2.col1) ' + 'AS anon_1 JOIN table3 ON table2.col1 = ' + 'table3.col1') def test_splice_2(self): t2a = table2.alias() t3a = table3.alias() - j1 = table1.join(t2a, table1.c.col1==t2a.c.col1).join(t3a, t2a.c.col2==t3a.c.col2) - + j1 = table1.join(t2a, table1.c.col1 == t2a.c.col1).join(t3a, + t2a.c.col2 == t3a.c.col2) t2b = table4.alias() - j2 = table1.join(t2b, table1.c.col3==t2b.c.col3) - - self.assert_compile(sql_util.splice_joins(table1, j1), - "table1 JOIN table2 AS table2_1 ON table1.col1 = table2_1.col1 "\ - "JOIN table3 AS table3_1 ON table2_1.col2 = table3_1.col2") - - self.assert_compile(sql_util.splice_joins(table1, j2), "table1 JOIN table4 AS table4_1 ON table1.col3 = table4_1.col3") - - self.assert_compile(sql_util.splice_joins(sql_util.splice_joins(table1, j1), j2), - "table1 JOIN table2 AS table2_1 ON table1.col1 = table2_1.col1 "\ - "JOIN table3 AS table3_1 ON table2_1.col2 = table3_1.col2 "\ - "JOIN table4 AS table4_1 ON table1.col3 = table4_1.col3") + j2 = table1.join(t2b, table1.c.col3 == t2b.c.col3) + self.assert_compile(sql_util.splice_joins(table1, j1), + 'table1 JOIN table2 AS table2_1 ON ' + 'table1.col1 = table2_1.col1 JOIN table3 ' + 'AS table3_1 ON table2_1.col2 = ' + 'table3_1.col2') + self.assert_compile(sql_util.splice_joins(table1, j2), + 'table1 JOIN table4 AS table4_1 ON ' + 'table1.col3 = table4_1.col3') + self.assert_compile(sql_util.splice_joins(sql_util.splice_joins(table1, + j1), j2), + 'table1 JOIN table2 AS table2_1 ON ' + 'table1.col1 = table2_1.col1 JOIN table3 ' + 'AS table3_1 ON table2_1.col2 = ' + 'table3_1.col2 JOIN table4 AS table4_1 ON ' + 'table1.col3 = table4_1.col3') class SelectTest(TestBase, AssertsCompiledSQL): @@ -764,66 +944,117 @@ class SelectTest(TestBase, AssertsCompiledSQL): ) 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) + 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") + '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() - self.assert_compile(s, "SELECT table1.col1, table1.col2, table1.col3 FROM table1") + self.assert_compile(s, + 'SELECT table1.col1, table1.col2, ' + 'table1.col3 FROM table1') select_copy = s.column('yyy') - self.assert_compile(select_copy, "SELECT table1.col1, table1.col2, table1.col3, yyy FROM table1") + self.assert_compile(select_copy, + 'SELECT table1.col1, table1.col2, ' + 'table1.col3, yyy FROM table1') assert s.columns is not select_copy.columns assert s._columns is not select_copy._columns assert s._raw_columns is not select_copy._raw_columns - self.assert_compile(s, "SELECT table1.col1, table1.col2, table1.col3 FROM table1") + self.assert_compile(s, + 'SELECT table1.col1, table1.col2, ' + 'table1.col3 FROM table1') def test_froms(self): s = t1.select() - self.assert_compile(s, "SELECT table1.col1, table1.col2, table1.col3 FROM table1") + self.assert_compile(s, + 'SELECT table1.col1, table1.col2, ' + 'table1.col3 FROM table1') select_copy = s.select_from(t2) - self.assert_compile(select_copy, "SELECT table1.col1, table1.col2, table1.col3 FROM table1, table2") + self.assert_compile(select_copy, + 'SELECT table1.col1, table1.col2, ' + 'table1.col3 FROM table1, table2') assert s._froms is not select_copy._froms - self.assert_compile(s, "SELECT table1.col1, table1.col2, table1.col3 FROM table1") + self.assert_compile(s, + '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) - 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 table1.col1 = table2.col1) WHERE table1.col2 = col2") - + 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) + 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 ' + '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") - 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 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(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 ' + 'table1.col1 = table2.col1) WHERE ' + 'table1.col2 = col2') 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(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') def test_prefixes(self): s = t1.select() - self.assert_compile(s, "SELECT table1.col1, table1.col2, table1.col3 FROM table1") - select_copy = s.prefix_with("FOOBER") - self.assert_compile(select_copy, "SELECT FOOBER table1.col1, table1.col2, table1.col3 FROM table1") - self.assert_compile(s, "SELECT table1.col1, table1.col2, table1.col3 FROM table1") + self.assert_compile(s, + 'SELECT table1.col1, table1.col2, ' + 'table1.col3 FROM table1') + select_copy = s.prefix_with('FOOBER') + self.assert_compile(select_copy, + 'SELECT FOOBER table1.col1, table1.col2, ' + 'table1.col3 FROM table1') + self.assert_compile(s, + 'SELECT table1.col1, table1.col2, ' + 'table1.col3 FROM table1') def test_execution_options(self): s = select().execution_options(foo='bar')