Support SYNC and PSYNC

redis/commands/core.py

@@ -637,6 +637,31 @@ def flushdb(self, asynchronous=False, **kwargs):
             args.append(b"ASYNC")
         return self.execute_command("FLUSHDB", *args, **kwargs)
 
+    def sync(self):
+        """
+        Initiates a replication stream from the master.
+
+        For more information check https://redis.io/commands/sync
+        """
+        from redis.client import NEVER_DECODE
+
+        options = {}
+        options[NEVER_DECODE] = []
+        return self.execute_command("SYNC", **options)
+
+    def psync(self, replicationid, offset):
+        """
+        Initiates a replication stream from the master.
+        Newer version for `sync`.
+
+        For more information check https://redis.io/commands/sync
+        """
+        from redis.client import NEVER_DECODE
+
+        options = {}
+        options[NEVER_DECODE] = []
+        return self.execute_command("PSYNC", replicationid, offset, **options)
+
     def swapdb(self, first, second, **kwargs):
         """
         Swap two databases

redis/connection.py

@@ -382,7 +382,7 @@ def __del__(self):
         except Exception:
             pass
 
-    def on_connect(self, connection):
+    def on_connect(self, connection, **kwargs):
         self._sock = connection._sock
         self._socket_timeout = connection.socket_timeout
         kwargs = {

tests/test_commands.py

@@ -4151,6 +4151,18 @@ def test_replicaof(self, r):
             assert r.replicaof("NO ONE")
         assert r.replicaof("NO", "ONE")
 
+    @skip_if_server_version_lt("2.8.0")
+    def test_sync(self, r):
+        r2 = redis.Redis(port=6380, decode_responses=False)
+        res = r2.sync()
+        assert b"REDIS" in res
+
+    @skip_if_server_version_lt("2.8.0")
+    def test_psync(self, r):
+        r2 = redis.Redis(port=6380, decode_responses=False)
+        res = r2.psync(r2.client_id(), 1)
+        assert b"FULLRESYNC" in res
+
 
 @pytest.mark.onlynoncluster
 class TestBinarySave: