13 — Low-Level Design, In Depth#

class BankAccount:
    def __init__(self, balance: float = 0):
        self.__balance = balance          # name-mangled → _BankAccount__balance
 
    @property
    def balance(self) -> float:           # read-only public view
        return self.__balance
 
    def deposit(self, amount: float) -> None:
        if amount <= 0:
            raise ValueError("deposit must be positive")
        self.__balance += amount          # the INVARIANT is enforced here

from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from enum import Enum, auto
from typing import Protocol, Optional
from datetime import datetime
 
class VehicleType(Enum):
    MOTORCYCLE = auto()
    CAR = auto()
    TRUCK = auto()
 
@dataclass
class Vehicle:
    license_plate: str
    vehicle_type: VehicleType
    entry_time: Optional[datetime] = None
 
class PaymentGateway(ABC):                  # abstract base class
    @abstractmethod
    def charge(self, amount: float) -> bool: ...
 
class Notifier(Protocol):                    # structural typing — no inheritance
    def notify(self, message: str) -> None: ...

# VIOLATION: three reasons to change — schema, mail provider, persistence
class User:
    def __init__(self, name, email): ...
    def save_to_db(self): ...
    def send_welcome_email(self): ...
 
# FIX: separate the axes of change
@dataclass
class User:
    name: str
    email: str
 
class UserRepository:
    def save(self, user: User) -> None: ...
 
class EmailService:
    def send_welcome(self, user: User) -> None: ...

# VIOLATION: adding a shape means editing this function forever
def area(shape):
    if shape.type == "circle":   return 3.14 * shape.r ** 2
    elif shape.type == "square": return shape.side ** 2
    # ... and it grows without bound
 
# FIX: polymorphism — new shapes add code, they don't edit code
class Shape(ABC):
    @abstractmethod
    def area(self) -> float: ...
 
class Circle(Shape):
    def __init__(self, r): self.r = r
    def area(self): return 3.14159 * self.r ** 2

# VIOLATION — the classic
class Rectangle:
    def set_width(self, w): self.w = w
    def set_height(self, h): self.h = h
    def area(self): return self.w * self.h
 
class Square(Rectangle):
    def set_width(self, w):  self.w = self.h = w    # silently changes height
    def set_height(self, h): self.w = self.h = h
 
def resize(rect: Rectangle):
    rect.set_width(5); rect.set_height(4)
    assert rect.area() == 20        # FAILS for Square — 16

# VIOLATION
class Worker(ABC):
    @abstractmethod
    def work(self): ...
    @abstractmethod
    def eat(self): ...           # a Robot has no business implementing this
 
# FIX: split
class Workable(Protocol):
    def work(self) -> None: ...
class Eatable(Protocol):
    def eat(self) -> None: ...

class PricingStrategy(ABC):
    @abstractmethod
    def price(self, hours: int) -> float: ...
 
class HourlyPricing(PricingStrategy):
    def __init__(self, rate): self.rate = rate
    def price(self, hours): return hours * self.rate
 
class FlatDayPricing(PricingStrategy):
    def price(self, hours): return 20.0
 
class ProgressivePricing(PricingStrategy):
    """First 2 hours cheap, then more expensive."""
    def price(self, hours):
        return min(hours, 2) * 2.0 + max(0, hours - 2) * 5.0
 
class Ticket:
    def __init__(self, strategy: PricingStrategy):
        self.strategy = strategy
    def total(self, hours): return self.strategy.price(hours)

class Observer(Protocol):
    def update(self, event: str, data: dict) -> None: ...
 
class Subject:
    def __init__(self):
        self._observers: list[Observer] = []
 
    def subscribe(self, o: Observer):   self._observers.append(o)
    def unsubscribe(self, o: Observer): self._observers.remove(o)
 
    def notify(self, event: str, data: dict):
        for o in list(self._observers):     # copy — observers may unsubscribe
            o.update(event, data)
 
class ParkingLot(Subject):
    def park(self, vehicle):
        ...
        if self.is_full():
            self.notify("LOT_FULL", {"lot_id": self.id})
 
class DisplayBoard:
    def update(self, event, data):
        if event == "LOT_FULL":
            print("FULL")

class Coffee(ABC):
    @abstractmethod
    def cost(self) -> float: ...
    @abstractmethod
    def description(self) -> str: ...
 
class Espresso(Coffee):
    def cost(self): return 2.0
    def description(self): return "Espresso"
 
class CondimentDecorator(Coffee):
    def __init__(self, coffee: Coffee): self.coffee = coffee
 
class Milk(CondimentDecorator):
    def cost(self): return self.coffee.cost() + 0.5
    def description(self): return self.coffee.description() + " + Milk"
 
drink = Milk(Milk(Espresso()))     # 3.0, "Espresso + Milk + Milk"

from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from enum import Enum
from typing import Optional
import threading
import itertools
 
 
# ---------- Enums ----------
class VehicleType(Enum):
    MOTORCYCLE = 1
    CAR = 2
    TRUCK = 3
 
class SpotSize(Enum):
    MOTORCYCLE = 1
    COMPACT = 2
    LARGE = 3
 
 
# ---------- Domain objects ----------
@dataclass(frozen=True)
class Vehicle:
    license_plate: str
    vehicle_type: VehicleType
 
    def required_size(self) -> SpotSize:
        return {
            VehicleType.MOTORCYCLE: SpotSize.MOTORCYCLE,
            VehicleType.CAR:        SpotSize.COMPACT,
            VehicleType.TRUCK:      SpotSize.LARGE,
        }[self.vehicle_type]
 
 
