Principles of Software Design: The Complete Guide¶

# TRAP 1: Over-engineering a simple problem
# BAD: building a factory + strategy + registry for something that needs an if/else
class DiscountStrategyRegistry:
    _strategies: dict[str, 'DiscountStrategy'] = {}

    @classmethod
    def register(cls, name: str, strategy: 'DiscountStrategy') -> None:
        cls._strategies[name] = strategy

    @classmethod
    def get(cls, name: str) -> 'DiscountStrategy':
        if name not in cls._strategies:
            raise KeyError(f"Unknown discount strategy: {name}")
        return cls._strategies[name]

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

class TenPercentDiscount(DiscountStrategy):
    def apply(self, price: float) -> float:
        return price * 0.90

class TwentyPercentDiscount(DiscountStrategy):
    def apply(self, price: float) -> float:
        return price * 0.80

# 4 classes, 30 lines to do what this does:

# GOOD: simple and clear — if requirements grow, refactor then
def apply_discount(price: float, discount_type: str) -> float:
    discounts = {"ten_percent": 0.90, "twenty_percent": 0.80}
    multiplier = discounts.get(discount_type, 1.0)
    return price * multiplier

# TRAP 2: Clever one-liners that nobody can read
# BAD: "clever" but unreadable
result = [x for x in (lambda l: (sorted(l), l)[1])(data)
          if (lambda v: v > 0 and v % 2 == 0)(x)]

# GOOD: readable — anyone can understand this in 5 seconds
def get_positive_even_numbers(numbers: list[int]) -> list[int]:
    return [n for n in numbers if n > 0 and n % 2 == 0]

result = get_positive_even_numbers(data)

# BAD: The shipping cost logic duplicated in two places
class OrderService:
    def calculate_order_total(self, order: Order) -> float:
        subtotal = sum(item.price * item.quantity for item in order.items)

        # Shipping logic here...
        if order.destination_country != "US":
            shipping = subtotal * 0.15  # 15% for international
        elif subtotal >= 50:
            shipping = 0               # free domestic over $50
        else:
            shipping = 5.99            # flat rate domestic

        return subtotal + shipping


class QuoteService:
    def generate_shipping_quote(self, cart: Cart) -> float:
        subtotal = sum(item.price * item.quantity for item in cart.items)

        # SAME shipping logic duplicated — if the rules change,
        # will you remember to update BOTH places?
        if cart.destination_country != "US":
            shipping = subtotal * 0.15
        elif subtotal >= 50:
            shipping = 0
        else:
            shipping = 5.99

        return shipping

# GOOD: One authoritative source for shipping logic
class ShippingCalculator:
    INTERNATIONAL_RATE = 0.15
    FREE_DOMESTIC_THRESHOLD = 50.00
    FLAT_DOMESTIC_RATE = 5.99

    def calculate(self, subtotal: float, destination_country: str) -> float:
        if destination_country != "US":
            return subtotal * self.INTERNATIONAL_RATE
        if subtotal >= self.FREE_DOMESTIC_THRESHOLD:
            return 0.0
        return self.FLAT_DOMESTIC_RATE


class OrderService:
    def __init__(self, shipping: ShippingCalculator):
        self._shipping = shipping

    def calculate_order_total(self, order: Order) -> float:
        subtotal = sum(item.price * item.quantity for item in order.items)
        return subtotal + self._shipping.calculate(subtotal, order.destination_country)


class QuoteService:
    def __init__(self, shipping: ShippingCalculator):
        self._shipping = shipping

    def generate_shipping_quote(self, cart: Cart) -> float:
        subtotal = sum(item.price * item.quantity for item in cart.items)
        return self._shipping.calculate(subtotal, cart.destination_country)

# This is NOT a DRY violation — same structure, different knowledge
def validate_username(username: str) -> bool:
    return len(username) >= 3 and username.isalnum()

def validate_product_code(code: str) -> bool:
    return len(code) >= 3 and code.isalnum()

