Closed timoniq closed 5 months ago
Suggested code:
class ShortStateStorage(dict[Identificator, ShortState[EventModel]]):
def __init__(self, *, maxlimit: int = 1000) -> None:
super().__init__()
self.maxlimit = maxlimit
self.queue: deque[Identificator] = deque(maxlen=maxlimit)
def add(self, id: Identificator, short_state: ShortState[EventModel]) -> None:
if len(self.queue) >= self.maxlimit:
self.pop(self.queue.popleft(), None)
if id not in self.queue:
self.queue.append(id)
self[id] = short_state
def pop(self, id: Identificator, default: T): # type: ignore
if id in self.queue:
self.queue.remove(id)
return super().pop(id, default)
Storage: typing.TypeAlias = dict[str, ShortStateStorage] # Store short state storage by view name
storage: Storage = {}
storage["MessageView"].add("some identificator", ShortState(...))
storage["MessageView"].pop("some identificator", None) # ShortState(...)
storage = {"CallbackView": ShortStateStorage(maxlimit=3)}
storage["CallbackView"].add(1, ShortState(...))
storage["CallbackView"].add(2, ShortState(...))
storage["CallbackView"].add(3, ShortState(...))
print(list(storage["CallbackView"].keys())) #> [1, 2, 3]
storage["CallbackView"].add(4, ShortState(...)) # Maximum limit exceeded, deleted ShortState with identificator 1
print(list(storage["CallbackView"].keys())) #> [2, 3, 4]
storage["CallbackView"].pop(1, None) # None
In order to avoid storing dead waiter states it is required to implement limited size dict to store short waiters that will delete old records on exceeding maximum size limit