Zen of Python — How Silent Errors Broke Authentication

# io.thecodeforge — Python tutorial

from decimal import Decimal
from enum import Enum
from dataclasses import dataclass
from typing import Optional


class Currency(Enum):
    USD = "USD"
    EUR = "EUR"
    GBP = "GBP"


@dataclass
class MoneyAmount:
    """
    Explicit value object — no implicit currency assumptions.
    Violating 'explicit is better than implicit' here caused a £0.01
    rounding error that accumulated to ~£4,000 across a quarter
    in a previous billing system that used raw floats with no currency tracking.
    """
    value: Decimal
    currency: Currency

    def __post_init__(self):
        # Enforce precision immediately — don't trust callers to round correctly
        if not isinstance(self.value, Decimal):
            raise TypeError(
                f"MoneyAmount.value must be Decimal, got {type(self.value).__name__}. "
                f"Implicit float conversion causes rounding drift over high transaction volumes."
            )


def apply_discount(
    original_price: MoneyAmount,
    discount_percentage: Decimal,
    *,
    allow_zero_result: bool = False,   # Explicit flag — caller must opt in consciously
) -> MoneyAmount:
    """
    'Explicit is better than implicit': every parameter that controls behaviour
    is named and typed. No *args magic that hides intent.
    """
    if not (Decimal("0") <= discount_percentage <= Decimal("100")):
        raise ValueError(
            f"discount_percentage must be between 0 and 100, got {discount_percentage}"
        )

    discount_multiplier = (Decimal("100") - discount_percentage) / Decimal("100")
    discounted_value = (original_price.value * discount_multiplier).quantize(
        Decimal("0.01")  # Explicit rounding to 2dp — no implicit banker's rounding surprises
    )

    if discounted_value == Decimal("0") and not allow_zero_result:
        raise ValueError(
            "Discount results in zero-value transaction. "
            "Pass allow_zero_result=True explicitly if this is intended (e.g. full comps)."
        )

    return MoneyAmount(value=discounted_value, currency=original_price.currency)


# --- Usage in a checkout service context ---

original = MoneyAmount(value=Decimal("49.99"), currency=Currency.GBP)
discounted = apply_discount(
    original_price=original,
    discount_percentage=Decimal("20"),
    # allow_zero_result not passed — caller doesn't need to opt in for a 20% discount
)

print(f"Original:   {original.currency.value} {original.value}")
print(f"Discounted: {discounted.currency.value} {discounted.value}")

# Demonstrate the guard against implicit float conversion
try:
    bad_amount = MoneyAmount(value=49.99, currency=Currency.GBP)  # float, not Decimal
except TypeError as e:
    print(f"\nCaught explicit type error: {e}")

# io.thecodeforge — Python tutorial

from dataclasses import dataclass, field
from typing import Optional
import json


# --- The 'flat is better than nested' failure in a real webhook handler ---

RAW_WEBHOOK_PAYLOAD = json.dumps({
    "event": "order.confirmed",
    "data": {
        "order": {
            "id": "ORD-8821",
            "user": {
                "id": "USR-441",
                "address": {
                    "billing": {
                        "postcode": "EC1A 1BB",
                        "country": "GB"
                    }
                }
            },
            "total_pence": 4999
        }
    }
})


# BAD: Nested access — every level is an implicit contract with the upstream API
def handle_webhook_dangerous(raw_payload: str) -> None:
    payload = json.loads(raw_payload)
    # If upstream ever changes the shape, this raises KeyError with no context
    # and the stack trace points here, not at the field that changed
    postcode = payload["data"]["order"]["user"]["address"]["billing"]["postcode"]
    order_id = payload["data"]["order"]["id"]
    print(f"Processing order {order_id} for postcode {postcode}")


# GOOD: Flat dataclass represents the shape explicitly — violations fail loudly at parsing time
@dataclass
class BillingAddress:
    postcode: str
    country: str