# BAD: building a plugin system "just in case" we need extensibility
class NotificationService:
    def __init__(self):
        self._channels: dict[str, 'NotificationChannel'] = {}
        self._formatters: dict[str, 'MessageFormatter'] = {}
        self._rate_limiters: dict[str, 'RateLimiter'] = {}
        self._fallback_chain: list[str] = []
        # ... 200 more lines of infrastructure for requirements that don't exist

    def register_channel(self, name: str, channel: 'NotificationChannel') -> None: ...
    def register_formatter(self, name: str, formatter: 'MessageFormatter') -> None: ...
    def send(self, user_id: str, message: str, channel: str = "default") -> None: ...
    # etc.

# GOOD: build what you need, refactor when requirements actually arrive
class NotificationService:
    def __init__(self, smtp_client: SmtpClient):
        self._smtp = smtp_client

    def send_email(self, to: str, subject: str, body: str) -> None:
        self._smtp.send(to=to, subject=subject, body=body)

# BAD: one function does everything — business logic, data access, formatting
def process_order(order_id: str) -> str:
    # Data access concern
    conn = psycopg2.connect("postgresql://...")
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM orders WHERE id = %s", (order_id,))
    row = cursor.fetchone()

    # Business logic concern
    total = row['subtotal']
    if row['membership'] == 'gold':
        total *= 0.90
    if row['country'] != 'US':
        total += total * 0.15

    # Persistence concern
    cursor.execute(
        "UPDATE orders SET total = %s, status = 'processed' WHERE id = %s",
        (total, order_id)
    )
    conn.commit()

    # Formatting / presentation concern
    return f"<html><body>Order {order_id}: ${total:.2f} processed</body></html>"

# GOOD: each concern in its own layer

# Data access layer
class OrderRepository:
    def find_by_id(self, order_id: str) -> Order: ...
    def save(self, order: Order) -> None: ...

# Business logic layer (domain)
class OrderProcessor:
    def __init__(self, shipping: ShippingCalculator, membership: MembershipService):
        self._shipping = shipping
        self._membership = membership

    def process(self, order: Order) -> Order:
        discount = self._membership.get_discount(order.customer_id)
        shipping = self._shipping.calculate(order.subtotal, order.country)
        order.apply_discount(discount)
        order.set_shipping(shipping)
        order.mark_processed()
        return order

# Presentation layer
class OrderSerializer:
    def to_html(self, order: Order) -> str:
        return f"<html><body>Order {order.id}: ${order.total:.2f} processed</body></html>"

    def to_json(self, order: Order) -> dict:
        return {"order_id": order.id, "total": order.total, "status": order.status}

# BAD: this class has THREE reasons to change
class UserReport:
    def __init__(self, db):
        self._db = db

    def get_user_data(self, user_id: str) -> dict:
        # Database team could ask to change this
        return self._db.query("SELECT * FROM users WHERE id = ?", user_id)

    def format_report(self, user_data: dict) -> str:
        # Design/product team could ask to change this
        return f"User Report\n{'='*40}\n" + "\n".join(
            f"{k}: {v}" for k, v in user_data.items()
        )

    def send_report(self, email: str, report: str) -> None:
        # Infrastructure/ops team could ask to change this
        smtp = smtplib.SMTP("smtp.company.com")
        smtp.sendmail("reports@company.com", email, report)

# GOOD: one reason to change each
class UserRepository:
    """Only reason to change: how we store/retrieve users."""
    def __init__(self, db):
        self._db = db

    def find_by_id(self, user_id: str) -> dict:
        return self._db.query("SELECT * FROM users WHERE id = ?", user_id)


class UserReportFormatter:
    """Only reason to change: how the report looks."""
    def format(self, user_data: dict) -> str:
        return f"User Report\n{'='*40}\n" + "\n".join(
            f"{k}: {v}" for k, v in user_data.items()
        )


class EmailSender:
    """Only reason to change: how we send email."""
    def __init__(self, smtp_host: str):
        self._smtp_host = smtp_host

    def send(self, to: str, subject: str, body: str) -> None:
        smtp = smtplib.SMTP(self._smtp_host)
        smtp.sendmail("reports@company.com", to, body)


# The orchestrator — its only job is to coordinate
class UserReportService:
    def __init__(self, repo: UserRepository, formatter: UserReportFormatter,
                 sender: EmailSender):
        self._repo = repo
        self._formatter = formatter
        self._sender = sender

    def send_user_report(self, user_id: str, recipient_email: str) -> None:
        user_data = self._repo.find_by_id(user_id)
        report = self._formatter.format(user_data)
        self._sender.send(recipient_email, "User Report", report)

