mirror of
https://github.com/invoke-ai/InvokeAI
synced 2024-08-30 20:32:17 +00:00
feat(item_storage): implement item_storage_memory with LRU eviction strategy
Implemented with OrderedDict.
This commit is contained in:
@ -1,3 +1,4 @@
|
||||
from collections import OrderedDict
|
||||
from contextlib import suppress
|
||||
from typing import Generic, Optional, TypeVar
|
||||
|
||||
@ -11,8 +12,7 @@ T = TypeVar("T", bound=BaseModel)
|
||||
class ItemStorageMemory(ItemStorageABC, Generic[T]):
|
||||
"""
|
||||
Provides a simple in-memory storage for items, with a maximum number of items to store.
|
||||
An item is deleted when the maximum number of items is reached and a new item is added.
|
||||
There is no guarantee about which item will be deleted.
|
||||
The storage uses the LRU strategy to evict items from storage when the max has been reached.
|
||||
"""
|
||||
|
||||
def __init__(self, id_field: str = "id", max_items: int = 10) -> None:
|
||||
@ -22,30 +22,29 @@ class ItemStorageMemory(ItemStorageABC, Generic[T]):
|
||||
if not id_field:
|
||||
raise ValueError("id_field must not be empty")
|
||||
self._id_field = id_field
|
||||
self._items: dict[str, T] = {}
|
||||
self._item_ids: set[str] = set()
|
||||
self._items: OrderedDict[str, T] = OrderedDict()
|
||||
self._max_items = max_items
|
||||
|
||||
def get(self, item_id: str) -> Optional[T]:
|
||||
# If the item exists, move it to the end of the OrderedDict.
|
||||
item = self._items.pop(item_id, None)
|
||||
if item is not None:
|
||||
self._items[item_id] = item
|
||||
return self._items.get(item_id)
|
||||
|
||||
def set(self, item: T) -> None:
|
||||
item_id = getattr(item, self._id_field)
|
||||
assert isinstance(item_id, str)
|
||||
if item_id in self._items or len(self._items) < self._max_items:
|
||||
# If the item is already stored, or we have room for more items, we can just add it.
|
||||
self._items[item_id] = item
|
||||
self._item_ids.add(item_id)
|
||||
else:
|
||||
# Otherwise, we need to make room for it first.
|
||||
self._items.pop(self._item_ids.pop())
|
||||
self._items[item_id] = item
|
||||
self._item_ids.add(item_id)
|
||||
if item_id in self._items:
|
||||
# If item already exists, remove it and add it to the end
|
||||
self._items.pop(item_id)
|
||||
elif len(self._items) >= self._max_items:
|
||||
# If cache is full, evict the least recently used item
|
||||
self._items.popitem(last=False)
|
||||
self._items[item_id] = item
|
||||
self._on_changed(item)
|
||||
|
||||
def delete(self, item_id: str) -> None:
|
||||
# Both of these are no-ops if the item doesn't exist.
|
||||
# This is a no-op if the item doesn't exist.
|
||||
with suppress(KeyError):
|
||||
del self._items[item_id]
|
||||
self._item_ids.remove(item_id)
|
||||
self._on_deleted(item_id)
|
||||
|
Reference in New Issue
Block a user