Facade, Strategy & Proxy

# BAD: every client repeats this flow and knows every service
validator, otp, tokens, audit = PasswordValidator(), OtpService(), TokenService(), AuditLogger()
if validator.validate(user, password):
    otp.send_otp(user)
    token = tokens.generate_token(user)
    audit.log(f"{user} logged in")

# --- subsystem: four independent services -------------------------------
class PasswordValidator:
    def validate(self, user: str, password: str) -> bool:
        return password == "secret"          # pretend DB check


class OtpService:
    def send_otp(self, user: str) -> None:
        print(f"OTP sent to {user}")


class TokenService:
    def generate_token(self, user: str) -> str:
        return f"token-{user}"


class AuditLogger:
    def log(self, msg: str) -> None:
        print(f"AUDIT: {msg}")


# --- the facade: one simple door in front of the subsystem ----------------
class AuthenticationFacade:
    def __init__(self):
        self.validator = PasswordValidator()
        self.otp = OtpService()
        self.tokens = TokenService()
        self.audit = AuditLogger()

    def login(self, user: str, password: str) -> str | None:
        if not self.validator.validate(user, password):   # step 1
            self.audit.log(f"failed login for {user}")
            return None
        self.otp.send_otp(user)                           # step 2
        token = self.tokens.generate_token(user)          # step 3
        self.audit.log(f"{user} logged in")               # step 4
        return token


# usage: the client knows ONE class and ONE method
auth = AuthenticationFacade()
print(auth.login("ganesh", "secret"))
# OTP sent to ganesh
# AUDIT: ganesh logged in
# token-ganesh

class Engine:
    def start(self) -> None: print("Engine started")

class FuelInjection:
    def inject(self) -> None: print("Fuel injected")

class TireController:
    def check(self) -> None: print("Tyre pressure OK")

class SafetyController:
    def arm(self) -> None: print("Airbags armed")


class CarController:                    # facade
    def __init__(self):
        self.engine = Engine()
        self.fuel = FuelInjection()
        self.tires = TireController()
        self.safety = SafetyController()

    def start(self) -> None:            # the order lives HERE, not in clients
        self.safety.arm()
        self.tires.check()
        self.fuel.inject()
        self.engine.start()


# usage
CarController().start()
# Airbags armed
# Tyre pressure OK
# Fuel injected
# Engine started

# BAD: every new payment method means editing this function
def pay(method: str, amount: int) -> None:
    if method == "card":
        print(f"Paid {amount} by credit card")
    elif method == "upi":
        print(f"Paid {amount} by UPI")
    elif method == "paypal":
        print(f"Paid {amount} by PayPal")

from abc import ABC, abstractmethod

class SortingStrategy(ABC):                 # the common contract
    @abstractmethod
    def sort(self, arr: list[int]) -> list[int]: ...


class MergeSort(SortingStrategy):
    def sort(self, arr: list[int]) -> list[int]:
        print("Sorting using Merge Sort")
        return sorted(arr)                  # real algorithm skipped for brevity


class QuickSort(SortingStrategy):
    def sort(self, arr: list[int]) -> list[int]:
        print("Sorting using Quick Sort")
        return sorted(arr)


class HeapSort(SortingStrategy):
    def sort(self, arr: list[int]) -> list[int]:
        print("Sorting using Heap Sort")
        return sorted(arr)


class SortLibrary:                          # the "context" (business logic)
    def __init__(self, strategy: SortingStrategy | None = None):
        self.strategy = strategy or MergeSort()   # sensible default

    def change_strategy(self, strategy: SortingStrategy) -> None:
        self.strategy = strategy
        print("Strategy changed")

    def sort(self, arr: list[int]) -> list[int]:
        return self.strategy.sort(arr)      # delegate; no if/else here


# usage
lib = SortLibrary(HeapSort())
print(lib.sort([3, 1, 2]))
lib.change_strategy(QuickSort())            # swapped at runtime
print(lib.sort([3, 1, 2]))
# Sorting using Heap Sort
# [1, 2, 3]
# Strategy changed
# Sorting using Quick Sort
# [1, 2, 3]

from abc import ABC, abstractmethod

class PaymentStrategy(ABC):
    @abstractmethod
    def pay(self, amount: int) -> None: ...


class CreditCard(PaymentStrategy):
    def pay(self, amount: int) -> None: print(f"Paid {amount} by credit card")

class PayPal(PaymentStrategy):
    def pay(self, amount: int) -> None: print(f"Paid {amount} by PayPal")

class UPI(PaymentStrategy):
    def pay(self, amount: int) -> None: print(f"Paid {amount} by UPI")


