summaryrefslogtreecommitdiff
path: root/test/dialect/postgresql/test_compiler.py
diff options
context:
space:
mode:
authorRobotScribe <quentinso@theodo.fr>2020-04-29 15:22:59 -0400
committerGord Thompson <gord@gordthompson.com>2020-05-15 15:50:32 -0600
commit103260ddb476c5354b3201f92636c474f2a83c35 (patch)
treefbec060d33e0a6c3770c9d5ce7818432497ad6ef /test/dialect/postgresql/test_compiler.py
parent9821bddfcb3c94cea13b7f19bcb27845b0dc1ed8 (diff)
downloadsqlalchemy-103260ddb476c5354b3201f92636c474f2a83c35.tar.gz
Add with_for_update mysql new functionalities
Fixes: #4860 # Description Add nowait, skip_lock, of arguments to for_update_clause for mysql ### Checklist This pull request is: - [ ] A documentation / typographical error fix - Good to go, no issue or tests are needed - [ ] A short code fix - please include the issue number, and create an issue if none exists, which must include a complete example of the issue. one line code fixes without an issue and demonstration will not be accepted. - Please include: `Fixes: #<issue number>` in the commit message - please include tests. one line code fixes without tests will not be accepted. - [x] A new feature implementation - please include the issue number, and create an issue if none exists, which must include a complete example of how the feature would look. - Please include: `Fixes: #<issue number>` in the commit message - please include tests. **Have a nice day!** Closes: #5290 Pull-request: https://github.com/sqlalchemy/sqlalchemy/pull/5290 Pull-request-sha: 490e822e73e92ffe63cf45df9c49f3b31af1954d Change-Id: Ibd2acc47b538c601c69c8fb954776035ecab4c6c
Diffstat (limited to 'test/dialect/postgresql/test_compiler.py')
-rw-r--r--test/dialect/postgresql/test_compiler.py66
1 files changed, 66 insertions, 0 deletions
diff --git a/test/dialect/postgresql/test_compiler.py b/test/dialect/postgresql/test_compiler.py
index 4cc9c837d..c707137a8 100644
--- a/test/dialect/postgresql/test_compiler.py
+++ b/test/dialect/postgresql/test_compiler.py
@@ -952,6 +952,24 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL):
self.assert_compile(
table1.select(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
+ ),
+ "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
),
"SELECT mytable.myid, mytable.name, mytable.description "
@@ -979,6 +997,15 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL):
self.assert_compile(
table1.select(table1.c.myid == 7).with_for_update(
+ key_share=True, read=True, nowait=True, of=table1
+ ),
+ "SELECT mytable.myid, mytable.name, mytable.description "
+ "FROM mytable WHERE mytable.myid = %(myid_1)s "
+ "FOR KEY SHARE OF mytable NOWAIT",
+ )
+
+ self.assert_compile(
+ table1.select(table1.c.myid == 7).with_for_update(
read=True, nowait=True, of=table1.c.myid
),
"SELECT mytable.myid, mytable.name, mytable.description "
@@ -997,6 +1024,27 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL):
self.assert_compile(
table1.select(table1.c.myid == 7).with_for_update(
+ read=True,
+ skip_locked=True,
+ of=[table1.c.myid, table1.c.name],
+ key_share=True,
+ ),
+ "SELECT mytable.myid, mytable.name, mytable.description "
+ "FROM mytable WHERE mytable.myid = %(myid_1)s "
+ "FOR KEY SHARE OF mytable SKIP LOCKED",
+ )
+
+ self.assert_compile(
+ table1.select(table1.c.myid == 7).with_for_update(
+ skip_locked=True, of=[table1.c.myid, table1.c.name]
+ ),
+ "SELECT mytable.myid, mytable.name, mytable.description "
+ "FROM mytable WHERE mytable.myid = %(myid_1)s "
+ "FOR UPDATE OF mytable SKIP LOCKED",
+ )
+
+ self.assert_compile(
+ table1.select(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 "
@@ -1060,6 +1108,15 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL):
self.assert_compile(
table1.select(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
),
"SELECT mytable.myid, mytable.name, mytable.description "
@@ -1067,6 +1124,15 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL):
"FOR KEY SHARE SKIP LOCKED",
)
+ self.assert_compile(
+ table1.select(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",
+ )
+
ta = table1.alias()
self.assert_compile(
ta.select(ta.c.myid == 7).with_for_update(