Support FT.PROFILE + test

Author: AvitalFineRedis

redis/commands/helpers.py

@@ -35,8 +35,7 @@ def delist(x):
 
 
 def parse_to_list(response):
-    """Optimistally parse the response to a list.
-    """
+    """Optimistically parse the response to a list."""
     res = []
     for item in response:
         try:
@@ -51,6 +50,40 @@ def parse_to_list(response):
     return res
 
 
+def parse_list_to_dict(response):
+    res = {}
+    for i in range(0, len(response), 2):
+        if isinstance(response[i], list):
+            res['Child iterators'].append(parse_list_to_dict(response[i]))
+        elif isinstance(response[i+1], list):
+            res['Child iterators'] = [parse_list_to_dict(response[i+1])]
+        else:
+            try:
+                res[response[i]] = float(response[i+1])
+            except (TypeError, ValueError):
+                res[response[i]] = response[i+1]
+    return res
+
+
+def parse_to_dict(response):
+    if response is None:
+        return {}
+
+    res = {}
+    for det in response:
+        if isinstance(det[1], list):
+            res[det[0]] = parse_list_to_dict(det[1])
+        else:
+            try:  # try to set the attribute. may be provided without value
+                try:  # try to convert the value to float
+                    res[det[0]] = float(det[1])
+                except (TypeError, ValueError):
+                    res[det[0]] = det[1]
+            except IndexError:
+                pass
+    return res
+
+
 def random_string(length=10):
     """
     Returns a random N character long string.

redis/commands/search/commands.py

@@ -7,6 +7,7 @@
 from ._util import to_string
 from .aggregation import AggregateRequest, AggregateResult, Cursor
 from .suggestion import SuggestionParser
+from ..helpers import parse_to_dict
 
 NUMERIC = "NUMERIC"
 
@@ -20,6 +21,7 @@
 EXPLAINCLI_CMD = "FT.EXPLAINCLI"
 DEL_CMD = "FT.DEL"
 AGGREGATE_CMD = "FT.AGGREGATE"
+PROFILE_CMD = "FT.PROFILE"
 CURSOR_CMD = "FT.CURSOR"
 SPELLCHECK_CMD = "FT.SPELLCHECK"
 DICT_ADD_CMD = "FT.DICTADD"
@@ -382,11 +384,11 @@ def explain_cli(self, query):  # noqa
 
     def aggregate(self, query):
         """
-        Issue an aggregation query
+        Issue an aggregation query.
 
         ### Parameters
 
-        **query**: This can be either an `AggeregateRequest`, or a `Cursor`
+        **query**: This can be either an `AggregateRequest`, or a `Cursor`
 
         An `AggregateResult` object is returned. You can access the rows from
         its `rows` property, which will always yield the rows of the result.
@@ -401,6 +403,10 @@ def aggregate(self, query):
             raise ValueError("Bad query", query)
 
         raw = self.execute_command(*cmd)
+        return self._get_AggregateResult(raw, query, has_cursor)
+
+    def _get_AggregateResult(self, raw, query, has_cursor):
+        # has_cursor = bool(query._cursor)
         if has_cursor:
             if isinstance(query, Cursor):
                 query.cid = raw[1]
@@ -418,8 +424,51 @@ def aggregate(self, query):
             schema = None
             rows = raw[1:]
 
-        res = AggregateResult(rows, cursor, schema)
-        return res
+        return AggregateResult(rows, cursor, schema)
+
+    def profile(self, query, limited=False):
+        """
+        Performs a search or aggregate command and collects performance
+        information.
+
+        ### Parameters
+
+        **query**: This can be either an `AggregateRequest`, `Query` or
+        string.
+        **limited**: If set to True, removes details of reader iterator.
+
+        """
+        st = time.time()
+        cmd = [PROFILE_CMD, self.index_name, ""]
+        if limited:
+            cmd.append("LIMITED")
+        cmd.append('QUERY')
+
+        if isinstance(query, AggregateRequest):
+            cmd[2] = "AGGREGATE"
+            cmd += query.build_args()
+        elif isinstance(query, Query):
+            cmd[2] = "SEARCH"
+            cmd += query.get_args()
+        elif isinstance(query, str):
+            cmd[2] = "SEARCH"
+            cmd.append(query)
+        else:
+            raise ValueError("Must provide AggregateRequest object, "
+                             "Query object or str.")
+
+        res = self.execute_command(*cmd)
+
+        if isinstance(query, AggregateRequest):
+            result = self._get_AggregateResult(res[0], query, query._cursor)
+        else:
+            result = Result(res[0],
+                            not query._no_content,
+                            duration=(time.time() - st) * 1000.0,
+                            has_payload=query._with_payloads,
+                            with_scores=query._with_scores,)
+
+        return result, parse_to_dict(res[1])
 
     def spellcheck(self, query, distance=None, include=None, exclude=None):
         """

tests/test_search.py

@@ -1313,3 +1313,28 @@ def test_json_with_jsonpath(client):
     assert res.docs[0].id == "doc:1"
     with pytest.raises(Exception):
         res.docs[0].name_unsupported
+
+
+@pytest.mark.redismod
+def test_profile(client):
+    client.ft().create_index((TextField('t'),))
+    client.ft().client.hset('1', 't', 'hello')
+    client.ft().client.hset('2', 't', 'world')
+
+    # check using Query
+    q = Query('hello|world').no_content()
+    res, det = client.ft().profile(q)
+    assert det['Iterators profile']['Counter'] == 2.0
+    assert len(det['Iterators profile']['Child iterators']) == 2
+    assert det['Iterators profile']['Type'] == 'UNION'
+    assert det['Parsing time'] < 0.3
+    assert len(res.docs) == 2  # check also the search result
+
+    # check using AggregateRequest
+    req = aggregations.AggregateRequest("*").load("t")\
+        .apply(prefix="startswith(@t, 'hel')")
+    res, det = client.ft().profile(req)
+    assert det['Iterators profile']['Counter'] == 2.0
+    assert det['Iterators profile']['Type'] == 'WILDCARD'
+    assert det['Parsing time'] < 0.3
+    assert len(res.rows) == 2  # check also the search result