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.

PythonData Integrity

Code

snippet
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()