Add support for COPY command new in Redis 6.2

redis/client.py

@@ -561,7 +561,7 @@ class Redis:
     """
     RESPONSE_CALLBACKS = {
         **string_keys_to_dict(
-            'AUTH EXPIRE EXPIREAT HEXISTS HMSET MOVE MSETNX PERSIST '
+            'AUTH COPY EXPIRE EXPIREAT HEXISTS HMSET MOVE MSETNX PERSIST '
             'PSETEX RENAMENX SISMEMBER SMOVE SETEX SETNX',
             bool
         ),
@@ -1612,6 +1612,24 @@ def bitpos(self, key, bit, start=None, end=None):
                             "when end is specified")
         return self.execute_command('BITPOS', *params)
 
+    def copy(self, source, destination, destination_db=None, replace=False):
+        """
+        Copy the value stored in the ``source`` key to the ``destination`` key.
+
+        ``destination_db`` an alternative destination database. By default,
+        the ``destination`` key is created in the source Redis database.
+
+        ``replace`` whether the ``destination`` key should be removed before
+        copying the value to it. By default, the value is not copied if
+        the ``destination`` key already exists.
+        """
+        params = [source, destination]
+        if destination_db is not None:
+            params.extend(["DB", destination_db])
+        if replace:
+            params.append("REPLACE")
+        return self.execute_command('COPY', *params)
+
     def decr(self, name, amount=1):
         """
         Decrements the value of ``key`` by ``amount``.  If no key exists,

tests/test_commands.py

@@ -578,6 +578,29 @@ def test_bitpos_wrong_arguments(self, r):
         with pytest.raises(exceptions.RedisError):
             r.bitpos(key, 7) == 12
 
+    @skip_if_server_version_lt('6.2.0')
+    def test_copy(self, r):
+        assert r.copy("a", "b") == 0
+        r.set("a", "foo")
+        assert r.copy("a", "b") == 1
+        assert r.get("a") == b"foo"
+        assert r.get("b") == b"foo"
+
+    @skip_if_server_version_lt('6.2.0')
+    def test_copy_and_replace(self, r):
+        r.set("a", "foo1")
+        r.set("b", "foo2")
+        assert r.copy("a", "b") == 0
+        assert r.copy("a", "b", replace=True) == 1
+
+    @skip_if_server_version_lt('6.2.0')
+    def test_copy_to_another_database(self, request):
+        r0 = _get_client(redis.Redis, request, db=0)
+        r1 = _get_client(redis.Redis, request, db=1)
+        r0.set("a", "foo")
+        assert r0.copy("a", "b", destination_db=1) == 1
+        assert r1.get("b") == b"foo"
+
     def test_decr(self, r):
         assert r.decr('a') == -1
         assert r['a'] == b'-1'