@dataclass
class OrderConfirmedEvent:
    order_id: str
    user_id: str
    billing_address: BillingAddress
    total_pence: int

    @classmethod
    def from_webhook_payload(cls, raw_payload: str) -> "OrderConfirmedEvent":
        """
        Parse and validate at the boundary — fail fast with a clear error
        rather than letting KeyError surface six function calls later.
        'Errors should never pass silently' applied at the data ingestion point.
        """
        try:
            payload = json.loads(raw_payload)
            order_data = payload["data"]["order"]
            billing_data = order_data["user"]["address"]["billing"]

            return cls(
                order_id=order_data["id"],
                user_id=order_data["user"]["id"],
                billing_address=BillingAddress(
                    postcode=billing_data["postcode"],
                    country=billing_data["country"],
                ),
                total_pence=order_data["total_pence"],
            )
        except KeyError as missing_field:
            # Name the missing field — not 'something went wrong'
            raise ValueError(
                f"Webhook payload missing required field: {missing_field}. "
                f"Check upstream API changelog — this shape may have changed."
            ) from missing_field


def handle_webhook_safe(raw_payload: str) -> None:
    event = OrderConfirmedEvent.from_webhook_payload(raw_payload)
    # All subsequent code is flat — no nested access, no KeyError risk
    print(f"Order ID:    {event.order_id}")
    print(f"User ID:     {event.user_id}")
    print(f"Postcode:    {event.billing_address.postcode}")
    print(f"Country:     {event.billing_address.country}")
    print(f"Total:       £{event.total_pence / 100:.2f}")


print("=== Safe handler output ===")
handle_webhook_safe(RAW_WEBHOOK_PAYLOAD)

# Simulate upstream API changing shape
BROKEN_PAYLOAD = json.dumps({"event": "order.confirmed", "data": {"order": {"id": "ORD-8822"}}})
print("\n=== Broken payload — explicit error ===")
try:
    handle_webhook_safe(BROKEN_PAYLOAD)
except ValueError as e:
    print(f"Caught: {e}")

# io.thecodeforge — Python tutorial

import logging
from dataclasses import dataclass, field
from decimal import Decimal
from typing import Optional

logging.basicConfig(level=logging.INFO, format="%(levelname)s | %(message)s")
logger = logging.getLogger(__name__)


@dataclass
class TransactionRecord:
    transaction_id: str
    amount_pence: int
    merchant_id: str


@dataclass
class ReconciliationResult:
    processed: list[TransactionRecord] = field(default_factory=list)
    explicitly_skipped: list[tuple[dict, str]] = field(default_factory=list)  # (record, reason)
    failed: list[tuple[dict, Exception]] = field(default_factory=list)        # (record, exception)


def parse_transaction_record(raw: dict) -> TransactionRecord:
    """Strict parser — raises on any malformed input rather than returning partial data."""
    required_fields = {"transaction_id", "amount_pence", "merchant_id"}
    missing = required_fields - raw.keys()
    if missing:
        raise ValueError(f"Missing required fields: {missing}")

    amount = raw["amount_pence"]
    if not isinstance(amount, int) or amount < 0:
        raise ValueError(
            f"amount_pence must be a non-negative int, got {type(amount).__name__}({amount!r})"
        )

    return TransactionRecord(
        transaction_id=raw["transaction_id"],
        amount_pence=amount,
        merchant_id=raw["merchant_id"],
    )


def reconcile_transactions(raw_records: list[dict]) -> ReconciliationResult:
    """
    'Errors should never pass silently' — every failure is categorised.
    'Unless explicitly silenced' — test-mode records are skipped intentionally and logged.
    """
    result = ReconciliationResult()

    for raw in raw_records:
        transaction_id = raw.get("transaction_id", "<unknown>")

        # Explicit, documented silencing — not a lazy except-pass
        # Test transactions from our QA pipeline share a known prefix.
        # Skipping them here is an intentional business rule, not error suppression.
        if transaction_id.startswith("TEST-"):
            result.explicitly_skipped.append((raw, "QA test transaction prefix — intentionally excluded from reconciliation"))
            logger.info("Skipped test transaction: %s", transaction_id)  # Visible in logs — not silent
            continue

        try:
            record = parse_transaction_record(raw)
            result.processed.append(record)
            logger.info("Processed: %s | £%.2f", record.transaction_id, record.amount_pence / 100)

        except ValueError as parse_error:
            # Capture the failure — don't swallow it. Downstream alerting can fire on result.failed
            result.failed.append((raw, parse_error))
            # Log at ERROR so your observability stack (Datadog, Grafana, whatever) catches it
            logger.error(
                "Failed to parse transaction %s: %s",
                transaction_id,
                parse_error,
            )
            # Do NOT re-raise — we want to process the remaining records.
            # But also do NOT silently continue — every failure is tracked.

    return result


