LPOS: add new command

fix #1353

Signed-off-by: Paul Spooren <[email protected]>

Author: aparcar

redis/client.py

@@ -2030,6 +2030,39 @@ def rpushx(self, name, value):
         "Push ``value`` onto the tail of the list ``name`` if ``name`` exists"
         return self.execute_command('RPUSHX', name, value)
 
+    def lpos(self, name, value, first=None, count=None, maxlen=None):
+        """
+        Get position of ``value`` withn the list ``name``
+
+         ``first`` "rank" is the position of the match, so if it is 1, the
+         first match is returned, if it is 2 the second match is returned and
+         so forth.  It is 1 by default. If negative has the same meaning but
+         the search is performed starting from the end of the list.
+
+         If ``count`` is given, instead of returning the single element, a list
+         of all the matching elements up to "num-matches" are returned.
+         ``count`` can be combiled with ``count`` in order to returning only
+         the element starting from the Nth. If ``count`` is zero, all the
+         matching elements are returned.
+
+         ``maxlen`` tells the command to scan a max of len elements. If zero
+         (the default), all the elements in the list are scanned if needed.
+
+         The returned elements indexes are always referring to what ``lindex``
+         would return. So first element from head is 0, and so forth.
+        """
+        pieces = [name, value]
+        if first is not None:
+            pieces.extend(['FIRST', first])
+
+        if count is not None:
+            pieces.extend(['COUNT', count])
+
+        if maxlen is not None:
+            pieces.extend(['MAXLEN', maxlen])
+
+        return self.execute_command('LPOS', *pieces)
+
     def sort(self, name, start=None, num=None, by=None, get=None,
              desc=False, alpha=False, store=None, groups=False):
         """

tests/test_commands.py

@@ -1052,6 +1052,36 @@ def test_rpush(self, r):
         assert r.rpush('a', '3', '4') == 4
         assert r.lrange('a', 0, -1) == [b'1', b'2', b'3', b'4']
 
+    def test_lpos(self, r):
+        assert r.rpush('a', 'a', 'b', 'c', '1', '2', '3', 'c', 'c') == 8
+        assert r.lpos('a', 'a') == 0
+        assert r.lpos('a', 'c') == 2
+
+        assert r.lpos('a', 'c', first=1) == 2
+        assert r.lpos('a', 'c', first=2) == 6
+        assert r.lpos('a', 'c', first=4) is None
+        assert r.lpos('a', 'c', first=-1) == 7
+        assert r.lpos('a', 'c', first=-2) == 6
+
+        assert r.lpos('a', 'c', count=0) == [2, 6, 7]
+        assert r.lpos('a', 'c', count=1) == [2]
+        assert r.lpos('a', 'c', count=2) == [2, 6]
+        assert r.lpos('a', 'c', count=100) == [2, 6, 7]
+
+        assert r.lpos('a', 'c', count=0, first=2) == [6, 7]
+        assert r.lpos('a', 'c', count=2, first=-1) == [7, 6]
+
+        assert r.lpos('axxx', 'c', count=0, first=2) == []
+
+        assert r.lpos('a', 'x', count=2, first=-1) == []
+        assert r.lpos('a', 'x', first=-1) is None
+
+        assert r.lpos('a', 'a', count=0, maxlen=1) == [0]
+        assert r.lpos('a', 'c', count=0, maxlen=1) == []
+        assert r.lpos('a', 'c', count=0, maxlen=3) == [2]
+        assert r.lpos('a', 'c', count=0, maxlen=3, first=-1) == [7, 6]
+        assert r.lpos('a', 'c', count=0, maxlen=7, first=2) == [6]
+
     def test_rpushx(self, r):
         assert r.rpushx('a', 'b') == 0
         assert r.lrange('a', 0, -1) == []