Refactor skipping of blank periods

Moved the logic to calculate how many periods a result should be
shifted into the response module. Each result class controls how it is
shifted so that the Query class can be greatly simplified.

backdrop/read/api.py

@@ -155,7 +155,7 @@ def fetch(bucket_config):
 
         try:
             query = Query.parse(request.args)
-            data = bucket.query(query)
+            data = bucket.query(query).data()
 
         except InvalidOperationError:
             return log_error_and_respond(

backdrop/read/query.py

@@ -94,49 +94,13 @@ def __calculate_start_and_end(period, date, delta):
 
         return start_at, end_at
 
-    def __first_nonempty(self, data, is_reversed):
-        if is_reversed:
-            data = reversed(data)
+    def __skip_blank_periods(self, results, repository):
+        amount_to_shift = results.amount_to_shift(self.delta)
+        if amount_to_shift != 0:
+            query = self.get_shifted_query(shift=amount_to_shift)
+            results = query.execute(repository)
 
-        first_nonempty_index = next(
-            (i for i, d in enumerate(data) if d['_count'] > 0),
-            None)
-
-        if is_reversed:
-            first_nonempty_index = -first_nonempty_index
-
-        return first_nonempty_index
-
-    def __skip_blanks(self, data, repository):
-        is_reversed = self.delta < 0
-        first_index = -1 if is_reversed else 0
-
-        if data[first_index]['_count'] == 0:
-            # we need to skip blank results
-            first_nonempty_index = self.__first_nonempty(data, is_reversed)
-
-            if first_nonempty_index is None:
-                # we currently return no results if none of the
-                # results in the specified range contained any data
-                data = tuple()
-            else:
-                query = self.get_shifted_query(shift=first_nonempty_index)
-                data = query.execute(repository)
-
-        return data
-
-    def __grouped_skip_blanks(self, data, repository):
-        is_reversed = self.delta < 0
-
-        first_nonempty_index = min([
-            self.__first_nonempty(i['values'], is_reversed) for i in data],
-            key=abs)
-
-        if first_nonempty_index != 0:
-            query = self.get_shifted_query(shift=first_nonempty_index)
-            data = query.execute(repository)
-
-        return data
+        return results
 
     def get_shifted_query(self, shift):
         """Return a new Query where the date is shifted by n periods"""
@@ -148,7 +112,6 @@ def get_shifted_query(self, shift):
         return Query.create(**args)
 
     def to_mongo_query(self):
-
         mongo_query = {}
         if self.start_at or self.end_at:
             mongo_query["_timestamp"] = {}
@@ -162,21 +125,18 @@ def to_mongo_query(self):
 
     def execute(self, repository):
         if self.group_by and self.period:
-            data = self.__execute_period_group_query(repository).data()
+            result = self.__execute_period_group_query(repository)
         elif self.group_by:
-            data = self.__execute_grouped_query(repository).data()
+            result = self.__execute_grouped_query(repository)
         elif self.period:
-            data = self.__execute_period_query(repository).data()
+            result = self.__execute_period_query(repository)
         else:
-            data = self.__execute_query(repository).data()
+            result = self.__execute_query(repository)
 
         if self.delta:
-            if self.group_by:
-                data = self.__grouped_skip_blanks(data, repository)
-            else:
-                data = self.__skip_blanks(data, repository)
+            result = self.__skip_blank_periods(result, repository)
 
-        return data
+        return result
 
     def __get_period_key(self):
         return self.period.start_at_key

backdrop/read/response.py

@@ -24,6 +24,20 @@ def create_period_group(doc, period):
     return datum
 
 
+def first_nonempty(data, is_reversed):
+    if is_reversed:
+        data = reversed(data)
+
+    first_nonempty_index = next(
+        (i for i, d in enumerate(data) if d['_count'] > 0),
+        None)
+
+    if is_reversed:
+        first_nonempty_index = -first_nonempty_index
+
+    return first_nonempty_index
+
+
 class SimpleData(object):
     def __init__(self, cursor):
         self._data = []
@@ -39,6 +53,10 @@ def __add(self, document):
     def data(self):
         return tuple(self._data)
 
+    def amount_to_shift(self, delta):
+        """This response type cannot be shifted"""
+        return 0
+
 
 class PeriodData(object):
     def __init__(self, cursor, period):
@@ -71,6 +89,11 @@ def __create_datum(self, doc):
 
         return dict(datum.items() + doc.items())
 
+    def amount_to_shift(self, delta):
+        is_reversed = delta < 0
+
+        return first_nonempty(self._data, is_reversed)
+
 
 class GroupedData(object):
     def __init__(self, cursor):
@@ -84,6 +107,10 @@ def __add(self, document):
     def data(self):
         return tuple(self._data)
 
+    def amount_to_shift(self, delta):
+        """This response type cannot be shifted"""
+        return 0
+
 
 class PeriodGroupedData(object):
     def __init__(self, cursor, period):
@@ -122,3 +149,10 @@ def fill_missing_periods(self, start_date, end_date, collect=None):
                 data=self._data[i]['values'],
                 default=default
             )
+
+    def amount_to_shift(self, delta):
+        is_reversed = delta < 0
+
+        return min([
+            first_nonempty(i['values'], is_reversed) for i in self._data],
+            key=abs)

tests/core/integration/test_bucket_integration.py

@@ -79,7 +79,7 @@ def test_period_queries_get_sorted_by__week_start_at(self):
         self.setup__timestamp_data()
         query = Query.create(period=WEEK)
         result = query.execute(self.bucket.repository)
