diff options
author | mike bayer <mike_mp@zzzcomputing.com> | 2021-07-12 21:52:09 +0000 |
---|---|---|
committer | Gerrit Code Review <gerrit@ci3.zzzcomputing.com> | 2021-07-12 21:52:09 +0000 |
commit | b7791290ed5499542d2e6d7bdfac252f7648f5fc (patch) | |
tree | 186e0ab28475565559a5f3cb3613d875d544d4a5 /test/dialect/postgresql/test_compiler.py | |
parent | 71fba6191d6b1858b443b5d8846d89dc1506c3a6 (diff) | |
parent | 4ff4760fade8020df0418005e0fdd130ed3589a4 (diff) | |
download | sqlalchemy-b7791290ed5499542d2e6d7bdfac252f7648f5fc.tar.gz |
Merge "Modernize tests - select(whereclause)"
Diffstat (limited to 'test/dialect/postgresql/test_compiler.py')
-rw-r--r-- | test/dialect/postgresql/test_compiler.py | 154 |
1 files changed, 91 insertions, 63 deletions
diff --git a/test/dialect/postgresql/test_compiler.py b/test/dialect/postgresql/test_compiler.py index 3b542880f..84901d178 100644 --- a/test/dialect/postgresql/test_compiler.py +++ b/test/dialect/postgresql/test_compiler.py @@ -1082,87 +1082,93 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): ) self.assert_compile( - table1.select(table1.c.myid == 7).with_for_update(), + table1.select().where(table1.c.myid == 7).with_for_update(), "SELECT mytable.myid, mytable.name, mytable.description " "FROM mytable WHERE mytable.myid = %(myid_1)s FOR UPDATE", ) self.assert_compile( - table1.select(table1.c.myid == 7).with_for_update(nowait=True), + table1.select() + .where(table1.c.myid == 7) + .with_for_update(nowait=True), "SELECT mytable.myid, mytable.name, mytable.description " "FROM mytable WHERE mytable.myid = %(myid_1)s FOR UPDATE NOWAIT", ) self.assert_compile( - table1.select(table1.c.myid == 7).with_for_update( - skip_locked=True - ), + table1.select() + .where(table1.c.myid == 7) + .with_for_update(skip_locked=True), "SELECT mytable.myid, mytable.name, mytable.description " "FROM mytable WHERE mytable.myid = %(myid_1)s " "FOR UPDATE SKIP LOCKED", ) self.assert_compile( - table1.select(table1.c.myid == 7).with_for_update(read=True), + table1.select() + .where(table1.c.myid == 7) + .with_for_update(read=True), "SELECT mytable.myid, mytable.name, mytable.description " "FROM mytable WHERE mytable.myid = %(myid_1)s FOR SHARE", ) self.assert_compile( - table1.select(table1.c.myid == 7).with_for_update( - read=True, nowait=True - ), + table1.select() + .where(table1.c.myid == 7) + .with_for_update(read=True, nowait=True), "SELECT mytable.myid, mytable.name, mytable.description " "FROM mytable WHERE mytable.myid = %(myid_1)s FOR SHARE NOWAIT", ) self.assert_compile( - table1.select(table1.c.myid == 7).with_for_update( - key_share=True, nowait=True - ), + table1.select() + .where(table1.c.myid == 7) + .with_for_update(key_share=True, nowait=True), "SELECT mytable.myid, mytable.name, mytable.description " "FROM mytable WHERE mytable.myid = %(myid_1)s " "FOR NO KEY UPDATE NOWAIT", ) self.assert_compile( - table1.select(table1.c.myid == 7).with_for_update( - key_share=True, read=True, nowait=True - ), + table1.select() + .where(table1.c.myid == 7) + .with_for_update(key_share=True, read=True, nowait=True), "SELECT mytable.myid, mytable.name, mytable.description " "FROM mytable WHERE mytable.myid = %(myid_1)s " "FOR KEY SHARE NOWAIT", ) self.assert_compile( - table1.select(table1.c.myid == 7).with_for_update( - read=True, skip_locked=True - ), + table1.select() + .where(table1.c.myid == 7) + .with_for_update(read=True, skip_locked=True), "SELECT mytable.myid, mytable.name, mytable.description " "FROM mytable WHERE mytable.myid = %(myid_1)s " "FOR SHARE SKIP LOCKED", ) self.assert_compile( - table1.select(table1.c.myid == 7).with_for_update( - of=table1.c.myid - ), + table1.select() + .where(table1.c.myid == 7) + .with_for_update(of=table1.c.myid), "SELECT mytable.myid, mytable.name, mytable.description " "FROM mytable WHERE mytable.myid = %(myid_1)s " "FOR UPDATE OF mytable", ) self.assert_compile( - table1.select(table1.c.myid == 7).with_for_update( - read=True, nowait=True, of=table1 - ), + table1.select() + .where(table1.c.myid == 7) + .with_for_update(read=True, nowait=True, of=table1), "SELECT mytable.myid, mytable.name, mytable.description " "FROM mytable WHERE mytable.myid = %(myid_1)s " "FOR SHARE OF mytable NOWAIT", ) self.assert_compile( - table1.select(table1.c.myid == 7).with_for_update( + table1.select() + .where(table1.c.myid == 7) + .with_for_update( key_share=True, read=True, nowait=True, of=table1 ), "SELECT mytable.myid, mytable.name, mytable.description " @@ -1171,16 +1177,18 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): ) self.assert_compile( - table1.select(table1.c.myid == 7).with_for_update( - read=True, nowait=True, of=table1.c.myid - ), + table1.select() + .where(table1.c.myid == 7) + .with_for_update(read=True, nowait=True, of=table1.c.myid), "SELECT mytable.myid, mytable.name, mytable.description " "FROM mytable WHERE mytable.myid = %(myid_1)s " "FOR SHARE OF mytable NOWAIT", ) self.assert_compile( - table1.select(table1.c.myid == 7).with_for_update( + table1.select() + .where(table1.c.myid == 7) + .with_for_update( read=True, nowait=True, of=[table1.c.myid, table1.c.name] ), "SELECT mytable.myid, mytable.name, mytable.description " @@ -1189,7 +1197,9 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): ) self.assert_compile( - table1.select(table1.c.myid == 7).with_for_update( + table1.select() + .where(table1.c.myid == 7) + .with_for_update( read=True, skip_locked=True, of=[table1.c.myid, table1.c.name], @@ -1201,7 +1211,9 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): ) self.assert_compile( - table1.select(table1.c.myid == 7).with_for_update( + table1.select() + .where(table1.c.myid == 7) + .with_for_update( skip_locked=True, of=[table1.c.myid, table1.c.name] ), "SELECT mytable.myid, mytable.name, mytable.description " @@ -1210,7 +1222,9 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): ) self.assert_compile( - table1.select(table1.c.myid == 7).with_for_update( + table1.select() + .where(table1.c.myid == 7) + .with_for_update( read=True, skip_locked=True, of=[table1.c.myid, table1.c.name] ), "SELECT mytable.myid, mytable.name, mytable.description " @@ -1219,7 +1233,9 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): ) self.assert_compile( - table1.select(table1.c.myid == 7).with_for_update( + table1.select() + .where(table1.c.myid == 7) + .with_for_update( key_share=True, nowait=True, of=[table1.c.myid, table1.c.name] ), "SELECT mytable.myid, mytable.name, mytable.description " @@ -1228,7 +1244,9 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): ) self.assert_compile( - table1.select(table1.c.myid == 7).with_for_update( + table1.select() + .where(table1.c.myid == 7) + .with_for_update( key_share=True, skip_locked=True, of=[table1.c.myid, table1.c.name], @@ -1239,7 +1257,9 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): ) self.assert_compile( - table1.select(table1.c.myid == 7).with_for_update( + table1.select() + .where(table1.c.myid == 7) + .with_for_update( key_share=True, of=[table1.c.myid, table1.c.name] ), "SELECT mytable.myid, mytable.name, mytable.description " @@ -1248,52 +1268,54 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): ) self.assert_compile( - table1.select(table1.c.myid == 7).with_for_update(key_share=True), + table1.select() + .where(table1.c.myid == 7) + .with_for_update(key_share=True), "SELECT mytable.myid, mytable.name, mytable.description " "FROM mytable WHERE mytable.myid = %(myid_1)s " "FOR NO KEY UPDATE", ) self.assert_compile( - table1.select(table1.c.myid == 7).with_for_update( - read=True, key_share=True - ), + table1.select() + .where(table1.c.myid == 7) + .with_for_update(read=True, key_share=True), "SELECT mytable.myid, mytable.name, mytable.description " "FROM mytable WHERE mytable.myid = %(myid_1)s " "FOR KEY SHARE", ) self.assert_compile( - table1.select(table1.c.myid == 7).with_for_update( - read=True, key_share=True, of=table1 - ), + table1.select() + .where(table1.c.myid == 7) + .with_for_update(read=True, key_share=True, of=table1), "SELECT mytable.myid, mytable.name, mytable.description " "FROM mytable WHERE mytable.myid = %(myid_1)s " "FOR KEY SHARE OF mytable", ) self.assert_compile( - table1.select(table1.c.myid == 7).with_for_update( - read=True, of=table1 - ), + table1.select() + .where(table1.c.myid == 7) + .with_for_update(read=True, of=table1), "SELECT mytable.myid, mytable.name, mytable.description " "FROM mytable WHERE mytable.myid = %(myid_1)s " "FOR SHARE OF mytable", ) self.assert_compile( - table1.select(table1.c.myid == 7).with_for_update( - read=True, key_share=True, skip_locked=True - ), + table1.select() + .where(table1.c.myid == 7) + .with_for_update(read=True, key_share=True, skip_locked=True), "SELECT mytable.myid, mytable.name, mytable.description " "FROM mytable WHERE mytable.myid = %(myid_1)s " "FOR KEY SHARE SKIP LOCKED", ) self.assert_compile( - table1.select(table1.c.myid == 7).with_for_update( - key_share=True, skip_locked=True - ), + table1.select() + .where(table1.c.myid == 7) + .with_for_update(key_share=True, skip_locked=True), "SELECT mytable.myid, mytable.name, mytable.description " "FROM mytable WHERE mytable.myid = %(myid_1)s " "FOR NO KEY UPDATE SKIP LOCKED", @@ -1301,9 +1323,9 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): ta = table1.alias() self.assert_compile( - ta.select(ta.c.myid == 7).with_for_update( - of=[ta.c.myid, ta.c.name] - ), + ta.select() + .where(ta.c.myid == 7) + .with_for_update(of=[ta.c.myid, ta.c.name]), "SELECT mytable_1.myid, mytable_1.name, mytable_1.description " "FROM mytable AS mytable_1 " "WHERE mytable_1.myid = %(myid_1)s FOR UPDATE OF mytable_1", @@ -1312,7 +1334,9 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): table2 = table("table2", column("mytable_id")) join = table2.join(table1, table2.c.mytable_id == table1.c.myid) self.assert_compile( - join.select(table2.c.mytable_id == 7).with_for_update(of=[join]), + join.select() + .where(table2.c.mytable_id == 7) + .with_for_update(of=[join]), "SELECT table2.mytable_id, " "mytable.myid, mytable.name, mytable.description " "FROM table2 " @@ -1323,7 +1347,9 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): join = table2.join(ta, table2.c.mytable_id == ta.c.myid) self.assert_compile( - join.select(table2.c.mytable_id == 7).with_for_update(of=[join]), + join.select() + .where(table2.c.mytable_id == 7) + .with_for_update(of=[join]), "SELECT table2.mytable_id, " "mytable_1.myid, mytable_1.name, mytable_1.description " "FROM table2 " @@ -1335,9 +1361,9 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): # ensure of=text() for of works self.assert_compile( - table1.select(table1.c.myid == 7).with_for_update( - of=text("table1") - ), + table1.select() + .where(table1.c.myid == 7) + .with_for_update(of=text("table1")), "SELECT mytable.myid, mytable.name, mytable.description " "FROM mytable WHERE mytable.myid = %(myid_1)s " "FOR UPDATE OF table1", @@ -1345,9 +1371,9 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): # ensure literal_column of works self.assert_compile( - table1.select(table1.c.myid == 7).with_for_update( - of=literal_column("table1") - ), + table1.select() + .where(table1.c.myid == 7) + .with_for_update(of=literal_column("table1")), "SELECT mytable.myid, mytable.name, mytable.description " "FROM mytable WHERE mytable.myid = %(myid_1)s " "FOR UPDATE OF table1", @@ -1360,7 +1386,9 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): ) self.assert_compile( - table1.select(table1.c.myid == 7).with_for_update(of=table1), + table1.select() + .where(table1.c.myid == 7) + .with_for_update(of=table1), "SELECT testschema.mytable.myid, testschema.mytable.name " "FROM testschema.mytable " "WHERE testschema.mytable.myid = %(myid_1)s " |