Software Design Patterns & Architectures

from flask import Flask, request, jsonify
from flask_sqlalchemy import SQLAlchemy

# ── Everything in one application ──────────────────
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///app.db'
db  = SQLAlchemy(app)

# ── Model (Data Layer) lives here ──────────────────
class User(db.Model):
    id       = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True)
    email    = db.Column(db.String(120))

# ── Business Logic + Route (UI Layer) — all mixed ─
@app.route('/users', methods=['POST'])
def create_user():
    data = request.get_json()
    user = User(username=data['username'], email=data['email'])
    db.session.add(user)
    db.session.commit()
    return jsonify({'id': user.id}), 201

if __name__ == '__main__':
    app.run(debug=True)

from abc import ABC, abstractmethod

# ── Abstract Base Class — the contract ─────────────
class Animal(ABC):
    def __init__(self, name: str):
        self._name = name  # Encapsulation: private attribute

    @property
    def name(self) -> str:
        return self._name

    @abstractmethod
    def speak(self) -> str: ...

# ── Inheritance: Dog IS-A Animal ────────────────────
class Dog(Animal):
    def __init__(self, name: str, breed: str):
        super().__init__(name)     # inherit constructor
        self.breed = breed

    def speak(self) -> str:
        return f"Woof! I am {self.name}"

class Cat(Animal):
    def speak(self) -> str:
        return f"Meow! I am {self.name}"

# ── Polymorphism: same call, different behaviour ────
animals: list[Animal] = [Dog("Rex", "Labrador"), Cat("Whiskers")]
for a in animals:
    print(a.speak())   # → "Woof! I am Rex"  /  "Meow! I am Whiskers"

# ── Encapsulation: accessing only the public API ────
dog = Dog("Rex", "Labrador")
print(dog.name)      # ✓  via property
print(dog._name)     # ✗  bad practice — bypass encapsulation

from functools import reduce
from typing import Callable

# ── Pure function: same input → same output ─────────
def add(a: int, b: int) -> int:
    return a + b        # no side effects, no global reads

def square(x: int) -> int:
    return x ** 2

# ── Immutability — return new objects, never mutate ──
original = (1, 2, 3, 4, 5)           # tuple = immutable
doubled   = tuple(map(lambda x: x*2, original))
# original is unchanged: (1,2,3,4,5)

# ── Higher-order: map / filter ───────────────────────
nums     = [1,2,3,4,5,6,7,8]
evens    = list(filter(lambda n: n % 2 == 0, nums))
squared  = list(map(square, evens))
total    = reduce(add, squared)     # 4+16+36+64 = 120

# ── Function composition ─────────────────────────────
def compose(*fns: Callable) -> Callable:
    return lambda x: reduce(lambda v, f: f(v), fns, x)

transform = compose(
    lambda x: x * 2,
    lambda x: x + 10,
    str
)
print(transform(5))  # → "20"   (5*2=10, 10+10=20, str(20))

# ── Handling side effects — push to the edges ────────
def process_pipeline(raw_data: list) -> list:
    # Pure pipeline — no I/O inside
    return (list(map(square,
             filter(lambda x: x > 0, raw_data))))

# Side effect (I/O) only at the boundary
raw   = [int(x) for x in input("nums: ").split()]
result = process_pipeline(raw)
print(result)            # print = side effect, but isolated here

# ── Layer 4: Repository (Data Access) ──────────────
class UserRepository:
    def __init__(self, db):
        self.db = db

    def get_by_id(self, uid: int):
        return self.db.session.get(User, uid)

    def save(self, user) -> None:
        self.db.session.add(user)
        self.db.session.commit()

# ── Layer 2: Service (Business Logic + Use-case) ───
class UserService:
    def __init__(self, repo: UserRepository):
        self.repo = repo               # dependency injection

    def promote_to_admin(self, uid: int) -> dict:
        user = self.repo.get_by_id(uid)
        if not user:
            raise ValueError(f"User {uid} not found")
        user.role = "admin"            # business rule
        self.repo.save(user)
        return {"id": user.id, "role": user.role}

