Add redis/redis-py@42a050c

Support for SCRIPT FLUSH with SYNC/ASYNC (redis/redis-py#1567)

Signed-off-by: Andrew-Chen-Wang <[email protected]>

Author: Andrew-Chen-Wang

aioredis/commands.py

@@ -3323,9 +3323,20 @@ def script_exists(self: _SELF_ANNOTATION, *args: str) -> Awaitable:
         """
         return self.execute_command("SCRIPT EXISTS", *args)
 
-    def script_flush(self: _SELF_ANNOTATION) -> Awaitable:
-        """Flush all scripts from the script cache"""
-        return self.execute_command("SCRIPT FLUSH")
+    def script_flush(
+        self: _SELF_ANNOTATION,
+        sync_type: Union[Literal["SYNC"], Literal["ASYNC"]] = "SYNC",
+    ) -> Awaitable:
+        """Flush all scripts from the script cache.
+        ``sync_type`` is by default SYNC (synchronous) but it can also be
+                      ASYNC.
+        See: https://redis.io/commands/script-flush
+        """
+        if sync_type not in ["SYNC", "ASYNC"]:
+            raise DataError("SCRIPT FLUSH defaults to SYNC or "
+                            "accepts SYNC/ASYNC")
+        pieces = [sync_type]
+        return self.execute_command('SCRIPT FLUSH', *pieces)
 
     def script_kill(self: _SELF_ANNOTATION) -> Awaitable:
         """Kill the currently executing Lua script"""

tests/test_scripting.py

@@ -33,6 +33,25 @@ async def test_eval(self, r):
         # 2 * 3 == 6
         assert await r.eval(multiply_script, 1, "a", 3) == 6
 
+    @pytest.mark.asyncio(forbid_global_loop=True)
+    async def test_script_flush(self, r):
+        await r.set('a', 2)
+        await r.script_load(multiply_script)
+        await r.script_flush('ASYNC')
+
+        await r.set('a', 2)
+        await r.script_load(multiply_script)
+        await r.script_flush('SYNC')
+
+        await r.set('a', 2)
+        await r.script_load(multiply_script)
+        await r.script_flush()
+
+        with pytest.raises(exceptions.DataError):
+            await r.set('a', 2)
+            await r.script_load(multiply_script)
+            await r.script_flush("NOTREAL")
+
     @pytest.mark.asyncio(forbid_global_loop=True)
     async def test_evalsha(self, r):
         await r.set("a", 2)