21 lines
685 B
Python
21 lines
685 B
Python
from __future__ import annotations
|
|
|
|
from collections import defaultdict
|
|
|
|
from app.modules.agent_api.domain.events.client_event import ClientEventRecord
|
|
|
|
|
|
class ReplayBuffer:
|
|
def __init__(self, limit: int = 200) -> None:
|
|
self._limit = limit
|
|
self._items: dict[str, list[ClientEventRecord]] = defaultdict(list)
|
|
|
|
def append(self, request_id: str, event: ClientEventRecord) -> None:
|
|
history = self._items[request_id]
|
|
history.append(event)
|
|
if len(history) > self._limit:
|
|
del history[: len(history) - self._limit]
|
|
|
|
def list(self, request_id: str) -> list[ClientEventRecord]:
|
|
return list(self._items.get(request_id, []))
|