from typing import Callable, Dict
from datetime import datetime
from app.pydantic_models import Product
# Type of messages sent to observers
MessageType = Dict[str, int]
# Type of observers' method: (MessageType -> None)
ObserverMethodType = Callable[[MessageType], None]
[docs]
class InventoryManager:
"""
Manage the product inventory using the Singleton design pattern.
"""
_instance = None # Instance for Singleton
_events: set[str] = {'decrement', 'threshold'} # Events for Observer
def __init__(self):
"""
Block direct instantiation.
Raises:
RuntimeError: If instantiated directly via the constructor, to force usage of get_instance().
"""
raise RuntimeError("You cannot use the public constructor `get_instance()` instead")
def _init(self):
"""
Initialize the inventory with a default product list.
"""
self._PRODUCTS: dict[int, Product] = {
1: Product(id=1, name="Wireless Mouse", price=13.99, stock=24, weight=0.2),
2: Product(id=2, name="USB-C Cable", price=4.49, stock=50, reorder_threshold=10, weight=0.08),
3: Product(id=3, name="Keyboard", price=29.00, stock=12, weight=0.8),
4: Product(id=4, name="Monitor", price=189.00, stock=6, weight=5.0),
5: Product(id=5, name="Notebook", price=1029.99, stock=2, reorder_threshold=1, weight=2)
}
# Create variable to store the observers.
# Keys are the events
# Values are dictionaries collecting observer (key) and method to invoke (value)
self._observers: dict[str, dict[object, ObserverMethodType]] = {event: {} for event in self._events}
[docs]
@staticmethod
def get_instance():
"""
Return the unique InventoryManager instance, creating it if needed.
Returns:
InventoryManager: The singleton instance.
"""
if InventoryManager._instance is None:
InventoryManager._instance = InventoryManager.__new__(InventoryManager)
InventoryManager._instance._init()
return InventoryManager._instance
[docs]
def register(self, event: str, observer: object, method: ObserverMethodType):
"""
Registers the observer `observer` for the event `event`.
Events description:
- `decrement`: triggered when a product's stock is decreased. Message: `{"pid": int, "amount": int (>0)}`
- `threshold`: triggered when a product's stock pass under its reorder threshold. Message: `{"pid": int}`
Args:
event: the concerned event
observer: the observer
method: the method to call when `event` happens. It should take one positional argument (message)
Raises:
ValueError: if `event` is not in `InventoryManager._events`.
"""
if event not in self._events:
raise ValueError(f"InventoryManager.register: argument `event` should be in {self._events}, but {event} found")
self._observers[event][observer] = method
[docs]
def unregister(self, event: str, observer: object):
"""
Unregisters `observer` from the event `event`.
Args:
event: the concerned event
observer: the observer
"""
if observer in self._observers[event]:
del self._observers[event][observer]
def _dispatch(self, event: str, message: MessageType):
"""
Dispatches the `message` to all observers subscribed to `event`.
Args:
event: the event that occurred
message: the associated message
"""
for method in self._observers[event].values():
method(message)
[docs]
def list_products(self) -> list[Product]:
"""
Return all products in the inventory.
Returns:
list[Product]: The list of all products.
"""
return list(self._PRODUCTS.values())
[docs]
def get_product(self, pid: int) -> Product | None:
"""
Returns a product by its identifier.
Args:
pid: The unique product identifier.
Returns:
Product | None: The matching product, or None if not found.
"""
return self._PRODUCTS.get(pid)
[docs]
def get_reorder_threshold(self, pid: int) -> int | None:
"""
Returns the reorder threshold of the product whose id is `pid`.
Args:
pid: the unique product identifier
Returns:
int | None: the matching reorder threshold, or None if not found
"""
product = self.get_product(pid)
if product is not None:
return product.reorder_threshold
[docs]
def has_stock(self, pid: int, qty: int) -> bool:
"""
Check whether a product has enough stock.
Args:
pid: The unique product identifier.
qty: The required quantity.
Returns:
bool: True if the product exists and has sufficient stock, False otherwise.
Raises:
ValueError: If qty is less than 1.
"""
if qty < 1:
raise ValueError("Quantity must be at least equal to 1")
p = self.get_product(pid)
return p is not None and p.stock >= qty
[docs]
def decrement_stock(self, pid: int, qty: int) -> None:
"""
Decrease the stock of a product.
Args:
pid: The unique product identifier.
qty: The quantity to decrement.
Raises:
ValueError: If the product is not found.
ValueError: If the stock is insufficient.
"""
# Check input
p = self.get_product(pid)
if p is None:
raise ValueError(f"Product {pid} not found")
if qty < 0:
raise ValueError("InventoryManager.decrement_stock: argument `qty` should be >= 0")
if p.stock < qty:
raise ValueError(f"Insufficient stock for product {pid}")
# Decrease stock
p.stock -= qty
# Dispatch events
self._dispatch("decrement", {"pid": pid, "amount": qty})
reorder_threshold: int = self.get_reorder_threshold(pid)
if p.stock <= reorder_threshold and not p.stock + qty <= reorder_threshold:
self._dispatch("threshold", {"pid": pid})
[docs]
class InventoryObserver:
"""
Observes and log the Inventory updates.
Implemented as a Singleton.
"""
_instance = None # Instance for Singleton
def __init__(self):
"""
Block direct instantiation.
Raises:
RuntimeError: If instantiated directly via the constructor, to force usage of get_instance().
"""
raise RuntimeError("You cannot use the public constructor `get_instance()` instead")
def _init(self):
"""Initialize a dictionary to store logs."""
# `self._logs[event]` will be the list of observed actions on `event`
self._logs: dict[str, list[str]] = {event: [] for event in InventoryManager._events}
[docs]
@staticmethod
def get_instance() -> 'InventoryObserver':
"""
Return the unique InventoryObserver instance, creating it if needed.
Returns:
InventoryObserver: The singleton instance.
"""
if InventoryObserver._instance is None:
InventoryObserver._instance = InventoryObserver.__new__(InventoryObserver)
InventoryObserver._instance._init()
return InventoryObserver._instance
[docs]
def subscribe(self):
"""Subscribes to the events of the InventoryManager"""
inventory_manager = InventoryManager.get_instance()
inventory_manager.register("decrement", self, self.decrement)
inventory_manager.register("threshold", self, self.threshold)
[docs]
def decrement(self, message: MessageType):
"""Handles notification from event `decrement`"""
pid = message["pid"]
amount = message["amount"]
product = InventoryManager.get_instance().get_product(pid)
if product is None:
name = "invalid pid"
else:
name = product.name
self._logs["decrement"].append(f"{datetime.now()} - Product #{pid} ({name}) decremented by {amount}")
[docs]
def threshold(self, message: MessageType):
"""Handles notification from event `threshold`"""
pid = message["pid"]
product = InventoryManager.get_instance().get_product(pid)
if product is None:
name = "invalid pid"
else:
name = product.name
self._logs["threshold"].append(f"{datetime.now()} - Stock for product #{pid} ({name}) is under reorder threshold")
[docs]
def get_logs(self) -> list[str]:
"""Returns the list of observed actions for the event `decrement`"""
return self._logs["decrement"]
[docs]
def get_reorders(self) -> list[str]:
"""Returns the list of observed actions for the event `threshold`"""
return self._logs["threshold"]