# Simulate a realistic batch with one good record, one test record, and two malformed ones
raw_batch = [
    {"transaction_id": "TXN-001", "amount_pence": 4999, "merchant_id": "MCH-88"},
    {"transaction_id": "TEST-002", "amount_pence": 100, "merchant_id": "MCH-QA"},  # Test record
    {"transaction_id": "TXN-003", "amount_pence": -50, "merchant_id": "MCH-22"},  # Negative amount
    {"transaction_id": "TXN-004", "merchant_id": "MCH-11"},                       # Missing amount_pence
]

result = reconcile_transactions(raw_batch)

print(f"\n--- Reconciliation Summary ---")
print(f"Processed:          {len(result.processed)}")
print(f"Explicitly skipped: {len(result.explicitly_skipped)}")
print(f"Failed:             {len(result.failed)}")

if result.failed:
    print("\nFailed transactions (requires investigation):")
    for raw_record, error in result.failed:
        print(f"  {raw_record.get('transaction_id', '<unknown>')}: {error}")

# io.thecodeforge — Python tutorial

"""
Demonstrates 'there should be one obvious way to do it' and
'if the implementation is hard to explain, it's a bad idea'.

Scenario: a per-user rate limiter in an API gateway.
Two implementations — one that's hard to explain, one that isn't.
Both 'work'. One of them will survive contact with a new teammate.
"""

import time
from collections import deque
from threading import Lock
from typing import Optional


# --- Hard to explain: clever sliding window using a sentinel class and descriptor protocol ---
# (This is real code I've seen in production. It took me 20 minutes to understand.)

class _RateWindow:
    """Descriptor that magically injects a sliding window onto any class that uses it."""
    def __set_name__(self, owner, name):
        self._name = f"_rw_{name}"

    def __get__(self, obj, objtype=None):
        if obj is None:
            return self
        return getattr(obj, self._name, deque())

    def __set__(self, obj, value):
        setattr(obj, self._name, value)


class CleverRateLimiter:
    """Implements sliding window via descriptor magic. Explain this in an incident call."""
    window = _RateWindow()

    def __init__(self, max_requests: int, window_seconds: float):
        self.max_requests = max_requests
        self.window_seconds = window_seconds
        self._lock = Lock()

    def is_allowed(self, user_id: str) -> bool:
        with self._lock:
            # You need to understand descriptors, __get__, deques, AND the business logic
            # simultaneously to read this. That violates 'simple is better than complex'.
            now = time.monotonic()
            user_window = getattr(self, f"_rw_window_{user_id}", deque())
            while user_window and now - user_window[0] > self.window_seconds:
                user_window.popleft()
            if len(user_window) < self.max_requests:
                user_window.append(now)
                setattr(self, f"_rw_window_{user_id}", user_window)
                return True
            return False


# --- One obvious way: explicit sliding window, no descriptor magic ---
# Explain this to any dev in 10 seconds: "We keep a deque of timestamps per user,
# prune entries older than the window, and check if we're under the limit."