# BAD: adding a new shape requires modifying this function
def calculate_area(shape: dict) -> float:
    if shape['type'] == 'circle':
        return math.pi * shape['radius'] ** 2
    elif shape['type'] == 'rectangle':
        return shape['width'] * shape['height']
    elif shape['type'] == 'triangle':
        return 0.5 * shape['base'] * shape['height']
    # Every new shape means modifying this function — opening it to bugs
    elif shape['type'] == 'pentagon':
        ...

# GOOD: open for extension (add new shapes), closed for modification
from abc import ABC, abstractmethod
import math

class Shape(ABC):
    @abstractmethod
    def area(self) -> float: ...


class Circle(Shape):
    def __init__(self, radius: float):
        self.radius = radius

    def area(self) -> float:
        return math.pi * self.radius ** 2


class Rectangle(Shape):
    def __init__(self, width: float, height: float):
        self.width = width
        self.height = height

    def area(self) -> float:
        return self.width * self.height


class Triangle(Shape):
    def __init__(self, base: float, height: float):
        self.base = base
        self.height = height

    def area(self) -> float:
        return 0.5 * self.base * self.height


# Adding Pentagon requires ZERO changes to existing code:
class Pentagon(Shape):
    def __init__(self, side: float):
        self.side = side

    def area(self) -> float:
        return (self.side ** 2 * math.sqrt(25 + 10 * math.sqrt(5))) / 4


def calculate_total_area(shapes: list[Shape]) -> float:
    return sum(shape.area() for shape in shapes)

# BAD: adding PayPal or Apple Pay means modifying PaymentProcessor
class PaymentProcessor:
    def process(self, amount: float, method: str, details: dict) -> bool:
        if method == "credit_card":
            return self._charge_stripe(amount, details['card_token'])
        elif method == "bank_transfer":
            return self._initiate_ach(amount, details['account_number'])
        # Adding PayPal: modify this class and risk breaking Stripe + ACH

# GOOD: each payment method is a separate, addable implementation
class PaymentGateway(ABC):
    @abstractmethod
    def process(self, amount: float, details: dict) -> PaymentResult: ...

class StripeGateway(PaymentGateway):
    def process(self, amount: float, details: dict) -> PaymentResult:
        # Stripe-specific logic
        ...

class ACHGateway(PaymentGateway):
    def process(self, amount: float, details: dict) -> PaymentResult:
        # ACH-specific logic
        ...

class PayPalGateway(PaymentGateway):     # Added without modifying anything above
    def process(self, amount: float, details: dict) -> PaymentResult:
        # PayPal-specific logic
        ...

class PaymentProcessor:
    def __init__(self, gateway: PaymentGateway):    # depends on abstraction
        self._gateway = gateway

    def charge(self, amount: float, details: dict) -> PaymentResult:
        return self._gateway.process(amount, details)

# BAD: Square extends Rectangle — but violates LSP
class Rectangle:
    def __init__(self, width: float, height: float):
        self.width = width
        self.height = height

    def area(self) -> float:
        return self.width * self.height


class Square(Rectangle):
    def __init__(self, side: float):
        super().__init__(side, side)

    # "Square" must override setters to keep width == height
    @Rectangle.width.setter
    def width(self, value: float):
        self._width = value
        self._height = value   # forces height to match

    @Rectangle.height.setter
    def height(self, value: float):
        self._height = value
        self._width = value   # forces width to match


# This function works correctly with Rectangle
def double_width(shape: Rectangle) -> float:
    shape.width *= 2
    return shape.area()   # expects: width * height (unchanged)

r = Rectangle(4, 5)
print(double_width(r))    # 40.0 ✅ correct

s = Square(4)
print(double_width(s))    # 64.0 ❌ expected 40.0 — LSP violated!
# Substituting Square for Rectangle broke the function

# GOOD: Square and Rectangle share an interface, not an inheritance chain
class Shape(ABC):
    @abstractmethod
    def area(self) -> float: ...

class Rectangle(Shape):
    def __init__(self, width: float, height: float):
        self._width = width
        self._height = height

    def area(self) -> float:
        return self._width * self._height