# ── Layer 1: Controller (Presentation) ─────────────
@app.route("/users/<int:uid>/promote", methods=["POST"])
def promote_user(uid: int):
    service = UserService(UserRepository(db))
    try:
        result = service.promote_to_admin(uid)
        return jsonify(result), 200
    except ValueError as e:
        return jsonify({"error": str(e)}), 404

from dataclasses import dataclass, field
from datetime   import datetime
from typing     import Callable
from collections import defaultdict

# ── Events: immutable facts about something that happened
@dataclass(frozen=True)       # frozen = immutable
class OrderCreated:
    order_id:   int
    user_id:    int
    total:      float
    created_at: datetime = field(default_factory=datetime.utcnow)

# ── In-process event bus (pub/sub) ───────────────────
class EventBus:
    def __init__(self):
        self._handlers: dict = defaultdict(list)

    def subscribe(self, event_type, handler: Callable):
        self._handlers[event_type].append(handler)

    def publish(self, event) -> None:
        for h in self._handlers[type(event)]:
            h(event)        # call each subscriber

# ── Producer: emits event after business logic ───────
class OrderService:
    def __init__(self, bus: EventBus):
        self.bus = bus

    def place_order(self, user_id: int, total: float) -> int:
        order_id = 42           # DB insert → returns id
        self.bus.publish(OrderCreated(order_id, user_id, total))
        return order_id         # ← does NOT know who listens

# ── Consumers: react to events independently ─────────
bus = EventBus()
bus.subscribe(OrderCreated,
    lambda e: print(f"📧 Email user {e.user_id}: order {e.order_id}"))
bus.subscribe(OrderCreated,
    lambda e: print(f"📦 Reserving stock for order {e.order_id}"))

svc = OrderService(bus)
svc.place_order(7, 149.99)

from typing   import Protocol
from dataclasses import dataclass

# ── Protocol = structural interface ────────────────
class SortStrategy(Protocol):
    def sort(self, data: list) -> list: ...

# ── Concrete strategies: each a different algorithm ─
class BubbleSort:
    def sort(self, data: list) -> list:
        d = data.copy()
        for i in range(len(d)):
            for j in range(len(d)-i-1):
                if d[j] > d[j+1]: d[j], d[j+1] = d[j+1], d[j]
        return d

