50 lines
1.3 KiB
Python
50 lines
1.3 KiB
Python
from abc import ABC, abstractmethod
|
|
from collections import deque
|
|
from typing import Callable, Optional, Self, override
|
|
|
|
|
|
class Queue[T](ABC):
|
|
def __init__(self, size: int) -> None:
|
|
self.size = size
|
|
self.queue: deque[T] = deque()
|
|
|
|
def __len__(self) -> int:
|
|
return len(self.queue)
|
|
|
|
def __bool__(self) -> bool:
|
|
return bool(self.queue)
|
|
|
|
@abstractmethod
|
|
def add(self, item: T) -> None: pass
|
|
|
|
@abstractmethod
|
|
def pop(self) -> T: pass
|
|
|
|
class PermissiveQueue[T](Queue):
|
|
def __init__(self, size: int, overload_function: Optional[Callable[[Self], None]] = None) -> None:
|
|
super().__init__(size)
|
|
self._on_overload = overload_function
|
|
|
|
@override
|
|
def add(self, item: T) -> None:
|
|
self.queue.append(item)
|
|
if self._on_overload and len(self.queue) > self.size:
|
|
self._on_overload(self)
|
|
|
|
@override
|
|
def pop(self) -> T:
|
|
return self.queue.popleft()
|
|
|
|
class EnforcingQueue[T](Queue):
|
|
def __init__(self, size: int) -> None:
|
|
super().__init__(size)
|
|
|
|
@override
|
|
def add(self, item: T) -> None:
|
|
if len(self.queue) >= self.size:
|
|
raise IndexError
|
|
self.queue.append(item)
|
|
|
|
@override
|
|
def pop(self) -> T:
|
|
return self.queue.popleft() |