diff --git a/tests/test_delete.py b/tests/test_delete.py index 94cf85b..bfe1ce3 100644 --- a/tests/test_delete.py +++ b/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 diff --git a/tests/test_get.py b/tests/test_get.py index 7cff549..95658e5 100644 --- a/tests/test_get.py +++ b/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] diff --git a/tests/test_put_node_to_end.py b/tests/test_put_node_to_end.py index 9834636..f9ab118 100644 --- a/tests/test_put_node_to_end.py +++ b/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 diff --git a/tests/test_set.py b/tests/test_set.py index 2369756..9171572 100644 --- a/tests/test_set.py +++ b/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() diff --git a/tests/test_setitem.py b/tests/test_setitem.py index fdad67e..122ec22 100644 --- a/tests/test_setitem.py +++ b/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( diff --git a/tests/test_update_by_ttl.py b/tests/test_update_by_ttl.py index 7c85b77..235a22e 100644 --- a/tests/test_update_by_ttl.py +++ b/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)} diff --git a/ttlru_map/_ttl_map.py b/ttlru_map/_ttl_map.py index 2198cd1..6e060d2 100644 --- a/ttlru_map/_ttl_map.py +++ b/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: