Use LRU cache

Author: cmanallen

sentry_sdk/_lru_cache.py

@@ -62,6 +62,8 @@
 
 """
 
+from copy import copy
+
 SENTINEL = object()
 
 
@@ -89,6 +91,13 @@ def __init__(self, max_size):
 
         self.hits = self.misses = 0
 
+    def __copy__(self):
+        cache = LRUCache(self.max_size)
+        cache.full = self.full
+        cache.cache = copy(self.cache)
+        cache.root = copy(self.root)
+        return cache
+
     def set(self, key, value):
         link = self.cache.get(key, SENTINEL)
 
@@ -154,3 +163,11 @@ def get(self, key, default=None):
         self.hits += 1
 
         return link[VALUE]
+
+    def get_all(self):
+        nodes = []
+        node = self.root[NEXT]
+        while node is not self.root:
+            nodes.append((node[KEY], node[VALUE]))
+            node = node[NEXT]
+        return nodes

sentry_sdk/flag_utils.py

@@ -1,7 +1,7 @@
 from copy import copy
 from typing import TYPE_CHECKING
 
-import itertools
+from sentry_sdk._lru_cache import LRUCache
 
 if TYPE_CHECKING:
     from typing import TypedDict
@@ -13,62 +13,23 @@ class FlagBuffer:
 
     def __init__(self, capacity):
         # type: (int) -> None
-        self.buffer = []  # type: list[Flag]
+        self.buffer = LRUCache(capacity)
         self.capacity = capacity
-        self.ip = 0
-
-    @property
-    def index(self):
-        # type: () -> int
-        return self.ip % self.capacity
 
     def clear(self):
         # type: () -> None
-        self.buffer = []
-        self.ip = 0
+        self.buffer = LRUCache(self.capacity)
 
     def __copy__(self):
         # type: () -> FlagBuffer
         buffer = FlagBuffer(capacity=self.capacity)
         buffer.buffer = copy(self.buffer)
-        buffer.ip = self.ip
         return buffer
 
     def get(self):
         # type: () -> list[FlagData]
-        if self.ip >= self.capacity:
-            iterator = itertools.chain(
-                range(self.index, self.capacity), range(0, self.index)
-            )
-            return [self.buffer[i].asdict for i in iterator]
-        else:
-            return [flag.asdict for flag in self.buffer]
+        return [{"flag": key, "result": value} for key, value in self.buffer.get_all()]
 
     def set(self, flag, result):
         # type: (str, bool) -> None
-        flag_ = Flag(flag, result)
-
-        if self.ip >= self.capacity:
-            self.buffer[self.index] = flag_
-        else:
-            self.buffer.append(flag_)
-
-        self.ip += 1
-
-
-class Flag:
-    __slots__ = ("flag", "result")
-
-    def __init__(self, flag, result):
-        # type: (str, bool) -> None
-        self.flag = flag
-        self.result = result
-
-    def __copy__(self):
-        # type: () -> Flag
-        return Flag(self.flag, self.result)
-
-    @property
-    def asdict(self):
-        # type: () -> FlagData
-        return {"flag": self.flag, "result": self.result}
+        self.buffer.set(flag, result)