Stralgo

redis/client.py

@@ -405,6 +405,34 @@ def parse_slowlog_get(response, **options):
     } for item in response]
 
 
+def parse_stralgo(response, **options):
+    """
+    Parse the response from `STRALGO` command.
+    Without modifiers the returned value is string.
+    When LEN is given the command returns the length of the result
+    (i.e integer).
+    When IDX is given the command returns a dictionary with the LCS
+    length and all the ranges in both the strings, start and end
+    offset for each string, where there are matches.
+    When WITHMATCHLEN is given, each array representing a match will
+    also have the length of the match at the beginning of the array.
+    """
+    if options.get('len', False):
+        return int(response)
+    if options.get('idx', False):
+        if options.get('withmatchlen', False):
+            matches = [[(int(match[-1]))] + list(map(tuple, match[:-1]))
+                       for match in response[1]]
+        else:
+            matches = [list(map(tuple, match))
+                       for match in response[1]]
+        return {
+            str_if_bytes(response[0]): matches,
+            str_if_bytes(response[2]): int(response[3])
+        }
+    return str_if_bytes(response)
+
+
 def parse_cluster_info(response, **options):
     response = str_if_bytes(response)
     return dict(line.split(':') for line in response.splitlines() if line)
@@ -668,6 +696,7 @@ class Redis(Commands, object):
         'MODULE LIST': lambda r: [pairs_to_dict(m) for m in r],
         'OBJECT': parse_object,
         'PING': lambda r: str_if_bytes(r) == 'PONG',
+        'STRALGO': parse_stralgo,
         'PUBSUB NUMSUB': parse_pubsub_numsub,
         'RANDOMKEY': lambda r: r and r or None,
         'SCAN': parse_scan,

redis/commands.py

@@ -1100,6 +1100,53 @@ def setrange(self, name, offset, value):
         """
         return self.execute_command('SETRANGE', name, offset, value)
 
+    def stralgo(self, algo, value1, value2, specific_argument='strings',
+                len=False, idx=False, minmatchlen=None, withmatchlen=False):
+        """
+        Implements complex algorithms that operate on strings.
+        Right now the only algorithm implemented is the LCS algorithm
+        (longest common substring). However new algorithms could be
+        implemented in the future.
+
+        ``algo`` Right now must be LCS
+        ``value1`` and ``value2`` Can be two strings or two keys
+        ``specific_argument`` Specifying if the arguments to the algorithm
+        will be keys or strings. strings is the default.
+        ``len`` Returns just the len of the match.
+        ``idx`` Returns the match positions in each string.
+        ``minmatchlen`` Restrict the list of matches to the ones of a given
+        minimal length. Can be provided only when ``idx`` set to True.
+        ``withmatchlen`` Returns the matches with the len of the match.
+        Can be provided only when ``idx`` set to True.
+        """
+        # check validity
+        supported_algo = ['LCS']
+        if algo not in supported_algo:
+            raise DataError("The supported algorithms are: %s"
+                            % (', '.join(supported_algo)))
+        if specific_argument not in ['keys', 'strings']:
+            raise DataError("specific_argument can be only"
+                            " keys or strings")
+        if len and idx:
+            raise DataError("len and idx cannot be provided together.")
+
+        pieces = [algo, specific_argument.upper(), value1, value2]
+        if len:
+            pieces.append(b'LEN')
+        if idx:
+            pieces.append(b'IDX')
+        try:
+            int(minmatchlen)
+            pieces.extend([b'MINMATCHLEN', minmatchlen])
+        except TypeError:
+            pass
+        if withmatchlen:
+            pieces.append(b'WITHMATCHLEN')
+
+        return self.execute_command('STRALGO', *pieces, len=len, idx=idx,
+                                    minmatchlen=minmatchlen,
+                                    withmatchlen=withmatchlen)
+
     def strlen(self, name):
         "Return the number of bytes stored in the value of ``name``"
         return self.execute_command('STRLEN', name)

tests/test_commands.py

@@ -1052,6 +1052,49 @@ def test_setrange(self, r):
         assert r.setrange('a', 6, '12345') == 11
         assert r['a'] == b'abcdef12345'
 
+    @skip_if_server_version_lt('6.0.0')
+    def test_stralgo_lcs(self, r):
+        key1 = 'key1'
+        key2 = 'key2'
+        value1 = 'ohmytext'
+        value2 = 'mynewtext'
+        res = 'mytext'
+        # test LCS of strings
+        assert r.stralgo('LCS', value1, value2) == res
+        # test using keys
+        r.mset({key1: value1, key2: value2})
+        assert r.stralgo('LCS', key1, key2, specific_argument="keys") == res
+        # test other labels
+        assert r.stralgo('LCS', value1, value2, len=True) == len(res)
+        assert r.stralgo('LCS', value1, value2, idx=True) == \
+               {
+                   'len': len(res),
+                   'matches': [[(4, 7), (5, 8)], [(2, 3), (0, 1)]]
+               }
+        assert r.stralgo('LCS', value1, value2,
+                         idx=True, withmatchlen=True) == \
+               {
+                   'len': len(res),
+                   'matches': [[4, (4, 7), (5, 8)], [2, (2, 3), (0, 1)]]
+               }
+        assert r.stralgo('LCS', value1, value2,
+                         idx=True, minmatchlen=4, withmatchlen=True) == \
+               {
+                   'len': len(res),
+                   'matches': [[4, (4, 7), (5, 8)]]
+               }
+
+    @skip_if_server_version_lt('6.0.0')
+    def test_stralgo_negative(self, r):
+        with pytest.raises(exceptions.DataError):
+            r.stralgo('ISSUB', 'value1', 'value2')
+        with pytest.raises(exceptions.DataError):
+            r.stralgo('LCS', 'value1', 'value2', len=True, idx=True)
+        with pytest.raises(exceptions.DataError):
+            r.stralgo('LCS', 'value1', 'value2', specific_argument="INT")
+        with pytest.raises(ValueError):
+            r.stralgo('LCS', 'value1', 'value2', idx=True, minmatchlen="one")
+
     def test_strlen(self, r):
         r['a'] = 'foo'
         assert r.strlen('a') == 3