class PaymentContext:                       # business logic (STRATEGY user)
    def __init__(self, strategy: PaymentStrategy):
        self.strategy = strategy

    def set_strategy(self, strategy: PaymentStrategy) -> None:
        self.strategy = strategy

    def pay(self, amount: int) -> None:
        self.strategy.pay(amount)


class PaymentFactory:                       # FACTORY: only creates objects
    _registry = {"card": CreditCard, "paypal": PayPal, "upi": UPI}

    @classmethod
    def get_instance(cls, kind: str) -> PaymentStrategy:
        return cls._registry[kind]()


# usage: factory builds the object, strategy uses it
ctx = PaymentContext(PaymentFactory.get_instance("card"))
ctx.pay(500)
ctx.set_strategy(PaymentFactory.get_instance("upi"))
ctx.pay(200)
# Paid 500 by credit card
# Paid 200 by UPI

from abc import ABC, abstractmethod

class Document(ABC):                        # same interface for real + proxy
    @abstractmethod
    def read(self) -> str: ...


class RealDocument(Document):
    def __init__(self, name: str):
        print(f"Loading {name} from DB...")  # expensive
        self.name = name

    def read(self) -> str:
        return f"secret contents of {self.name}"


class DocumentProxy(Document):
    def __init__(self, name: str, role: str):
        self.name, self.role = name, role
        self._real: RealDocument | None = None   # not created yet

    def read(self) -> str:
        if self.role != "ADMIN":                       # protection proxy
            raise PermissionError("Only admin can read documents")
        if self._real is None:                         # virtual proxy (lazy)
            self._real = RealDocument(self.name)
        print(f"AUDIT: {self.role} read {self.name}")  # logging
        return self._real.read()                       # forward to real object


# usage
doc = DocumentProxy("salary.pdf", "ADMIN")
print(doc.read())
try:
    DocumentProxy("salary.pdf", "USER").read()
except PermissionError as e:
    print("Blocked:", e)
# Loading salary.pdf from DB...
# AUDIT: ADMIN read salary.pdf
# secret contents of salary.pdf
# Blocked: Only admin can read documents

import time
from abc import ABC, abstractmethod

class VideoService(ABC):                    # step 1: subject interface
    @abstractmethod
    def get_video(self, video_id: str) -> str: ...


class RealVideoService(VideoService):       # step 2: real, slow object
    def get_video(self, video_id: str) -> str:
        print("Fetching from remote server...")
        time.sleep(1)                       # simulate network latency
        return f"Video data for {video_id}"


class VideoServiceProxy(VideoService):      # step 3: proxy with a cache
    def __init__(self):
        self.real_service = RealVideoService()  # needed to forward misses
        self.cache: dict[str, str] = {}         # a dict is our "memory"

    def get_video(self, video_id: str) -> str:
        if video_id in self.cache:
            print("Returning from cache")
            return self.cache[video_id]
        data = self.real_service.get_video(video_id)  # cache miss
        self.cache[video_id] = data
        return data


# usage: client only knows the VideoService interface
service: VideoService = VideoServiceProxy()
print(service.get_video("abc"))   # slow
print(service.get_video("abc"))   # fast
print(service.get_video("xyz"))   # slow again (new id)
# Fetching from remote server...
# Video data for abc
# Returning from cache
# Video data for abc
# Fetching from remote server...
# Video data for xyz

from dataclasses import dataclass, field

@dataclass
class Wallet:
    user_id: str
    balance: int = 0
    transactions: list[str] = field(default_factory=list)


class WalletRepository:                     # data layer: a dict as the "DB"
    def __init__(self):
        self._wallets: dict[str, Wallet] = {}

    def get(self, user_id: str) -> Wallet:
        return self._wallets.setdefault(user_id, Wallet(user_id))


class WalletManager:                        # service layer: business rules
    def __init__(self, repo: WalletRepository):
        self.repo = repo

    def deposit(self, user_id: str, amount: int) -> None:
        w = self.repo.get(user_id)
        w.balance += amount
        w.transactions.append(f"+{amount}")

    def pay(self, user_id: str, amount: int) -> None:
        w = self.repo.get(user_id)
        if amount > w.balance:
            raise ValueError("Insufficient balance")
        w.balance -= amount
        w.transactions.append(f"-{amount}")

    def balance(self, user_id: str) -> int:
        return self.repo.get(user_id).balance


# usage
mgr = WalletManager(WalletRepository())
mgr.deposit("u1", 500)
mgr.pay("u1", 120)
print(mgr.balance("u1"), mgr.repo.get("u1").transactions)  # 380 ['+500', '-120']