Overwriting Ring Buffer
Keep the newest N items in a fixed-size circular buffer, return the evicted value on overflow, and read the latest slice.
Code
from collections.abc import Iterator
from typing import Generic, TypeVar
T = TypeVar("T")
class RingBuffer(Generic[T]):
def __init__(self, capacity: int) -> None:
if capacity < 1:
raise ValueError("capacity must be positive")
self._slots: list[T | None] = [None] * capacity
self._head = 0
self._count = 0
@property
def capacity(self) -> int:
return len(self._slots)
def __len__(self) -> int:
return self._count
def push(self, item: T) -> T | None:
evicted: T | None = None
if self._count == self.capacity:
evicted = self._slots[self._head]
self._head = (self._head + 1) % self.capacity
index = (self._head + self._count - 1) % self.capacity
else:
index = (self._head + self._count) % self.capacity
self._slots[index] = item
if self._count < self.capacity:
self._count += 1
return evicted
def items(self) -> Iterator[T]:
for offset in range(self._count):
value = self._slots[(self._head + offset) % self.capacity]
if value is not None:
yield value
def latest(self, count: int) -> list[T]:
if count < 0:
raise ValueError("count must not be negative")
if count == 0:
return []
return list(self.items())[-count:]
def main() -> None:
buffer: RingBuffer[str] = RingBuffer(3)
for event in ["a", "b", "c"]:
print("push", event, "evicted", buffer.push(event))
print("full:", list(buffer.items()))
print("push d evicted:", buffer.push("d"))
print("after:", list(buffer.items()))
print("latest 2:", buffer.latest(2))
print("length:", len(buffer))
if __name__ == "__main__":
main()Report a problem with this exampleOpens a pre-filled issue on GitHub.