class QuickSort:
    def sort(self, data: list) -> list:
        if len(data) <= 1: return data
        pivot = data[len(data)//2]
        left   = [x for x in data if x < pivot]
        middle = [x for x in data if x == pivot]
        right  = [x for x in data if x > pivot]
        return self.sort(left) + middle + self.sort(right)

# ── Context: uses a strategy without knowing its type
@dataclass
class Sorter:
    strategy: SortStrategy

    def sort(self, data: list) -> list:
        return self.strategy.sort(data)

# ── Swap strategy at runtime — no code changes ───────
data = [5, 2, 8, 1, 9]

sorter = Sorter(BubbleSort())
print(sorter.sort(data))           # [1,2,5,8,9] via bubble

sorter.strategy = QuickSort()     # swap — zero code change in Sorter
print(sorter.sort(data))           # [1,2,5,8,9] via quick

# ── S: Single Responsibility Principle ─────────────
class Invoice:
    def __init__(self, items: list, discount: float = 0.0):
        self.items    = items
        self.discount = discount

    def total(self) -> float:
        base = sum(i['price']*i['qty'] for i in self.items)
        return base * (1 - self.discount)  # only invoice maths

class InvoicePrinter:                 # separate concern: formatting
    def print_pdf(self, inv: Invoice): ...

class InvoiceRepository:             # separate concern: persistence
    def save(self, inv: Invoice): ...

# ── O: Open/Closed Principle ────────────────────────
from abc import ABC, abstractmethod

class Discount(ABC):
    @abstractmethod
    def apply(self, price: float) -> float: ...

class PercentDiscount(Discount):
    def __init__(self, pct: float): self.pct = pct
    def apply(self, p): return p * (1 - self.pct)

# To add "Fixed Discount" — ADD a class. DON'T modify Invoice.
class FixedDiscount(Discount):
    def __init__(self, amount: float): self.amount = amount
    def apply(self, p): return max(0, p - self.amount)

# ── D: Dependency Inversion Principle ───────────────
class InvoiceStorage(ABC):          # abstract = stable
    @abstractmethod
    def save(self, inv: Invoice): ...

class PostgresStorage(InvoiceStorage):
    def save(self, inv): ...  # write to DB

class InvoiceService:
    def __init__(self, storage: InvoiceStorage):
        self.storage = storage  # inject — don't construct

    def complete(self, inv: Invoice):
        self.storage.save(inv)  # depends on abstraction only

import uuid, time, threading
from dataclasses  import dataclass, field
from collections  import deque, defaultdict
from typing       import Any, Callable

# ── Message: immutable envelope ─────────────────────
@dataclass(frozen=True)
class Message:
    topic:      str
    payload:    Any
    message_id: str   = field(default_factory=lambda: str(uuid.uuid4()))
    retries:    int   = 0
    # frozen=True: once created, payload cannot change

# ── In-process Broker (simulates RabbitMQ / SQS) ────
class MessageBroker:
    MAX_RETRIES = 3

    def __init__(self):
        self._queues: dict = defaultdict(deque)   # topic → deque
        self._dlq:    deque = deque()              # dead-letter queue
        self._lock = threading.Lock()

    def publish(self, msg: Message) -> None:
        with self._lock:
            self._queues[msg.topic].append(msg)
        print(f"📤 Published [{msg.topic}] id={msg.message_id[:8]}")

    def consume(self, topic: str) -> Message | None:
        with self._lock:
            q = self._queues[topic]
            return q.popleft() if q else None

    def nack(self, msg: Message) -> None:
        # negative-ACK: requeue or send to DLQ
        if msg.retries < self.MAX_RETRIES:
            retried = Message(msg.topic, msg.payload,
                               msg.message_id, msg.retries+1)
            self._queues[msg.topic].appendleft(retried)   # front of queue
        else:
            self._dlq.append(msg)
            print(f"☠ DLQ: {msg.message_id[:8]} (failed {msg.retries}×)")

# ── Producer: fires and returns immediately ──────────
class OrderService:
    def __init__(self, broker: MessageBroker):
        self.broker = broker

    def place_order(self, user_id: int, item: str) -> str:
        order_id = str(uuid.uuid4())[:8]
        # ↓ async: hand off and return — don't wait
        self.broker.publish(Message(
            topic   = "orders",
            payload = {"order_id": order_id,
                       "user_id": user_id, "item": item}
        ))
        return order_id   # returns before any worker runs

# ── Worker: long-running consumer loop ───────────────
class OrderWorker:
    def __init__(self, broker: MessageBroker, name: str):
        self.broker, self.name = broker, name

    def run(self) -> None:
        print(f"⚙ {self.name} started")
        while True:
            msg = self.broker.consume("orders")
            if not msg:
                time.sleep(0.1)       # poll interval
                continue
            try:
                self._process(msg)
                print(f"✅ {self.name} ACK {msg.payload['order_id']}")
            except Exception as e:
                print(f"❌ {self.name} NACK: {e}")
                self.broker.nack(msg)    # requeue or DLQ

    def _process(self, msg) -> None:
        time.sleep(0.05)               # simulate DB write

# ── Dead Letter Queue: inspect failed messages ───────
def inspect_dlq(broker: MessageBroker) -> None:
    print(f"\n🔍 Dead Letter Queue ({len(broker._dlq)} messages):")
    for m in broker._dlq:
        print(f"  id={m.message_id[:8]}  retries={m.retries}  payload={m.payload}")

# ── Wire it all together ──────────────────────────────
broker  = MessageBroker()
service = OrderService(broker)

# Producers publish fast (sync):
for i in range(3):
    service.place_order(i, f"item-{i}")

# Workers run in background threads:
for name in ["worker-1", "worker-2"]:   # competing consumers
    t = threading.Thread(target=OrderWorker(broker, name).run, daemon=True)
    t.start()