Update Observer

Author: ELC

patterns/behavioural/observer/solution_04/__init__.py

@@ -0,0 +1,3 @@
+from .customer import Customer
+
+__all__ = ["Customer"]

patterns/behavioural/observer/solution_04/customer/__init__.py

@@ -0,0 +1,21 @@
+from dataclasses import dataclass, field
+from typing import List, Set
+
+from ..product import Product
+
+
+@dataclass
+class Customer:
+    name: str
+    interests: List[Product]
+    satisfied: bool = False
+    own_products: Set[Product] = field(default_factory=set)
+
+    def __call__(self, product: Product) -> None:
+        if product not in self.interests:
+            print(f"I'm {self.name} and I am not interested in {product.name}")
+            return
+
+        print(f"I'm {self.name} and I bought a {product.name}")
+        self.own_products.add(product)
+        self.satisfied = all(product in self.own_products for product in self.interests)

patterns/behavioural/observer/solution_04/customer/customer.py

@@ -0,0 +1,50 @@
+import time
+from .customer import Customer
+from .product import Product
+from .store import Store
+
+
+def main() -> None:
+    cellphone = Product("MobileX", "TechS", 300)
+    couch = Product("GiantSofa", "AllComfort", 800)
+    mug = Product("CoffeeM", "IndustrialTea", 20)
+
+    customers = [
+        Customer(name="John", interests=[cellphone]),
+        Customer(name="Mary", interests=[couch, mug]),
+        Customer(name="Alicia", interests=[cellphone, mug]),
+    ]
+
+    store = Store(name="AllYouNeed")
+
+    for customer in customers:
+        store.subscribe(customer, customer.interests)
+
+    late_customer = Customer(name="bob", interests=[cellphone, couch, mug])
+    late_customer_arrival = 3
+
+    customers.append(late_customer)
+
+    initial_time = time.time()
+
+    products = [cellphone, couch, mug]
+    estimated_time_arrivals = [1, 2, 4]
+
+    while True:
+        current_time = time.time() - initial_time
+
+        for product, eta in zip(products, estimated_time_arrivals):
+            if current_time >= eta and product not in store.products:
+                store.add_product(product)
+
+        if current_time >= late_customer_arrival and not store.is_subscribed(late_customer):
+            store.subscribe(late_customer, late_customer.interests)
+
+        if all(customer.satisfied for customer in customers):
+            break
+
+        time.sleep(0.05)  # To avoid blocking CPU
+
+
+if __name__ == "__main__":
+    main()

patterns/behavioural/observer/solution_04/main.py

@@ -0,0 +1,3 @@
+from .product import Product
+
+__all__ = ["Product"]

patterns/behavioural/observer/solution_04/product/__init__.py

@@ -0,0 +1,7 @@
+from typing import NamedTuple
+
+
+class Product(NamedTuple):
+    name: str
+    brand: str
+    price: float

patterns/behavioural/observer/solution_04/product/product.py

@@ -0,0 +1,3 @@
+from .store import Store
+
+__all__ = ["Store"]

patterns/behavioural/observer/solution_04/store/__init__.py

@@ -0,0 +1,38 @@
+from asyncio import Protocol
+from dataclasses import dataclass, field
+from typing import Callable, Dict, List, Set
+from collections import defaultdict
+
+from ..product import Product
+
+
+NOTIFY_FUNCTION = Callable[[Product], None]
+
+
+@dataclass
+class Store:
+    name: str
+    products: Set[Product] = field(default_factory=set)
+    product_notifiers: Dict[Product, List[NOTIFY_FUNCTION]] = field(
+        default_factory=lambda: defaultdict(list)
+    )
+
+    def subscribe(
+        self, notifier: NOTIFY_FUNCTION, relevant_products: List[Product]
+    ) -> None:
+        for product in relevant_products:
+            self.product_notifiers[product].append(notifier)
+            if product in self.products:
+                notifier(product)
+
+    def is_subscribed(self, notifier: NOTIFY_FUNCTION) -> bool:
+        all_notifiers: List[NOTIFY_FUNCTION] = sum(self.product_notifiers.values(), [])
+        return notifier in all_notifiers
+
+    def add_product(self, product: Product) -> None:
+        self.products.add(product)
+        self.notify(product)
+
+    def notify(self, product: Product) -> None:
+        for notifier in self.product_notifiers[product]:
+            notifier(product)