summaryrefslogtreecommitdiff
path: root/test/dialect/postgresql/test_compiler.py
diff options
context:
space:
mode:
authormike bayer <mike_mp@zzzcomputing.com>2021-07-12 21:52:09 +0000
committerGerrit Code Review <gerrit@ci3.zzzcomputing.com>2021-07-12 21:52:09 +0000
commitb7791290ed5499542d2e6d7bdfac252f7648f5fc (patch)
tree186e0ab28475565559a5f3cb3613d875d544d4a5 /test/dialect/postgresql/test_compiler.py
parent71fba6191d6b1858b443b5d8846d89dc1506c3a6 (diff)
parent4ff4760fade8020df0418005e0fdd130ed3589a4 (diff)
downloadsqlalchemy-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.py154
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 "