class Square(Shape):
    def __init__(self, side: float):
        self._side = side

    def area(self) -> float:
        return self._side ** 2

# BAD: Penguin can't fly — inheriting from Bird violates LSP
class Bird:
    def fly(self) -> str:
        return "I am flying!"

class Penguin(Bird):
    def fly(self) -> str:
        raise NotImplementedError("Penguins can't fly!")   # LSP violation

def make_bird_fly(bird: Bird) -> str:
    return bird.fly()   # breaks when called with Penguin

# GOOD: model what birds actually have in common
class Bird(ABC):
    @abstractmethod
    def move(self) -> str: ...

class FlyingBird(Bird):
    def move(self) -> str:
        return "flying"

    def fly(self) -> str:
        return "I am flying!"

class SwimmingBird(Bird):
    def move(self) -> str:
        return "swimming"

class Eagle(FlyingBird): ...
class Duck(FlyingBird): ...    # ducks can also swim — add SwimmingBird mixin
class Penguin(SwimmingBird): ...

# BAD: one giant interface — not every device can do everything
class MultifunctionDevice(ABC):
    @abstractmethod
    def print_document(self, doc: Document) -> None: ...

    @abstractmethod
    def scan_document(self) -> Document: ...

    @abstractmethod
    def send_fax(self, doc: Document, number: str) -> None: ...

    @abstractmethod
    def copy_document(self, doc: Document) -> None: ...


class OldPrinter(MultifunctionDevice):
    def print_document(self, doc: Document) -> None:
        # actual implementation
        ...

    def scan_document(self) -> Document:
        raise NotImplementedError("This printer cannot scan")   # forced stub

    def send_fax(self, doc: Document, number: str) -> None:
        raise NotImplementedError("This printer cannot fax")    # forced stub

    def copy_document(self, doc: Document) -> None:
        raise NotImplementedError("This printer cannot copy")   # forced stub

# GOOD: small, specific interfaces — implement only what you can do
class Printer(ABC):
    @abstractmethod
    def print_document(self, doc: Document) -> None: ...

class Scanner(ABC):
    @abstractmethod
    def scan_document(self) -> Document: ...

class Fax(ABC):
    @abstractmethod
    def send_fax(self, doc: Document, number: str) -> None: ...


class OldPrinter(Printer):
    def print_document(self, doc: Document) -> None:
        # prints
        ...

class ModernMFD(Printer, Scanner, Fax):
    def print_document(self, doc: Document) -> None: ...
    def scan_document(self) -> Document: ...
    def send_fax(self, doc: Document, number: str) -> None: ...

# Python Protocol — structural typing, no explicit inheritance needed
from typing import Protocol

class Readable(Protocol):
    def read(self, n: int = -1) -> bytes: ...

class Writable(Protocol):
    def write(self, data: bytes) -> int: ...

class Seekable(Protocol):
    def seek(self, pos: int) -> int: ...
    def tell(self) -> int: ...


# Functions depend only on what they actually need
def compress(source: Readable, dest: Writable) -> None:
    # only needs read and write — not seek
    ...

def copy_section(source: Readable & Seekable, dest: Writable,
                 start: int, length: int) -> None:
    # needs read AND seek
    source.seek(start)
    data = source.read(length)
    dest.write(data)

# BAD: high-level module directly depends on low-level module
class OrderService:
    def __init__(self):
        # Hard dependency on specific implementation
        self._db = psycopg2.connect("postgresql://localhost/orders")
        self._emailer = smtplib.SMTP("smtp.sendgrid.net")

    def place_order(self, cart: Cart) -> str:
        # Uses _db directly — impossible to test without PostgreSQL
        cursor = self._db.cursor()
        cursor.execute("INSERT INTO orders ...")
        order_id = cursor.fetchone()[0]
        self._emailer.sendmail(...)
        return order_id

# GOOD: both high-level and low-level depend on abstractions
from abc import ABC, abstractmethod

# Abstractions (the boundary)
class OrderRepository(ABC):
    @abstractmethod
    def save(self, order: Order) -> str: ...

class EmailService(ABC):
    @abstractmethod
    def send_confirmation(self, order: Order, recipient: str) -> None: ...


