refactor: use expire_at in ll instead of time of creation, replace time() with monotonic()

Author: zoola969

tests/test_delete.py

@@ -17,7 +17,7 @@ def test_delete():
     value = 2
     time_ = 10
 
-    node = DoubleLinkedListNode(value=_LinkedListValue(time_=time_, key=key))
+    node = DoubleLinkedListNode(value=_LinkedListValue(expire_at=time_, key=key))
     item = _DictValue(node=node, value=value)
     d._dict[key] = item
     d._ll_head = node
@@ -54,7 +54,7 @@ def test_delete__item_expired():
     value = 2
     time_ = 10
 
-    node = DoubleLinkedListNode(value=_LinkedListValue(time_=time_, key=key))
+    node = DoubleLinkedListNode(value=_LinkedListValue(expire_at=time_, key=key))
     d._dict[key] = _DictValue(node=node, value=value)
     d._ll_head = node
     d._ll_end = node

tests/test_get.py

@@ -10,20 +10,21 @@
 
 @pytest.mark.parametrize("update_ttl_on_get", [True, False])
 def test_get(update_ttl_on_get: bool):
-    d = TTLMap(ttl=timedelta(seconds=1000), update_ttl_on_get=update_ttl_on_get)
+    ttl = timedelta(seconds=1000)
+    d = TTLMap(ttl=ttl, update_ttl_on_get=update_ttl_on_get)
     lock_mock = LockMock()
     d._lock = lock_mock
     key = 1
     value = 2
     time_ = 10
 