class Spot:
    def __init__(self, spot_id: str, size: SpotSize, level: int):
        self.spot_id = spot_id
        self.size = size
        self.level = level
        self.vehicle: Optional[Vehicle] = None
 
    def is_free(self) -> bool:
        return self.vehicle is None
 
    def fits(self, vehicle: Vehicle) -> bool:
        # a vehicle fits any spot at least as large as it needs
        return self.size.value >= vehicle.required_size().value
 
    def occupy(self, vehicle: Vehicle) -> None:
        if not self.is_free():
            raise ValueError(f"spot {self.spot_id} already occupied")
        self.vehicle = vehicle
 
    def release(self) -> None:
        self.vehicle = None
 
 
@dataclass
class Ticket:
    ticket_id: str
    vehicle: Vehicle
    spot: Spot
    entry_time: datetime
    exit_time: Optional[datetime] = None
 
    def duration_hours(self) -> float:
        end = self.exit_time or datetime.now()
        return max((end - self.entry_time).total_seconds() / 3600, 0.0)
 
 
# ---------- Strategy: pricing ----------
class PricingStrategy(ABC):
    @abstractmethod
    def calculate(self, ticket: Ticket) -> float: ...
 
class HourlyPricing(PricingStrategy):
    RATES = {VehicleType.MOTORCYCLE: 1.0,
             VehicleType.CAR: 2.0,
             VehicleType.TRUCK: 3.5}
    def calculate(self, ticket: Ticket) -> float:
        import math
        hours = math.ceil(ticket.duration_hours())      # round up partial hours
        return hours * self.RATES[ticket.vehicle.vehicle_type]
 
class FlatDayPricing(PricingStrategy):
    def calculate(self, ticket: Ticket) -> float:
        return 20.0
 
 
# ---------- Observer: displays ----------
class LotObserver(ABC):
    @abstractmethod
    def on_event(self, event: str, payload: dict) -> None: ...
 
class DisplayBoard(LotObserver):
    def on_event(self, event, payload):
        if event == "OCCUPANCY_CHANGED":
            print(f"Available: {payload['available']}/{payload['total']}")
 
 
# ---------- Level ----------
class Level:
    def __init__(self, level_number: int, spots: list[Spot]):
        self.level_number = level_number
        self.spots = spots
        self._lock = threading.Lock()
 
    def find_and_occupy(self, vehicle: Vehicle) -> Optional[Spot]:
        """Smallest fitting free spot. Locked: find-and-claim must be atomic."""
        with self._lock:
            candidates = [s for s in self.spots if s.is_free() and s.fits(vehicle)]
            if not candidates:
                return None
            spot = min(candidates, key=lambda s: s.size.value)
            spot.occupy(vehicle)
            return spot
 
    def available_count(self) -> int:
        return sum(1 for s in self.spots if s.is_free())
 
 
# ---------- Aggregate root ----------
class ParkingLot:
    def __init__(self, levels: list[Level], pricing: PricingStrategy):
        self.levels = levels
        self.pricing = pricing                     # injected — Dependency Inversion
        self.active_tickets: dict[str, Ticket] = {}
        self._observers: list[LotObserver] = []
        self._ticket_seq = itertools.count(1)
        self._lock = threading.Lock()
 
    # --- observer plumbing ---
    def subscribe(self, o: LotObserver): self._observers.append(o)
    def _notify(self, event: str, payload: dict):
        for o in list(self._observers):
            o.on_event(event, payload)
 
    # --- core flows ---
    def park(self, vehicle: Vehicle) -> Ticket:
        for level in self.levels:
            spot = level.find_and_occupy(vehicle)
            if spot:
                ticket = Ticket(
                    ticket_id=f"T{next(self._ticket_seq):06d}",
                    vehicle=vehicle,
                    spot=spot,
                    entry_time=datetime.now(),
                )
                with self._lock:
                    self.active_tickets[ticket.ticket_id] = ticket
                self._notify("OCCUPANCY_CHANGED", self.occupancy())
                return ticket
        raise RuntimeError("parking lot full")
 
    def unpark(self, ticket_id: str) -> float:
        with self._lock:
            ticket = self.active_tickets.pop(ticket_id, None)
        if ticket is None:
            raise ValueError(f"unknown or already-closed ticket: {ticket_id}")
        ticket.exit_time = datetime.now()
        ticket.spot.release()
        fee = self.pricing.calculate(ticket)
        self._notify("OCCUPANCY_CHANGED", self.occupancy())
        return fee
 
    def occupancy(self) -> dict:
        total = sum(len(l.spots) for l in self.levels)
        available = sum(l.available_count() for l in self.levels)
        return {"total": total, "available": available}

import time, threading
 
class TokenBucket:
    """Allows bursts up to `capacity`, refills at `rate` tokens/second."""
    def __init__(self, capacity: int, refill_rate: float):
        self.capacity = capacity
        self.rate = refill_rate
        self.tokens = float(capacity)
        self.last = time.monotonic()          # monotonic — immune to clock changes
        self._lock = threading.Lock()
 
    def allow(self, cost: int = 1) -> bool:
        with self._lock:
            now = time.monotonic()
            self.tokens = min(self.capacity,
                              self.tokens + (now - self.last) * self.rate)
            self.last = now
            if self.tokens >= cost:
                self.tokens -= cost
                return True
            return False