class SlidingWindowRateLimiter:
    """
    Per-user sliding window rate limiter.
    Stores a deque of request timestamps per user_id.
    Thread-safe via a single lock — acceptable for moderate throughput.
    For high concurrency, move to Redis with ZADD/ZREMRANGEBYSCORE.
    """

    def __init__(self, max_requests: int, window_seconds: float):
        self.max_requests = max_requests
        self.window_seconds = window_seconds
        # One deque per user — timestamps of allowed requests within the current window
        self._user_request_timestamps: dict[str, deque] = {}
        self._lock = Lock()  # Coarse lock — acceptable unless this becomes a bottleneck

    def is_allowed(self, user_id: str) -> bool:
        with self._lock:
            now = time.monotonic()

            if user_id not in self._user_request_timestamps:
                self._user_request_timestamps[user_id] = deque()

            user_window = self._user_request_timestamps[user_id]

            # Remove timestamps that have fallen outside the rolling window
            while user_window and now - user_window[0] > self.window_seconds:
                user_window.popleft()

            if len(user_window) < self.max_requests:
                user_window.append(now)  # Record this request's timestamp
                return True

            return False  # Window is full — caller should respond with HTTP 429


# --- Demonstrate both produce identical results but one is explainable ---

limiter = SlidingWindowRateLimiter(max_requests=3, window_seconds=5.0)

test_user = "USR-9912"
print("Sliding window rate limiter — 3 req / 5 sec")

for attempt in range(1, 6):
    allowed = limiter.is_allowed(test_user)
    status = "ALLOWED" if allowed else "DENIED (429)"
    print(f"  Request {attempt}: {status}")

print("\n(In production: denied requests return HTTP 429 with Retry-After header)")

// io.thecodeforge — python tutorial

// Ugly: a crime scene
x=1;y=2
if x==1:
    if y==2:
     print('yes')      // note: inconsistent indent

// Beautiful: obvious intent
is_ready = True
is_configured = True

if is_ready and is_configured:
    print('System operational')

// io.thecodeforge — python tutorial

// Complicated: don't do this
class PaymentProcessor:
    def __init__(self):
        self.handlers = {}
    
    def register(self, method, handler):
        self.handlers[method] = handler
    
    def process(self, payment):
        method = payment.get('method')
        if method not in self.handlers:
            raise ValueError()
        return self.handlers[method](payment)

// Complex: but only because the problem demands it
match payment["method"]:
    case "credit_card":
        charge_card(payment)
    case "paypal":
        redirect_to_paypal(payment)
    case _:
        raise ValueError(f"Unknown method: {payment['method']}")

// io.thecodeforge — python tutorial

BEFORE: 9 levels of nested conditionals tracking order status

def process_order_status(order):
    if order:
        if order.status:
            if order.status.name == 'shipped':
                if order.shipment:
                    if order.shipment.carrier:
                        # 4 more levels of hell
                        pass
    return None

AFTER: flat explicit mapping — one level, one purpose

STATUS_HANDLERS = {
    'shipped': lambda o: o.shipment.carrier.name,
    'pending': lambda o: 'awaiting_payment',
    'cancelled': lambda o: 'refund'
}

def process_order_status(order):
    handler = STATUS_HANDLERS.get(order.status.name)
    return handler(order) if handler else None

// io.thecodeforge — python tutorial

# BAD: ambiguous parameter demands a guess

def process(data, mode='auto'):
    if mode == 'auto':
        return infer_and_transform(data)  # guess the structure
    
# GOOD: explicit is better than implicit

from typing import Dict, List, Union

def process(
    user_data: Union[List[Dict], Dict],
    transformation: str = 'flatten'
) -> List[Dict]:
    """
    Transforms user data. Accepts a list of dicts or a single dict.
    transformation: 'flatten' | 'expand' — must be explicit.
    """
    match transformation:
        case 'flatten':
            return flatten_records(user_data)
        case 'expand':
            return expand_records(user_data)

// io.thecodeforge — python tutorial
import this
# Outputs the Zen of Python poem
# Example: "Beautiful is better than ugly."
# Use as a quick sanity check in code reviews

// io.thecodeforge — python tutorial
# Example: Applying Zen to refactor a nested dict
data = {"user": {"address": {"city": "NYC"}}}
# Flat alternative (Explicit > Implicit)
city = data.get("user", {}).get("address", {}).get("city", "unknown")
print(city)  # 'NYC'