Online Shopping

  1  from __future__ import annotations
  2  from abc import ABC, abstractmethod
  3  from enum import Enum
  4  from dataclasses import dataclass, field
  5  from datetime import datetime
  6  from threading import Lock
  7  from typing import Protocol
  8  import uuid
  9  
 10  
 11  class OrderStatus(Enum):
 12      PENDING = "pending"
 13      CONFIRMED = "confirmed"
 14      SHIPPED = "shipped"
 15      DELIVERED = "delivered"
 16      CANCELLED = "cancelled"
 17  
 18  
 19  VALID_TRANSITIONS: dict[OrderStatus, set[OrderStatus]] = {
 20      OrderStatus.PENDING: {OrderStatus.CONFIRMED, OrderStatus.CANCELLED},
 21      OrderStatus.CONFIRMED: {OrderStatus.SHIPPED, OrderStatus.CANCELLED},
 22      OrderStatus.SHIPPED: {OrderStatus.DELIVERED},
 23      OrderStatus.DELIVERED: set(),
 24      OrderStatus.CANCELLED: set(),
 25  }
 26  
 27  
 28  @dataclass
 29  class Product:
 30      id: str
 31      name: str
 32      price: float
 33      description: str
 34      category: str
 35  
 36  
 37  class InventoryManager:
 38      def __init__(self):
 39          self._stock: dict[str, int] = {}
 40          self._reserved: dict[str, int] = {}
 41          self._lock = Lock()
 42          self._observers: list["InventoryObserver"] = []
 43          self._low_stock_threshold = 5
 44  
 45      def add_observer(self, observer: "InventoryObserver") -> None:
 46          self._observers.append(observer)
 47  
 48      def set_stock(self, product_id: str, quantity: int) -> None:
 49          with self._lock:
 50              self._stock[product_id] = quantity
 51              self._reserved.setdefault(product_id, 0)
 52  
 53      def get_available(self, product_id: str) -> int:
 54          with self._lock:
 55              total = self._stock.get(product_id, 0)
 56              reserved = self._reserved.get(product_id, 0)
 57              return total - reserved
 58  
 59      def reserve(self, product_id: str, quantity: int) -> bool:
 60          with self._lock:
 61              available = self._stock.get(product_id, 0) - self._reserved.get(product_id, 0)
 62              if available < quantity:
 63                  return False
 64              self._reserved[product_id] = self._reserved.get(product_id, 0) + quantity
 65              remaining = self._stock[product_id] - self._reserved[product_id]
 66              if remaining <= self._low_stock_threshold:
 67                  for obs in self._observers:
 68                      obs.on_low_stock(product_id, remaining)
 69              return True
 70  
 71      def release(self, product_id: str, quantity: int) -> None:
 72          with self._lock:
 73              self._reserved[product_id] = max(0, self._reserved.get(product_id, 0) - quantity)
 74  
 75      def commit(self, product_id: str, quantity: int) -> None:
 76          """Finalize a reservation by reducing actual stock."""
 77          with self._lock:
 78              self._stock[product_id] = max(0, self._stock.get(product_id, 0) - quantity)
 79              self._reserved[product_id] = max(0, self._reserved.get(product_id, 0) - quantity)
 80  
 81  
 82  class InventoryObserver(ABC):
 83      @abstractmethod
 84      def on_low_stock(self, product_id: str, remaining: int) -> None: ...
 85  
 86  
 87  class OrderObserver(ABC):
 88      @abstractmethod
 89      def on_status_change(self, order_id: str, user_id: str,
 90                           old_status: OrderStatus, new_status: OrderStatus) -> None: ...
 91  
 92  
 93  class ConsoleNotifier(InventoryObserver, OrderObserver):
 94      def on_low_stock(self, product_id: str, remaining: int) -> None:
 95          print(f"  [ALERT] Low stock for product {product_id}: {remaining} remaining")
 96  
 97      def on_status_change(self, order_id: str, user_id: str,
 98                           old_status: OrderStatus, new_status: OrderStatus) -> None:
 99          print(f"  [ORDER {order_id}] {old_status.value} -> {new_status.value} | user: {user_id}")
