Python: add WATCH and UNWATCH commands

python/python/glide/async_commands/cluster_commands.py

@@ -774,3 +774,26 @@ async def wait(
             int,
             await self._execute_command(RequestType.Wait, args, route),
         )
+
+    async def unwatch(self, route: Optional[Route] = None) -> TOK:
+        """
+        Flushes all the previously watched keys for a transaction. Executing a transaction will
+        automatically flush all previously watched keys.
+
+        See https://valkey.io/commands/unwatch for more details.
+
+        Args:
+            route (Optional[Route]): The command will be routed to all primary nodes, unless `route` is provided,
+                in which case the client will route the command to the nodes defined by `route`.
+
+        Returns:
+            TOK: A simple "OK" response.
+
+        Examples:
+            >>> await client.unwatch()
+                'OK'
+        """
+        return cast(
+            TOK,
+            await self._execute_command(RequestType.UnWatch, [], route),
+        )

python/python/glide/async_commands/core.py

@@ -5440,6 +5440,41 @@ async def sscan(
             await self._execute_command(RequestType.SScan, args),
         )
 
+    async def watch(self, keys: List[str]) -> TOK:
+        """
+        Marks the given keys to be watched for conditional execution of a transaction. Transactions
+        will only execute commands if the watched keys are not modified before execution of the
+        transaction.
+
+        See https://valkey.io/commands/watch for more details.
+
+        Args:
+            keys (List[str]): The keys to watch.
+
+        Returns:
+            TOK: A simple "OK" response.
+
+        Examples:
+            >>> await client.watch("sampleKey")
+                'OK'
+            >>> transaction.set("sampleKey", "foobar")
+            >>> await redis_client.exec(transaction)
+                'OK' # Executes successfully and keys are unwatched.
+
+            >>> await client.watch("sampleKey")
+                'OK'
+            >>> transaction.set("sampleKey", "foobar")
+            >>> await client.set("sampleKey", "hello world")
+                'OK'
+            >>> await redis_client.exec(transaction)
+                None  # None is returned when the watched key is modified before transaction execution.
+        """
+
+        return cast(
+            TOK,
+            await self._execute_command(RequestType.Watch, keys),
+        )
+
     @dataclass
     class PubSubMsg:
         """

python/python/glide/async_commands/standalone_commands.py

@@ -695,3 +695,24 @@ async def wait(
             int,
             await self._execute_command(RequestType.Wait, args),
         )
+
+    async def unwatch(self) -> TOK:
+        """
+        Flushes all the previously watched keys for a transaction. Executing a transaction will
+        automatically flush all previously watched keys.
+
+        See https://valkey.io/commands/unwatch for more details.
+
+        Returns:
+            TOK: A simple "OK" response.
+
+        Examples:
+            >>> await client.watch("sampleKey")
+                'OK'
+            >>> await client.unwatch()
+                'OK'
+        """
+        return cast(
+            TOK,
+            await self._execute_command(RequestType.UnWatch, []),
+        )

python/python/tests/test_async_client.py

@@ -68,6 +68,7 @@
     TrimByMaxLen,
     TrimByMinId,
 )
+from glide.async_commands.transaction import Transaction
 from glide.config import (
     ClusterClientConfiguration,
     GlideClientConfiguration,
@@ -7072,18 +7073,18 @@ async def test_flushall(self, redis_client: TGlideClient):
 
         await redis_client.set(key, value)
         assert await redis_client.dbsize() > 0
-        assert await redis_client.flushall() is OK
-        assert await redis_client.flushall(FlushMode.ASYNC) is OK
+        assert await redis_client.flushall() == OK
+        assert await redis_client.flushall(FlushMode.ASYNC) == OK
         if not await check_if_server_version_lt(redis_client, min_version):
-            assert await redis_client.flushall(FlushMode.SYNC) is OK
+            assert await redis_client.flushall(FlushMode.SYNC) == OK
         assert await redis_client.dbsize() == 0
 
         if isinstance(redis_client, GlideClusterClient):
             await redis_client.set(key, value)
-            assert await redis_client.flushall(route=AllPrimaries()) is OK
-            assert await redis_client.flushall(FlushMode.ASYNC, AllPrimaries()) is OK
+            assert await redis_client.flushall(route=AllPrimaries()) == OK
+            assert await redis_client.flushall(FlushMode.ASYNC, AllPrimaries()) == OK
             if not await check_if_server_version_lt(redis_client, min_version):
-                assert await redis_client.flushall(FlushMode.SYNC, AllPrimaries()) is OK
+                assert await redis_client.flushall(FlushMode.SYNC, AllPrimaries()) == OK
             assert await redis_client.dbsize() == 0
 
     @pytest.mark.parametrize("cluster_mode", [False])
@@ -7095,30 +7096,30 @@ async def test_standalone_flushdb(self, redis_client: GlideClient):
         value = get_random_string(5)
 
         # fill DB 0 and check size non-empty
-        assert await redis_client.select(0) is OK
+        assert await redis_client.select(0) == OK
         await redis_client.set(key1, value)
         assert await redis_client.dbsize() > 0
 
         # fill DB 1 and check size non-empty
-        assert await redis_client.select(1) is OK
+        assert await redis_client.select(1) == OK
         await redis_client.set(key2, value)
         assert await redis_client.dbsize() > 0
 
         # flush DB 1 and check again
-        assert await redis_client.flushdb() is OK
+        assert await redis_client.flushdb() == OK
         assert await redis_client.dbsize() == 0
 
         # swith to DB 0, flush, and check
-        assert await redis_client.select(0) is OK
+        assert await redis_client.select(0) == OK
         assert await redis_client.dbsize() > 0
-        assert await redis_client.flushdb(FlushMode.ASYNC) is OK
+        assert await redis_client.flushdb(FlushMode.ASYNC) == OK
         assert await redis_client.dbsize() == 0
 
         # verify flush SYNC
         if not await check_if_server_version_lt(redis_client, min_version):
             await redis_client.set(key2, value)
             assert await redis_client.dbsize() > 0
-            assert await redis_client.flushdb(FlushMode.SYNC) is OK
+            assert await redis_client.flushdb(FlushMode.SYNC) == OK
             assert await redis_client.dbsize() == 0
 
     @pytest.mark.parametrize("cluster_mode", [True, False])
@@ -7503,6 +7504,55 @@ async def test_lcs_idx(self, redis_client: GlideClient):
         with pytest.raises(RequestError):
             await redis_client.lcs_idx(key1, lcs_non_string_key)
 
+    @pytest.mark.parametrize("cluster_mode", [False])
+    @pytest.mark.parametrize("protocol", [ProtocolVersion.RESP2, ProtocolVersion.RESP3])
+    async def test_watch(self, redis_client: GlideClient):
+        # watched key didn't change outside of transaction before transaction execution, transaction will execute
+        assert await redis_client.set("key1", "original_value") == OK
+        assert await redis_client.watch(["key1"]) == OK
+        transaction = Transaction()
+        transaction.set("key1", "transaction_value")
+        transaction.get("key1")
+        assert await redis_client.exec(transaction) is not None
+
+        # watched key changed outside of transaction before transaction execution, transaction will not execute
+        assert await redis_client.set("key1", "original_value") == OK
+        assert await redis_client.watch(["key1"]) == OK
+        transaction = Transaction()
+        transaction.set("key1", "transaction_value")
+        assert await redis_client.set("key1", "standalone_value") == OK
+        transaction.get("key1")
+        assert await redis_client.exec(transaction) is None
+
+        # empty list not supported
+        with pytest.raises(RequestError):
+            await redis_client.watch([])
+
+    @pytest.mark.parametrize("cluster_mode", [True, False])
+    @pytest.mark.parametrize("protocol", [ProtocolVersion.RESP2, ProtocolVersion.RESP3])
+    async def test_unwatch(self, redis_client: GlideClient):
+
+        # watched key unwatched before transaction execution even if changed
+        # outside of transaction, transaction will still execute
+        assert await redis_client.set("key1", "original_value") == OK
+        assert await redis_client.watch(["key1"]) == OK
+        transaction = Transaction()
+        transaction.set("key1", "transaction_value")
+        assert await redis_client.set("key1", "standalone_value") == OK
+        transaction.get("key1")
+        assert await redis_client.unwatch() == OK
+        result = await redis_client.exec(transaction)
+        assert result is not None
+        assert isinstance(result, list)
+        assert len(result) == 2
+        assert result[0] == "OK"
+        assert result[1] == b"transaction_value"
+
+    @pytest.mark.parametrize("cluster_mode", [True])
+    @pytest.mark.parametrize("protocol", [ProtocolVersion.RESP2, ProtocolVersion.RESP3])
+    async def test_unwatch_with_route(self, redis_client: GlideClient):
+        assert await redis_client.unwatch(RandomNode()) == OK
+
 
 class TestMultiKeyCommandCrossSlot:
     @pytest.mark.parametrize("cluster_mode", [True])
@@ -7867,18 +7917,18 @@ async def test_cluster_flushdb(self, redis_client: GlideClusterClient):
 
         await redis_client.set(key, value)
         assert await redis_client.dbsize() > 0
-        assert await redis_client.flushdb(route=AllPrimaries()) is OK
+        assert await redis_client.flushdb(route=AllPrimaries()) == OK
         assert await redis_client.dbsize() == 0
 
         await redis_client.set(key, value)
         assert await redis_client.dbsize() > 0
-        assert await redis_client.flushdb(FlushMode.ASYNC, AllPrimaries()) is OK
+        assert await redis_client.flushdb(FlushMode.ASYNC, AllPrimaries()) == OK
         assert await redis_client.dbsize() == 0
 
         if not await check_if_server_version_lt(redis_client, min_version):
             await redis_client.set(key, value)
             assert await redis_client.dbsize() > 0
-            assert await redis_client.flushdb(FlushMode.SYNC, AllPrimaries()) is OK
+            assert await redis_client.flushdb(FlushMode.SYNC, AllPrimaries()) == OK
             assert await redis_client.dbsize() == 0
 
     @pytest.mark.parametrize("cluster_mode", [True, False])

python/python/tests/test_transaction.py

@@ -749,7 +749,7 @@ async def test_can_return_null_on_watch_transaction_failures(
         keyslot = get_random_string(3)
         transaction = ClusterTransaction() if is_cluster else Transaction()
         transaction.get(keyslot)
-        result1 = await redis_client.custom_command(["WATCH", keyslot])
+        result1 = await redis_client.watch([keyslot])
         assert result1 == OK
 
         result2 = await client2.set(keyslot, "foo")