# High-level module — depends only on abstractions
class OrderService:
    def __init__(self, order_repo: OrderRepository, email_service: EmailService):
        self._orders = order_repo
        self._email = email_service

    def place_order(self, cart: Cart, customer_email: str) -> str:
        order = Order.from_cart(cart)
        order_id = self._orders.save(order)
        self._email.send_confirmation(order, customer_email)
        return order_id


# Low-level implementations — depend on abstractions too
class PostgresOrderRepository(OrderRepository):
    def __init__(self, connection_string: str):
        self._conn = psycopg2.connect(connection_string)

    def save(self, order: Order) -> str:
        # PostgreSQL-specific implementation
        ...

class SendGridEmailService(EmailService):
    def send_confirmation(self, order: Order, recipient: str) -> None:
        # SendGrid-specific implementation
        ...

# For tests — no real database or email needed:
class InMemoryOrderRepository(OrderRepository):
    def __init__(self):
        self._store: dict[str, Order] = {}

    def save(self, order: Order) -> str:
        self._store[order.id] = order
        return order.id

class FakeEmailService(EmailService):
    def __init__(self):
        self.sent_confirmations: list[tuple[Order, str]] = []

    def send_confirmation(self, order: Order, recipient: str) -> None:
        self.sent_confirmations.append((order, recipient))

# Tests are now fast, isolated, and reliable
def test_place_order_sends_confirmation_email():
    repo = InMemoryOrderRepository()
    email = FakeEmailService()
    service = OrderService(repo, email)

    cart = Cart.with_items([CartItem("Book A", 29.99)])
    order_id = service.place_order(cart, "customer@example.com")

    assert order_id is not None
    assert len(email.sent_confirmations) == 1
    assert email.sent_confirmations[0][1] == "customer@example.com"

# BAD: violates Law of Demeter — "train wreck" chaining
class OrderService:
    def notify_customer(self, order: Order) -> None:
        # Reaching through order → customer → contact_info → email
        # OrderService now depends on: Order, Customer, ContactInfo, Email
        email = order.get_customer().get_contact_info().get_primary_email().get_address()
        self._email_client.send(email, "Your order is ready")

# GOOD: tell Order to handle the notification, or pass email directly
class Order:
    def get_customer_email(self) -> str:
        return self._customer.primary_email  # encapsulated internally

class OrderService:
    def notify_customer(self, order: Order) -> None:
        email = order.get_customer_email()   # one level deep
        self._email_client.send(email, "Your order is ready")

# BAD: inheritance hierarchy that gets tangled fast
class Animal:
    def breathe(self): ...

class Pet(Animal):
    def be_trained(self): ...

class Dog(Pet):
    def bark(self): ...

class RobotDog(???):    # Robot dogs don't breathe — now what?
                        # Can't extend Animal without getting breathe()
    def bark(self): ...


# GOOD: compose behaviors from smaller pieces
class Barker:
    def bark(self) -> str:
        return "Woof!"

class Trainer:
    def train(self, command: str) -> str:
        return f"Learning: {command}"

class Breather:
    def breathe(self) -> str:
        return "Inhale... exhale..."


class Dog:
    def __init__(self):
        self._barker = Barker()
        self._trainer = Trainer()
        self._breather = Breather()

    def bark(self) -> str: return self._barker.bark()
    def train(self, cmd: str) -> str: return self._trainer.train(cmd)
    def breathe(self) -> str: return self._breather.breathe()


class RobotDog:
    def __init__(self):
        self._barker = Barker()            # shares Barker behavior
        self._trainer = Trainer()          # shares training behavior
        # No Breather — robots don't breathe. No problem.

    def bark(self) -> str: return self._barker.bark()
    def train(self, cmd: str) -> str: return self._trainer.train(cmd)

# BAD: method both changes state AND returns data
class ShoppingCart:
    def remove_item_and_get_total(self, item_id: str) -> float:
        self._items = [i for i in self._items if i.id != item_id]
        return sum(i.price for i in self._items)   # mutation + query mixed

# BAD: "is_valid" sounds like a query but has side effects
class Form:
    def is_valid(self) -> bool:
        result = self._validate()
        self._save_validation_errors()   # hidden side effect!
        self._send_validation_event()    # caller is surprised
        return result