100  
101  
102  @dataclass
103  class CartItem:
104      product_id: str
105      quantity: int
106  
107  
108  class Cart:
109      def __init__(self, user_id: str):
110          self.user_id = user_id
111          self._items: dict[str, CartItem] = {}
112  
113      def add_item(self, product_id: str, quantity: int) -> None:
114          if quantity <= 0:
115              raise ValueError("Quantity must be positive")
116          if product_id in self._items:
117              self._items[product_id].quantity += quantity
118          else:
119              self._items[product_id] = CartItem(product_id, quantity)
120  
121      def remove_item(self, product_id: str) -> None:
122          if product_id not in self._items:
123              raise ValueError(f"Product {product_id} not in cart")
124          del self._items[product_id]
125  
126      def update_quantity(self, product_id: str, quantity: int) -> None:
127          if product_id not in self._items:
128              raise ValueError(f"Product {product_id} not in cart")
129          if quantity <= 0:
130              self.remove_item(product_id)
131          else:
132              self._items[product_id].quantity = quantity
133  
134      def get_items(self) -> list[CartItem]:
135          return list(self._items.values())
136  
137      def is_empty(self) -> bool:
138          return len(self._items) == 0
139  
140      def clear(self) -> None:
141          self._items.clear()
142  
143  
144  class DiscountStrategy(Protocol):
145      def apply(self, subtotal: float, code: str | None) -> float: ...
146  
147  
148  class NoDiscount:
149      def apply(self, subtotal: float, code: str | None) -> float:
150          return 0.0
151  
152  
153  class PercentageDiscount:
154      def __init__(self, valid_codes: dict[str, float]):
155          self._codes = valid_codes  # code -> percentage (0-100)
156  
157      def apply(self, subtotal: float, code: str | None) -> float:
158          if code and code in self._codes:
159              return round(subtotal * self._codes[code] / 100, 2)
160          return 0.0
161  
162  
163  class FlatDiscount:
164      def __init__(self, valid_codes: dict[str, float]):
165          self._codes = valid_codes  # code -> flat amount
166  
167      def apply(self, subtotal: float, code: str | None) -> float:
168          if code and code in self._codes:
169              return min(self._codes[code], subtotal)
170          return 0.0
171  
172  
173  @dataclass
174  class OrderLine:
175      product_id: str
176      product_name: str
177      unit_price: float
178      quantity: int
179  
180      @property
181      def line_total(self) -> float:
182          return round(self.unit_price * self.quantity, 2)
183  
184  
185  class Order:
186      def __init__(self, order_id: str, user_id: str, lines: list[OrderLine],
187                   subtotal: float, discount: float):
188          self.id = order_id
189          self.user_id = user_id
190          self.lines = list(lines)
191          self.subtotal = subtotal
192          self.discount = discount
193          self.total = round(subtotal - discount, 2)
194          self._status = OrderStatus.PENDING
195          self.created_at = datetime.now()
196          self._observers: list[OrderObserver] = []
197  
198      @property
199      def status(self) -> OrderStatus:
200          return self._status
201  
202      def add_observer(self, observer: OrderObserver) -> None:
203          self._observers.append(observer)
204  
205      def transition_to(self, new_status: OrderStatus) -> None:
206          if new_status not in VALID_TRANSITIONS[self._status]:
207              raise ValueError(
208                  f"Invalid transition from {self._status.value} to {new_status.value}")
209          old_status = self._status
210          self._status = new_status
211          for obs in self._observers:
212              obs.on_status_change(self.id, self.user_id, old_status, new_status)
213  
214  
215  class ShoppingService:
216      def __init__(self, discount_strategy: DiscountStrategy | None = None):
217          self._products: dict[str, Product] = {}
218          self._carts: dict[str, Cart] = {}
219          self._orders: dict[str, Order] = {}
220          self._inventory = InventoryManager()
221          self._discount = discount_strategy or NoDiscount()
222          self._notifier = ConsoleNotifier()
223          self._inventory.add_observer(self._notifier)
224          self._lock = Lock()
225  
226      def add_product(self, product: Product, stock: int) -> None:
227          self._products[product.id] = product
228          self._inventory.set_stock(product.id, stock)
229  
230      def browse_products(self) -> list[tuple[Product, int]]:
231          return [(p, self._inventory.get_available(p.id))
232                  for p in self._products.values()
233                  if self._inventory.get_available(p.id) > 0]
234  
235      def get_cart(self, user_id: str) -> Cart:
236          if user_id not in self._carts:
237              self._carts[user_id] = Cart(user_id)
238          return self._carts[user_id]
239  
240      def add_to_cart(self, user_id: str, product_id: str, quantity: int) -> Cart:
241          if product_id not in self._products:
242              raise ValueError(f"Product {product_id} not found")
243          cart = self.get_cart(user_id)
244          cart.add_item(product_id, quantity)
245          return cart
246  
247      def remove_from_cart(self, user_id: str, product_id: str) -> Cart:
248          cart = self.get_cart(user_id)
249          cart.remove_item(product_id)
250          return cart
251  
252      def checkout(self, user_id: str, coupon_code: str | None = None) -> Order:
253          with self._lock:
254              cart = self.get_cart(user_id)
255              if cart.is_empty():
256                  raise ValueError("Cart is empty")
257  
258              order_lines: list[OrderLine] = []
259              reserved_items: list[tuple[str, int]] = []
260  
261              try:
262                  for cart_item in cart.get_items():
263                      product = self._products.get(cart_item.product_id)
264                      if product is None:
265                          raise ValueError(f"Product {cart_item.product_id} no longer exists")
266  
267                      if not self._inventory.reserve(cart_item.product_id, cart_item.quantity):
268                          raise ValueError(
269                              f"Insufficient stock for '{product.name}'. "
270                              f"Available: {self._inventory.get_available(cart_item.product_id)}")
271                      reserved_items.append((cart_item.product_id, cart_item.quantity))
272  
273                      order_lines.append(OrderLine(
274                          product_id=product.id,
275                          product_name=product.name,
276                          unit_price=product.price,
277                          quantity=cart_item.quantity
278                      ))
279  
280              except ValueError:
281                  for pid, qty in reserved_items:
282                      self._inventory.release(pid, qty)
283                  raise
284  
285              subtotal = sum(line.line_total for line in order_lines)
286              discount = self._discount.apply(subtotal, coupon_code)
287              order_id = str(uuid.uuid4())[:8]
288              order = Order(order_id, user_id, order_lines, subtotal, discount)
289              order.add_observer(self._notifier)
290              self._orders[order_id] = order
291              cart.clear()
292              return order
293  
294      def confirm_order(self, order_id: str) -> None:
295          order = self._get_order(order_id)
296          order.transition_to(OrderStatus.CONFIRMED)
297          for line in order.lines:
298              self._inventory.commit(line.product_id, line.quantity)
299  
300      def ship_order(self, order_id: str) -> None:
301          order = self._get_order(order_id)
302          order.transition_to(OrderStatus.SHIPPED)
303  
304      def deliver_order(self, order_id: str) -> None:
305          order = self._get_order(order_id)
306          order.transition_to(OrderStatus.DELIVERED)
307  
308      def cancel_order(self, order_id: str) -> None:
309          with self._lock:
310              order = self._get_order(order_id)
311              order.transition_to(OrderStatus.CANCELLED)
312              for line in order.lines:
313                  self._inventory.release(line.product_id, line.quantity)
314  
315      def _get_order(self, order_id: str) -> Order:
316          order = self._orders.get(order_id)
317          if not order:
318              raise ValueError(f"Order {order_id} not found")
319          return order
320  
321  
322  if __name__ == "__main__":
323      discount = PercentageDiscount({"SAVE10": 10, "SAVE20": 20})
324      service = ShoppingService(discount_strategy=discount)
325  
326      service.add_product(Product("p1", "Wireless Headphones", 79.99, "Noise-cancelling over-ear", "Electronics"), 15)
327      service.add_product(Product("p2", "USB-C Cable", 12.99, "2m braided cable", "Accessories"), 50)
328      service.add_product(Product("p3", "Laptop Stand", 45.99, "Adjustable aluminum stand", "Accessories"), 3)
329      service.add_product(Product("p4", "Mechanical Keyboard", 129.99, "Cherry MX Blue switches", "Electronics"), 8)
330  
331      print("=== Browsing products ===")
332      for product, stock in service.browse_products():
333          print(f"  {product.name}: ${product.price:.2f} ({stock} in stock)")
334  
335      print("\n=== Building cart ===")
336      user_id = "user-alice"
337      service.add_to_cart(user_id, "p1", 1)
338      service.add_to_cart(user_id, "p2", 3)
339      service.add_to_cart(user_id, "p3", 1)
340      cart = service.get_cart(user_id)
341      for item in cart.get_items():
342          product = service._products[item.product_id]
343          print(f"  {product.name} x{item.quantity} = ${product.price * item.quantity:.2f}")
344  
345      print("\n=== Checkout with 10% discount ===")
346      order = service.checkout(user_id, coupon_code="SAVE10")
347      print(f"  Order {order.id}")
348      for line in order.lines:
349          print(f"    {line.product_name} x{line.quantity} = ${line.line_total:.2f}")
350      print(f"  Subtotal: ${order.subtotal:.2f}")
351      print(f"  Discount: -${order.discount:.2f}")
352      print(f"  Total: ${order.total:.2f}")
353  
354      print("\n=== Confirming order ===")
355      service.confirm_order(order.id)
356  
357      print("\n=== Shipping order ===")
358      service.ship_order(order.id)
359  
360      print("\n=== Delivering order ===")
361      service.deliver_order(order.id)
362  
363      print("\n=== Checking inventory after order ===")
364      for product, stock in service.browse_products():
365          print(f"  {product.name}: {stock} in stock")
366  
367      print("\n=== Testing oversell protection ===")
368      user2 = "user-bob"
369      service.add_to_cart(user2, "p3", 5)
370      try:
371          service.checkout(user2)
372      except ValueError as e:
373          print(f"  Correctly rejected: {e}")
374  
375      print("\n=== Testing order cancellation with inventory release ===")
376      user3 = "user-charlie"
377      service.add_to_cart(user3, "p4", 2)
378      order2 = service.checkout(user3)
379      print(f"  Placed order {order2.id} for 2 keyboards")
380      print(f"  Keyboard stock before cancel: {service._inventory.get_available('p4')}")
381      service.cancel_order(order2.id)
382      print(f"  Keyboard stock after cancel: {service._inventory.get_available('p4')}")