potential improvements to the class

src/sentry/integrations/request_buffer.py

@@ -2,12 +2,14 @@
 
 from django.conf import settings
 
-from sentry.utils import json, redis
+from sentry.utils import redis
 
 BUFFER_SIZE = 30  # 30 days
 KEY_EXPIRY = 60 * 60 * 24 * 30  # 30 days
 
-IS_BROKEN_RANGE = 7  # 7 days
+BROKEN_RANGE_DAYS = 7  # 7 days
+
+VALID_KEYS = ["success", "error", "fatal"]
 
 
 class IntegrationRequestBuffer:
@@ -17,93 +19,95 @@ class IntegrationRequestBuffer:
     """
 
     def __init__(self, key):
-        self.integrationkey = key
+        self.integration_key = key
 
         cluster_id = settings.SENTRY_INTEGRATION_ERROR_LOG_REDIS_CLUSTER
         self.client = redis.redis_clusters.get(cluster_id)
 
-    def _convert_obj_to_dict(self, redis_object):
-        """
-        Convert the request string stored in Redis to a python dict
-        """
-        return json.loads(redis_object)
-
-    def _get_all_from_buffer(self):
-        """
-        Get the list at the buffer key.
-        """
-
-        ret = []
-        now = datetime.now()
-        i = 0
-        buffer_key = f"{self.integrationkey}:{now.strftime('%Y-%m-%d')}"
-        while self.client.hgetall(buffer_key):
-            ret.append(self.client.hgetall(buffer_key))
-            cur = (now - timedelta(days=i)).strftime("%Y-%m-%d")
-            buffer_key = f"{self.integrationkey}:{cur}"
-            i += 1
-
-        return ret
-
-    def _get_broken_range_from_buffer(self):
-        """
-        Get the list at the buffer key in the broken range.
-        """
+    def record_error(self):
+        self._add("error")
 
-        ret = []
-        now = datetime.now()
-        for i in range(IS_BROKEN_RANGE):
-            cur = (now - timedelta(days=i)).strftime("%Y-%m-%d")
-            buffer_key = f"{self.integrationkey}:{cur}"
-            ret.append(self.client.hgetall(buffer_key))
+    def record_success(self):
+        self._add("success")
 
-        return ret
+    def record_fatal(self):
+        self._add("fatal")
 
     def is_integration_broken(self):
         """
         Integration is broken if we have 7 consecutive days of errors and no successes OR have a fatal error
 
         """
-        items = self._get_broken_range_from_buffer()
+        broken_range_days_counts = self._get_broken_range_from_buffer()
 
-        data = [item for item in items if int(item.get("fatal_count", 0)) > 0]
+        days_fatal = []
+        days_error = []
 
-        if len(data) > 0:
-            return True
+        for day_count in broken_range_days_counts:
+            if int(day_count.get("fatal_count", 0)) > 0:
+                days_fatal.append(day_count)
+            elif (
+                int(day_count.get("error_count", 0)) > 0
+                and int(day_count.get("success_count", 0)) == 0
+            ):
+                days_error.append(day_count)
 
-        data = [
-            item
-            for item in items
-            if int(item.get("error_count", 0)) > 0 and int(item.get("success_count", 0)) == 0
-        ]
+        if len(days_fatal) > 0:
+            return True
 
-        if not len(data):
+        if not len(days_error):
             return False
 
-        if len(data) < IS_BROKEN_RANGE:
+        if len(days_error) < BROKEN_RANGE_DAYS:
             return False
 
         return True
 
-    def add(self, count: str):
-        VALID_KEYS = ["success", "error", "fatal"]
+    def _add(self, count: str):
         if count not in VALID_KEYS:
             raise Exception("Requires a valid key param.")
 
         now = datetime.now().strftime("%Y-%m-%d")
+        buffer_key = f"{self.integration_key}:{now}"
 
-        buffer_key = f"{self.integrationkey}:{now}"
         pipe = self.client.pipeline()
         pipe.hincrby(buffer_key, count + "_count", 1)
         pipe.expire(buffer_key, KEY_EXPIRY)
         pipe.execute()
-        pipe.reset()
 
-    def record_error(self):
-        self.add("error")
+    def _get_all_from_buffer(self):
+        """
+        Get the list at the buffer key.
+        """
 
-    def record_success(self):
-        self.add("success")
+        now = datetime.now()
+        all_range = [
+            f"{self.integration_key}:{(now - timedelta(days=i)).strftime('%Y-%m-%d')}"
+            for i in range(BUFFER_SIZE)
+        ]
 
-    def record_fatal(self):
-        self.add("fatal")
+        return self._get_range_buffers(all_range)
+
+    def _get_broken_range_from_buffer(self):
+        """
+        Get the list at the buffer key in the broken range.
+        """
+
+        now = datetime.now()
+        broken_range_keys = [
+            f"{self.integration_key}:{(now - timedelta(days=i)).strftime('%Y-%m-%d')}"
+            for i in range(BROKEN_RANGE_DAYS)
+        ]
+
+        return self._get_range_buffers(broken_range_keys)
+
+    def _get_range_buffers(self, keys):
+        pipe = self.client.pipeline()
+        ret = []
+        for key in keys:
+            pipe.hgetall(key)
+        response = pipe.execute()
+        for item in response:
+            ret.append(item)
+
+        return ret