# GOOD: separate commands from queries
class ShoppingCart:
    def remove_item(self, item_id: str) -> None:        # Command: changes state
        self._items = [i for i in self._items if i.id != item_id]

    def total(self) -> float:                            # Query: reads state
        return sum(i.price for i in self._items)


class Form:
    def validate(self) -> None:                         # Command: runs validation
        self._errors = self._run_validation()

    def is_valid(self) -> bool:                         # Query: reads errors
        return len(self._errors) == 0

    def errors(self) -> list[str]:                      # Query: reads errors
        return list(self._errors)

# BAD: asking for data, making the decision externally
class OrderProcessor:
    def apply_discount(self, customer: Customer, order: Order) -> None:
        # Asking customer for its data, making the decision here
        if customer.membership_tier == "gold":
            if customer.total_purchases > 1000:
                order.discount = 0.20
            else:
                order.discount = 0.10
        elif customer.membership_tier == "silver":
            order.discount = 0.05

# GOOD: tell the customer to determine its own discount
class Customer:
    def applicable_discount(self) -> float:
        if self.membership_tier == "gold":
            return 0.20 if self.total_purchases > 1000 else 0.10
        if self.membership_tier == "silver":
            return 0.05
        return 0.0

class OrderProcessor:
    def apply_discount(self, customer: Customer, order: Order) -> None:
        order.apply_discount(customer.applicable_discount())   # tell, don't ask

# BAD: fails late — error is thrown deep in the call stack
def process_payment(amount, currency, user_id):
    # No validation here...
    payment = Payment(amount=amount, currency=currency)
    user = UserService.get(user_id)    # fails here if user_id is None
    # If we got here with amount=-100, we'll find out later in billing
    PaymentGateway.charge(user.payment_method, payment)

# GOOD: fails fast — errors caught at the entry point
def process_payment(amount: float, currency: str, user_id: str) -> None:
    # Validate at the boundary before doing anything
    if amount <= 0:
        raise ValueError(f"Payment amount must be positive, got: {amount}")
    if len(currency) != 3:
        raise ValueError(f"Currency must be ISO 3-letter code, got: {currency}")
    if not user_id:
        raise ValueError("user_id is required")

    payment = Payment(amount=amount, currency=currency)
    user = UserService.get(user_id)
    PaymentGateway.charge(user.payment_method, payment)

@dataclass(frozen=True)
class Money:
    amount: Decimal
    currency: str

    def __post_init__(self):
        # Fail fast in the constructor — no invalid Money objects can exist
        if self.amount < 0:
            raise ValueError(f"Money amount cannot be negative: {self.amount}")
        if len(self.currency) != 3:
            raise ValueError(f"Currency must be 3-letter ISO code: {self.currency}")

# You're adding a new feature to calculate_invoice()
# You notice: confusing variable names, magic numbers, no tests

# BEFORE (what you found)
def calculate_invoice(c, items, d=False):
    t = 0
    for i in items:
        t += i[0] * i[1]
    if d:
        t = t * 0.85
    if t > 500:
        t = t - 25
    return t

# AFTER (what you leave — naming, constants, clarity)
BULK_ORDER_THRESHOLD = 500.00
BULK_ORDER_DISCOUNT = 25.00
MEMBER_DISCOUNT_RATE = 0.15

def calculate_invoice(
    customer_id: str,
    line_items: list[tuple[float, int]],   # (unit_price, quantity)
    is_member: bool = False
) -> float:
    subtotal = sum(unit_price * quantity for unit_price, quantity in line_items)

    if is_member:
        subtotal *= (1 - MEMBER_DISCOUNT_RATE)

    if subtotal > BULK_ORDER_THRESHOLD:
        subtotal -= BULK_ORDER_DISCOUNT

    return subtotal

# BAD: assumes the external service always responds
class ProductService:
    def get_price(self, product_id: str) -> float:
        response = requests.get(f"{CATALOG_URL}/products/{product_id}")
        return response.json()['price']   # what if this times out? what if catalog is down?


# GOOD: circuit breaker + fallback + timeout
import tenacity
from circuitbreaker import circuit