-    node = DoubleLinkedListNode(value=_LinkedListValue(time_=time_, key=key))
+    node = DoubleLinkedListNode(value=_LinkedListValue(expire_at=time_, key=key))
     d._dict[key] = _DictValue(node=node, value=value)
     d._ll_head = node
     d._ll_end = node
 
     with (
-        patch("time.time", return_value=time_) as time_mock,
+        patch("time.monotonic", return_value=time_) as time_mock,
         patch.object(
             TTLMap,
             "_setitem",
@@ -35,7 +36,7 @@ def test_get(update_ttl_on_get: bool):
         time_mock.assert_called_once()
         update_by_ttl_mock.assert_called_once_with(current_time=time_)
         if update_ttl_on_get:
-            setitem_mock.assert_called_once_with(key, value, time_)
+            setitem_mock.assert_called_once_with(key, value, time_ + ttl.total_seconds())
         else:
             setitem_mock.assert_not_called()
 
@@ -54,11 +55,11 @@ def test_get__item_expired():
     value = 2
     time_ = 10
 
-    node = DoubleLinkedListNode(value=_LinkedListValue(time_=time_, key=key))
+    node = DoubleLinkedListNode(value=_LinkedListValue(expire_at=time_, key=key))
     d._dict[key] = _DictValue(node=node, value=value)
     d._ll_head = node
     d._ll_end = node
 
-    with patch("time.time", return_value=time_ + ttl.total_seconds() + 1):  # noqa: SIM117
+    with patch("time.monotonic", return_value=time_ + ttl.total_seconds() + 1):  # noqa: SIM117
         with pytest.raises(KeyError):
             _ = d[key]

tests/test_put_node_to_end.py

@@ -9,7 +9,7 @@ def test_put_node_to_end__empty_dict():
     d = TTLMap(ttl=timedelta(seconds=100))
     assert d._ll_head is None
     assert d._ll_end is None
-    node = DoubleLinkedListNode(value=_LinkedListValue(time_=1, key=1))
+    node = DoubleLinkedListNode(value=_LinkedListValue(expire_at=1, key=1))
     d._put_node_to_end(node=node)
 
     assert d._ll_head is node
@@ -31,7 +31,7 @@ def test_put_node_to_end__not_empty_dict():
     assert head_node is d._dict[1].node
     assert end_node is d._dict[2].node
 
-    node = DoubleLinkedListNode(value=_LinkedListValue(time_=1, key=3))
+    node = DoubleLinkedListNode(value=_LinkedListValue(expire_at=1, key=3))
     d._put_node_to_end(node=node)
 
     assert d._ll_head is head_node

tests/test_set.py

@@ -8,15 +8,16 @@
 
 
 def test_set__first():
-    d = TTLMap(ttl=timedelta(seconds=1000))
+    ttl = timedelta(seconds=1000)
+    d = TTLMap(ttl=ttl)
     lock_mock = LockMock()
     d._lock = lock_mock
     key = 1
     value = 2
     time_ = 10
-    expected_node = DoubleLinkedListNode(value=_LinkedListValue(time_=time_, key=key))
+    expected_node = DoubleLinkedListNode(value=_LinkedListValue(expire_at=time_ + ttl.total_seconds(), key=key))
     with (
-        patch("time.time", return_value=time_) as time_mock,
+        patch("time.monotonic", return_value=time_) as time_mock,
         patch.object(
             TTLMap,
             "_setitem",
@@ -35,7 +36,7 @@ def test_set__first():
     ):
         d[key] = value
         time_mock.assert_called_once()
-        setitem_mock.assert_called_once_with(key, value, time_)
+        setitem_mock.assert_called_once_with(key, value, time_ + ttl.total_seconds())
         update_by_ttl_mock.assert_called_once_with(current_time=time_)
         update_by_size_mock.assert_called_once()
         lock_mock.__enter__.assert_called_once()
@@ -47,7 +48,8 @@ def test_set__first():
 
 
 def test_set__second():
-    d = TTLMap(ttl=timedelta(seconds=1000))
+    ttl = timedelta(seconds=1000)
+    d = TTLMap(ttl=ttl)
     head_key = 1
     head_value = 2
     d[head_key] = head_value
@@ -57,9 +59,9 @@ def test_set__second():
     new_key = 5
     new_value = 20
     time_ = 10
-    expected_node = DoubleLinkedListNode(value=_LinkedListValue(time_=time_, key=new_key))
+    expected_node = DoubleLinkedListNode(value=_LinkedListValue(expire_at=time_ + ttl.total_seconds(), key=new_key))
     with (
-        patch("time.time", return_value=time_) as time_mock,
+        patch("time.monotonic", return_value=time_) as time_mock,
         patch.object(
             TTLMap,
             "_setitem",
@@ -78,7 +80,7 @@ def test_set__second():
     ):
         d[new_key] = new_value
         time_mock.assert_called_once()
-        setitem_mock.assert_called_once_with(new_key, new_value, time_)
+        setitem_mock.assert_called_once_with(new_key, new_value, time_ + ttl.total_seconds())
         update_by_ttl_mock.assert_called_once_with(current_time=time_)
         update_by_size_mock.assert_called_once()
         lock_mock.__enter__.assert_called_once()

tests/test_setitem.py

@@ -11,7 +11,7 @@ def test_setitem__new_item():
     time_ = 100.0
     key = 1
     value = 1
-    expected_node = DoubleLinkedListNode(value=_LinkedListValue(time_=time_, key=key))
+    expected_node = DoubleLinkedListNode(value=_LinkedListValue(expire_at=time_, key=key))
     with (
         patch.object(TTLMap, "_pop_ll_node") as mock_pop_ll_node,
         patch.object(
@@ -33,7 +33,7 @@ def test_setitem__existing_item():
     value = 1
     d[key] = value
     old_node = d._dict[key].node
-    expected_node = DoubleLinkedListNode(value=_LinkedListValue(time_=time_, key=key))
+    expected_node = DoubleLinkedListNode(value=_LinkedListValue(expire_at=time_, key=key))
     with (
         patch.object(TTLMap, "_pop_ll_node") as mock_pop_ll_node,
         patch.object(

tests/test_update_by_ttl.py

@@ -15,7 +15,7 @@ def test_update_by_ttl__empty_dict():
 def test_update_by_ttl__last_item():
     ttl = timedelta(seconds=100)
     d = TTLMap(ttl=ttl)
-    time_ = time.time() + ttl.total_seconds() + 1
+    time_ = time.monotonic() + ttl.total_seconds() + 1
     d._update_by_ttl(current_time=time_)
     assert d._ll_head is None
     assert d._ll_end is None
@@ -31,11 +31,11 @@ def test_update_by_ttl__expired_head():
     v2 = 2
     set_time_1 = 1
     set_time_2 = set_time_1 + ttl.total_seconds() + 1
-    with patch("time.time", side_effect=[set_time_1, set_time_2]):
+    with patch("time.monotonic", side_effect=[set_time_1, set_time_2]):
         d[k1] = v1
         d[k2] = v2
     d._update_by_ttl(current_time=set_time_2)
-    node_2 = DoubleLinkedListNode(value=_LinkedListValue(time_=set_time_2, key=k2))
+    node_2 = DoubleLinkedListNode(value=_LinkedListValue(expire_at=set_time_2 + ttl.total_seconds(), key=k2))
     assert d._ll_head == node_2
     assert d._ll_end == node_2
     assert d._dict == {k2: _DictValue(node=node_2, value=v2)}

ttlru_map/_ttl_map.py

@@ -19,13 +19,13 @@
 
 @dataclass(frozen=True)
 class _LinkedListValue(Generic[_TKey]):
-    __slots__ = ("key", "time_")
+    __slots__ = ("expire_at", "key")
 
-    time_: float
+    expire_at: float
     key: _TKey
 
     def __repr__(self) -> str:  # pragma: no cover
-        return f"{self.__class__.__name__}(time_={self.time_}, key={self.key})"
+        return f"{self.__class__.__name__}(expire_at={self.expire_at}, key={self.key})"
 
 
 @dataclass(frozen=True)
@@ -69,7 +69,7 @@ def __init__(
         self._ll_head: DoubleLinkedListNode[_LinkedListValue[_TKey]] | None = None
         self._ll_end: DoubleLinkedListNode[_LinkedListValue[_TKey]] | None = None
         self._max_size = max_size
-        self._ttl = ttl
+        self._ttl = ttl.total_seconds() if ttl is not None else None
         self._update_ttl_on_get = update_ttl_on_get
         self._lock = Lock()
 
@@ -97,9 +97,9 @@ def _update_by_ttl(self, current_time: float | None = None) -> None:
         """Remove items that have expired."""
         if self._ttl is None:
             return
-        current_time = current_time if current_time is not None else time.time()
+        current_time = current_time if current_time is not None else time.monotonic()
         while self._ll_head is not None:
-            if self._ll_head.value.time_ + self._ttl.total_seconds() >= current_time:
+            if self._ll_head.value.expire_at >= current_time:
                 break
             del self._dict[self._ll_head.value.key]
             self._pop_ll_node(self._ll_head)
@@ -136,9 +136,9 @@ def _put_node_to_end(self, node: DoubleLinkedListNode[_LinkedListValue[_TKey]])
             node.prev = self._ll_end
             self._ll_end = node
 
-    def _setitem(self, __key: _TKey, __value: _TValue, time_: float, /) -> None:
+    def _setitem(self, __key: _TKey, __value: _TValue, expire_at: float | None, /) -> None:
         """Set an item in the dictionary and put it to the end of the linked list."""
-        new_node = DoubleLinkedListNode(value=_LinkedListValue(time_=time_, key=__key))
+        new_node = DoubleLinkedListNode(value=_LinkedListValue(expire_at=expire_at, key=__key))
 
         if (item := self._dict.get(__key, None)) is not None:
             self._pop_ll_node(item.node)
@@ -155,9 +155,10 @@ def _delitem(self, item: _DictValue[_TKey, _TValue]) -> None:
 
     def __setitem__(self, __key: _TKey, __value: _TValue, /) -> None:
         with self._lock:
-            time_ = time.time()
-            self._setitem(__key, __value, time_)
-            self._update_by_ttl(current_time=time_)
+            current_time = time.monotonic()
+            expire_at = current_time + self._ttl if self._ttl is not None else None
+            self._setitem(__key, __value, expire_at)
+            self._update_by_ttl(current_time=current_time)
             self._update_by_size()
 
     def __delitem__(self, __key: _TKey, /) -> None:
@@ -168,11 +169,11 @@ def __delitem__(self, __key: _TKey, /) -> None:
 
     def __getitem__(self, __key: _TKey, /) -> _TValue:
         with self._lock:
-            time_ = time.time()
-            self._update_by_ttl(current_time=time_)
+            current_time = time.monotonic()
+            self._update_by_ttl(current_time=current_time)
             item = self._dict[__key].value
-            if self._update_ttl_on_get:
-                self._setitem(__key, item, time_)
+            if self._update_ttl_on_get and self._ttl is not None:
+                self._setitem(__key, item, current_time + self._ttl)
             return item
 
     def __len__(self) -> int: