[fix] fixes APPLY / SORTBY / GROUPBY / REDUCE order on FT.AGGREGATE s… (#41)

* [fix] fixes APPLY / SORTBY / GROUPBY / REDUCE order on FT.AGGREGATE support.

* [add] setted sorby max default to 0. [add] set apply alias as not mandatory

* [add] added supported for filter expressions on aggregations. include more examples on test_builder.py

* [fix] corrected FT.AGGREGATE filter expressions to relate to the current state of the pipeline

Author: filipecosta90

.circleci/config.yml

@@ -77,7 +77,13 @@ jobs:
           name: run tests
           command: |
             . venv/bin/activate
-            REDIS_PORT=6379 python test/test.py 
+            REDIS_PORT=6379 python test/test.py
+
+      - run:
+          name: run query builder tests
+          command: |
+            . venv/bin/activate
+            python test/test.py
 
       # no need for store_artifacts on nightly builds 
 

redisearch/aggregation.py

@@ -99,15 +99,58 @@ def __init__(self, fields, reducers):
         self.limit = Limit()
 
     def build_args(self):
-        ret = [str(len(self.fields))]
+        ret = ['GROUPBY', str(len(self.fields))]
         ret.extend(self.fields)
         for reducer in self.reducers:
             ret += ['REDUCE', reducer.NAME, str(len(reducer.args))]
             ret.extend(reducer.args)
-            if reducer._alias:
+            if reducer._alias is not None:
                 ret += ['AS', reducer._alias]
         return ret
 
+class Projection(object):
+    """
+    This object automatically created in the `AggregateRequest.apply()`
+    """
+
+    def __init__(self, projector, alias=None ):
+
+        self.alias = alias
+        self.projector = projector
+
+    def build_args(self):
+        ret = ['APPLY', self.projector]
+        if self.alias is not None:
+            ret += ['AS', self.alias]
+
+        return ret
+
+class SortBy(object):
+    """
+    This object automatically created in the `AggregateRequest.sort_by()`
+    """
+
+    def __init__(self, fields, max=0):
+        self.fields = fields
+        self.max = max
+
+
+
+    def build_args(self):
+        fields_args = []
+        for f in self.fields:
+            if isinstance(f, SortDirection):
+                fields_args += [f.field, f.DIRSTRING]
+            else:
+                fields_args += [f]
+
+        ret = ['SORTBY', str(len(fields_args))]
+        ret.extend(fields_args)
+        if self.max > 0:
+            ret += ['MAX', str(self.max)]
+
+        return ret
+
 
 class AggregateRequest(object):
     """
@@ -127,11 +170,9 @@ def __init__(self, query='*'):
         return the object itself, making them useful for chaining.
         """
         self._query = query
-        self._groups = []
-        self._projections = []
+        self._aggregateplan = []
         self._loadfields = []
         self._limit = Limit()
-        self._sortby = []
         self._max = 0
         self._with_schema = False
         self._verbatim = False
@@ -162,7 +203,7 @@ def group_by(self, fields, *reducers):
             `aggregation` module.
         """
         group = Group(fields, reducers)
-        self._groups.append(group)
+        self._aggregateplan.extend(group.build_args())
 
         return self
 
@@ -177,7 +218,8 @@ def apply(self, **kwexpr):
             expression itself, for example `apply(square_root="sqrt(@foo)")`
         """
         for alias, expr in kwexpr.items():
-            self._projections.append([alias, expr])
+            projection = Projection(expr, alias )
+            self._aggregateplan.extend(projection.build_args())
 
         return self
 
@@ -224,10 +266,7 @@ def limit(self, offset, num):
 
         """
         limit = Limit(offset, num)
-        if self._groups:
-            self._groups[-1].limit = limit
-        else:
-            self._limit = limit
+        self._limit = limit
         return self
 
     def sort_by(self, *fields, **kwargs):
@@ -258,16 +297,34 @@ def sort_by(self, *fields, **kwargs):
             .sort_by(Desc('@paid'), max=10)
         ```
         """
-        self._max = kwargs.get('max', 0)
         if isinstance(fields, (string_types, SortDirection)):
             fields = [fields]
-        for f in fields:
-            if isinstance(f, SortDirection):
-                self._sortby += [f.field, f.DIRSTRING]
-            else:
-                self._sortby.append(f)
+
+        max = kwargs.get('max', 0)
+        sortby = SortBy(fields, max)
+
+        self._aggregateplan.extend(sortby.build_args())
+        return self
+
+    def filter(self, expressions):
+        """
+        Specify filter for post-query results using predicates relating to values in the result set.
+
+        ### Parameters
+
+        - **fields**: Fields to group by. This can either be a single string,
+            or a list of strings.
+        """
+        if isinstance(expressions, (string_types)):
+            expressions = [expressions]
+
+        for expression in expressions:
+            self._aggregateplan.extend(['FILTER', expression])
+
         return self
 
+
+
     def with_schema(self):
         """
         If set, the `schema` property will contain a list of `[field, type]`
@@ -312,18 +369,8 @@ def build_args(self):
             ret.append('LOAD')
             ret.append(str(len(self._loadfields)))
             ret.extend(self._loadfields)
-        for group in self._groups:
-            ret += ['GROUPBY'] + group.build_args() + group.limit.build_args()
-        for alias, projector in self._projections:
-            ret += ['APPLY', projector]
-            if alias:
-                ret += ['AS', alias]
-
-        if self._sortby:
-            ret += ['SORTBY', str(len(self._sortby))]
-            ret += self._sortby
-            if self._max:
-                ret += ['MAX', str(self._max)]
+
+        ret.extend(self._aggregateplan)
 
         ret += self._limit.build_args()
 

test/test_builder.py

@@ -1,9 +1,9 @@
-from unittest import TestCase
+import unittest
 import redisearch.aggregation as a
 import redisearch.querystring as q
 import redisearch.reducers as r
 
-class QueryBuilderTest(TestCase):
+class QueryBuilderTest(unittest.TestCase):
     def testBetween(self):
         b = q.between(1, 10)
         self.assertEqual('[1 10]', str(b))
@@ -42,16 +42,16 @@ def testGroup(self):
         # Single field, single reducer
         g = a.Group('foo', r.count())
         ret = g.build_args()
-        self.assertEqual(['1', 'foo', 'REDUCE', 'COUNT', '0'], ret)
+        self.assertEqual(['GROUPBY', '1', 'foo', 'REDUCE', 'COUNT', '0'], ret)
 
         # Multiple fields, single reducer
         g = a.Group(['foo', 'bar'], r.count())
-        self.assertEqual(['2', 'foo', 'bar', 'REDUCE', 'COUNT', '0'],
+        self.assertEqual(['GROUPBY', '2', 'foo', 'bar', 'REDUCE', 'COUNT', '0'],
                          g.build_args())
 
         # Multiple fields, multiple reducers
         g = a.Group(['foo', 'bar'], [r.count(), r.count_distinct('@fld1')])
-        self.assertEqual(['2', 'foo', 'bar', 'REDUCE', 'COUNT', '0', 'REDUCE', 'COUNT_DISTINCT', '1', '@fld1'],
+        self.assertEqual(['GROUPBY', '2', 'foo', 'bar', 'REDUCE', 'COUNT', '0', 'REDUCE', 'COUNT_DISTINCT', '1', '@fld1'],
                          g.build_args())
 
     def testAggRequest(self):
@@ -62,13 +62,38 @@ def testAggRequest(self):
         req = a.AggregateRequest().group_by('@foo', r.count())
         self.assertEqual(['*', 'GROUPBY', '1', '@foo', 'REDUCE', 'COUNT', '0'], req.build_args())
 
+        # Test with group_by and alias on reducer
+        req = a.AggregateRequest().group_by('@foo', r.count().alias('foo_count'))
+        self.assertEqual(['*', 'GROUPBY', '1', '@foo', 'REDUCE', 'COUNT', '0', 'AS', 'foo_count'], req.build_args())
+
         # Test with limit
-        req = a.AggregateRequest().\
-            group_by('@foo', r.count()).\
+        req = a.AggregateRequest(). \
+            group_by('@foo', r.count()). \
             sort_by('@foo')
         self.assertEqual(['*', 'GROUPBY', '1', '@foo', 'REDUCE', 'COUNT', '0', 'SORTBY', '1',
                           '@foo'], req.build_args())
 
+        # Test with apply
+        req = a.AggregateRequest(). \
+            apply(foo="@bar / 2"). \
+            group_by('@foo', r.count())
+
+        self.assertEqual(['*', 'APPLY', '@bar / 2', 'AS', 'foo', 'GROUPBY', '1', '@foo', 'REDUCE', 'COUNT', '0'],
+                         req.build_args())
+
+        # Test with filter
+        req = a.AggregateRequest().group_by('@foo', r.count()).filter( "@foo=='bar'")
+        self.assertEqual(['*', 'GROUPBY', '1', '@foo', 'REDUCE', 'COUNT', '0', 'FILTER', "@foo=='bar'" ], req.build_args())
+
+        # Test with filter on different state of the pipeline
+        req = a.AggregateRequest().filter("@foo=='bar'").group_by('@foo', r.count())
+        self.assertEqual(['*', 'FILTER', "@foo=='bar'", 'GROUPBY', '1', '@foo','REDUCE', 'COUNT', '0' ], req.build_args())
+
+        # Test with filter on different state of the pipeline
+        req = a.AggregateRequest().filter(["@foo=='bar'","@foo2=='bar2'"]).group_by('@foo', r.count())
+        self.assertEqual(['*', 'FILTER', "@foo=='bar'", 'FILTER', "@foo2=='bar2'", 'GROUPBY', '1', '@foo', 'REDUCE', 'COUNT', '0'],
+                         req.build_args())
+
         # Test with sort_by
         req = a.AggregateRequest().group_by('@foo', r.count()).sort_by('@date')
         # print req.build_args()
@@ -105,4 +130,8 @@ def test_reducers(self):
         self.assertEqual(('f1', 'BY', 'f2', 'ASC'), r.first_value('f1', a.Asc('f2')).args)
         self.assertEqual(('f1', 'BY', 'f1', 'ASC'), r.first_value('f1', a.Asc).args)
 
-        self.assertEqual(('f1', '50'), r.random_sample('f1', 50).args)
+        self.assertEqual(('f1', '50'), r.random_sample('f1', 50).args)
+
+if __name__ == '__main__':
+
+    unittest.main()