-        assert_that(result, contains(
+        assert_that(result.data(), contains(
             has_entry('_start_at', d_tz(2012, 12, 31)),
             has_entry('_start_at', d_tz(2013, 1, 28)),
             has_entry('_start_at', d_tz(2013, 2, 25))

tests/core/test_bucket.py

@@ -68,7 +68,7 @@ def test_group_by_query(self):
         ]
 
         query = Query.create(group_by="name")
-        query_result = self.bucket.query(query)
+        query_result = self.bucket.query(query).data()
 
         self.mock_repository.group.assert_called_once_with(
             "name", query, None, None, [])
@@ -148,7 +148,7 @@ def test_week_query(self):
         ]
 
         query = Query.create(period=WEEK)
-        query_result = self.bucket.query(query)
+        query_result = self.bucket.query(query).data()
 
         self.mock_repository.group.assert_called_once_with(
             "_week_start_at", query, sort=['_week_start_at', 'ascending'],
@@ -173,7 +173,7 @@ def test_month_query(self):
         ]
 
         query = Query.create(period=MONTH)
-        query_result = self.bucket.query(query)
+        query_result = self.bucket.query(query).data()
         self.mock_repository.group.assert_called_once_with(
             "_month_start_at", query, sort=['_month_start_at', 'ascending'],
             limit=None, collect=[])
@@ -235,7 +235,7 @@ def test_period_query_adds_missing_periods_in_correct_order(self):
                                                 end_at=d_tz(2013, 2, 18, 0, 0,
                                                             0)))
 
-        assert_that(result, contains(
+        assert_that(result.data(), contains(
             has_entries({"_start_at": d_tz(2013, 1, 7), "_count": 0}),
             has_entries({"_start_at": d_tz(2013, 1, 14), "_count": 32}),
             has_entries({"_start_at": d_tz(2013, 1, 21), "_count": 45}),
@@ -279,32 +279,35 @@ def test_week_and_group_query(self):
         ]
         query_result = self.bucket.query(
             Query.create(period=WEEK, group_by="some_group"))
-        assert_that(query_result, has_length(2))
-        assert_that(query_result, has_item(has_entries({
+
+        data = query_result.data()
+
+        assert_that(data, has_length(2))
+        assert_that(data, has_item(has_entries({
             "values": has_item({
                 "_start_at": d_tz(2013, 1, 7, 0, 0, 0),
                 "_end_at": d_tz(2013, 1, 14, 0, 0, 0),
                 "_count": 1
             }),
             "some_group": "val1"
         })))
-        assert_that(query_result, has_item(has_entries({
+        assert_that(data, has_item(has_entries({
             "values": has_item({
                 "_start_at": d_tz(2013, 1, 14, 0, 0, 0),
                 "_end_at": d_tz(2013, 1, 21, 0, 0, 0),
                 "_count": 5
             }),
             "some_group": "val1"
         })))
-        assert_that(query_result, has_item(has_entries({
+        assert_that(data, has_item(has_entries({
             "values": has_item({
                 "_start_at": d_tz(2013, 1, 7, 0, 0, 0),
                 "_end_at": d_tz(2013, 1, 14, 0, 0, 0),
                 "_count": 2
             }),
             "some_group": "val2"
         })))
-        assert_that(query_result, has_item(has_entries({
+        assert_that(data, has_item(has_entries({
             "values": has_item({
                 "_start_at": d_tz(2013, 1, 14, 0, 0, 0),
                 "_end_at": d_tz(2013, 1, 21, 0, 0, 0),
@@ -353,9 +356,10 @@ def test_month_and_group_query(self):
 
         query_result = self.bucket.query(Query.create(period=MONTH,
                                                       group_by="some_group"))
-        assert_that(query_result,
+        data = query_result.data()
+        assert_that(data,
                     has_item(has_entries({"values": has_length(2)})))
-        assert_that(query_result,
+        assert_that(data,
                     has_item(has_entries({"values": has_length(3)})))
 
     def test_month_and_group_query_with_start_and_end_at(self):
@@ -400,18 +404,19 @@ def test_month_and_group_query_with_start_and_end_at(self):
                                                       group_by="some_group",
                                                       start_at=d(2013, 1, 1),
                                                       end_at=d(2013, 4, 2)))
-        assert_that(query_result,
+        data = query_result.data()
+        assert_that(data,
                     has_item(has_entries({"values": has_length(4)})))
-        assert_that(query_result,
+        assert_that(data,
                     has_item(has_entries({"values": has_length(4)})))
 
-        first_group = query_result[0]["values"]
+        first_group = data[0]["values"]
         assert_that(first_group, has_item(has_entries({
             "_start_at": d_tz(2013, 3, 1)})))
         assert_that(first_group, has_item(has_entries({
             "_start_at": d_tz(2013, 4, 1)})))
 
-        first_group = query_result[1]["values"]
+        first_group = data[1]["values"]
         assert_that(first_group, has_item(has_entries({
             "_start_at": d_tz(2013, 1, 1)})))
         assert_that(first_group, has_item(has_entries({
@@ -456,7 +461,7 @@ def test_period_group_query_adds_missing_periods_in_correct_order(self):
                          start_at=d_tz(2013, 1, 7, 0, 0, 0),
                          end_at=d_tz(2013, 2, 4, 0, 0, 0)))
 
-        assert_that(query_result, has_item(has_entries({
+        assert_that(query_result.data(), has_item(has_entries({
             "some_group": "val1",
             "values": contains(
                 has_entries({"_start_at": d_tz(2013, 1, 7), "_count": 0}),
@@ -466,7 +471,7 @@ def test_period_group_query_adds_missing_periods_in_correct_order(self):
             ),
         })))
 
-        assert_that(query_result, has_item(has_entries({
+        assert_that(query_result.data(), has_item(has_entries({
             "some_group": "val2",
             "values": contains(
                 has_entries({"_start_at": d_tz(2013, 1, 7), "_count": 0}),