add support for lcs (#1924)

redis/commands/core.py

@@ -1884,6 +1884,35 @@ def unlink(self, *names):
         """
         return self.execute_command("UNLINK", *names)
 
+    def lcs(
+        self,
+        key1: str,
+        key2: str,
+        len: Optional[bool] = False,
+        idx: Optional[bool] = False,
+        minmatchlen: Optional[int] = 0,
+        withmatchlen: Optional[bool] = False,
+    ) -> Union[str, int, list]:
+        """
+        Find the longest common subsequence between ``key1`` and ``key2``.
+        If ``len`` is true the length of the match will will be returned.
+        If ``idx`` is true the match position in each strings will be returned.
+        ``minmatchlen`` restrict the list of matches to the ones of
+        the given ``minmatchlen``.
+        If ``withmatchlen`` the length of the match also will be returned.
+        For more information check https://redis.io/commands/lcs
+        """
+        pieces = [key1, key2]
+        if len:
+            pieces.append("LEN")
+        if idx:
+            pieces.append("IDX")
+        if minmatchlen != 0:
+            pieces.extend(["MINMATCHLEN", minmatchlen])
+        if withmatchlen:
+            pieces.append("WITHMATCHLEN")
+        return self.execute_command("LCS", *pieces)
+
 
 class ListCommands:
     """

tests/test_commands.py

@@ -968,6 +968,17 @@ def test_unlink_with_multiple_keys(self, r):
         assert r.get("a") is None
         assert r.get("b") is None
 
+    @pytest.mark.onlynoncluster
+    # @skip_if_server_version_lt("7.0.0") turn on after redis 7 release
+    def test_lcs(self, unstable_r):
+        unstable_r.mset({"foo": "ohmytext", "bar": "mynewtext"})
+        assert unstable_r.lcs("foo", "bar") == b"mytext"
+        assert unstable_r.lcs("foo", "bar", len=True) == 6
+        result = [b"matches", [[[4, 7], [5, 8]]], b"len", 6]
+        assert unstable_r.lcs("foo", "bar", idx=True, minmatchlen=3) == result
+        with pytest.raises(redis.ResponseError):
+            assert unstable_r.lcs("foo", "bar", len=True, idx=True)
+
     @skip_if_server_version_lt("2.6.0")
     def test_dump_and_restore(self, r):
         r["a"] = "foo"