class ProductService:
    def __init__(self, catalog_client: CatalogClient, cache: Cache):
        self._catalog = catalog_client
        self._cache = cache

    @circuit(failure_threshold=5, recovery_timeout=30)
    @tenacity.retry(
        stop=tenacity.stop_after_attempt(3),
        wait=tenacity.wait_exponential(min=1, max=8),
        retry=tenacity.retry_if_exception_type(RequestException),
    )
    def get_price(self, product_id: str) -> float:
        try:
            price = self._catalog.get_price(product_id)
            self._cache.set(f"price:{product_id}", price, ttl=300)
            return price
        except (RequestException, Timeout):
            # Fallback: serve cached value if catalog is unavailable
            cached = self._cache.get(f"price:{product_id}")
            if cached is not None:
                return cached
            raise   # propagate if no fallback available

# Logs + Metrics + Traces — the observability triad
import structlog
from opentelemetry import trace, metrics
import time

log = structlog.get_logger()
tracer = trace.get_tracer("order-service")
meter = metrics.get_meter("order-service")

# Metrics
orders_placed = meter.create_counter("orders.placed.total")
order_duration = meter.create_histogram("orders.duration.ms")
order_errors = meter.create_counter("orders.errors.total")

def place_order(cart_id: str, customer_id: str) -> str:
    start = time.time()

    with tracer.start_as_current_span("place_order") as span:
        span.set_attribute("cart.id", cart_id)
        span.set_attribute("customer.id", customer_id)

        log.info("placing_order", cart_id=cart_id, customer_id=customer_id)

        try:
            order = _do_place_order(cart_id, customer_id)

            orders_placed.add(1, {"status": "success"})
            log.info("order_placed", order_id=order.id, total=str(order.total))
            span.set_attribute("order.id", order.id)

            return order.id

        except Exception as e:
            order_errors.add(1, {"error_type": type(e).__name__})
            log.error("order_failed", cart_id=cart_id, error=str(e))
            span.record_exception(e)
            raise

        finally:
            duration_ms = (time.time() - start) * 1000
            order_duration.record(duration_ms)
            log.info("order_duration", duration_ms=duration_ms)

# BAD: double-processing an order charges the customer twice
class OrderProcessor:
    def process_payment(self, order_id: str, amount: float) -> None:
        stripe.Charge.create(amount=int(amount * 100), currency="usd")
        # If this method is called twice (retry after network error),
        # customer is charged twice. 😱


# GOOD: idempotency key prevents double-charging
class OrderProcessor:
    def process_payment(self, order_id: str, amount: float) -> str:
        # order_id as idempotency key: same order_id = same charge
        # Stripe deduplicates on the idempotency key
        charge = stripe.Charge.create(
            amount=int(amount * 100),
            currency="usd",
            idempotency_key=f"order-payment-{order_id}",  # safe to retry
        )
        return charge.id

# Database upsert as idempotency pattern
def upsert_inventory_reservation(order_id: str, product_id: str, qty: int) -> None:
    db.execute("""
        INSERT INTO inventory_reservations (order_id, product_id, quantity)
        VALUES (:order_id, :product_id, :qty)
        ON CONFLICT (order_id, product_id)
        DO UPDATE SET quantity = :qty, updated_at = NOW()
    """, {"order_id": order_id, "product_id": product_id, "qty": qty})
    # Safe to call multiple times — same result every time

# BAD: server-side session state — sticky sessions required
class OrderService:
    _user_sessions: dict = {}    # In-memory state — dies on restart

    def start_checkout(self, user_id: str, cart: dict) -> str:
        session_id = str(uuid4())
        self._user_sessions[session_id] = cart   # stored in THIS instance only
        return session_id

    def complete_checkout(self, session_id: str) -> None:
        cart = self._user_sessions.get(session_id)  # only works on same instance
        if not cart:
            raise SessionNotFoundError(session_id)


# GOOD: externalized state — any instance can serve any request
class OrderService:
    def __init__(self, session_store: Redis):
        self._sessions = session_store

    def start_checkout(self, user_id: str, cart: dict) -> str:
        session_id = str(uuid4())
        self._sessions.setex(
            f"checkout:{session_id}",
            3600,
            json.dumps(cart)
        )   # stored in Redis — any pod can read it
        return session_id

    def complete_checkout(self, session_id: str) -> None:
        data = self._sessions.get(f"checkout:{session_id}")
        if not data:
            raise SessionNotFoundError(session_id)
        cart = json.loads(data)
        # ... process order from cart