Functions, Decorators, Context Managers & Dataclasses

orders = [
    {"order_id": 101, "status": "paid", "amount": 40.0},
    # ... orders 102 to 106 ...
]

def paid_total(orders):
    return sum(o["amount"] for o in orders if o["status"] == "paid")

steps = [paid_total]
print(steps[0](orders), paid_total.__name__)

def on_paid(order): return f"book {order['amount']}"
def on_refunded(order): return f"reverse {order['amount']}"
def on_pending(order): return "wait"

handlers = {"paid": on_paid, "refunded": on_refunded, "pending": on_pending}

for order in orders[2:5]:
    print(order["order_id"], handlers[order["status"]](order))

def make_counter():
    count = 0
    def add():
        nonlocal count
        count += 1
        return count
    return add

a = make_counter()
b = make_counter()
print(a(), a(), b())

checks = [lambda value: value > limit for limit in [10, 50]]
print([check(30) for check in checks])

checks = [lambda value, limit=limit: value > limit for limit in [10, 50]]
print([check(30) for check in checks])

import functools

def retry(fn):
    @functools.wraps(fn)
    def wrapper(*args, **kwargs):
        for attempt in range(3):
            try:
                return fn(*args, **kwargs)
            except ConnectionError:
                if attempt == 2:
                    raise
    return wrapper

calls = []

@retry
def load_order(order_id):
    """Load one order from the shop database."""
    calls.append(order_id)
    if len(calls) < 3:
        raise ConnectionError("database busy")
    return {"order_id": order_id, "status": "pending"}

print(load_order(105), len(calls))
print(load_order.__name__, "|", load_order.__doc__)

def bare(fn):
    def wrapper(*args, **kwargs):
        return fn(*args, **kwargs)
    return wrapper

@bare
def count_rows(rows):
    """Count the rows."""
    return len(rows)

print(count_rows.__name__, count_rows.__doc__)

def retries(times):                  # 1. takes the setting
    def decorate(fn):                # 2. takes the function
        @functools.wraps(fn)
        def wrapper(*args, **kwargs):  # 3. takes the call
            for attempt in range(times):
                try:
                    return fn(*args, **kwargs)
                except ConnectionError:
                    if attempt == times - 1:
                        raise
        return wrapper
    return decorate

tries = []

@retries(2)
def load_customer(customer_id):
    tries.append(customer_id)
    if len(tries) < 2:
        raise ConnectionError("database busy")
    return {"customer_id": customer_id, "name": "Carla"}

print(load_customer(3), len(tries))

from contextlib import contextmanager

log = []

@contextmanager
def batch(name):
    log.append(f"open {name}")
    try:
        yield []                      # what "as rows" receives
    finally:
        log.append(f"close {name}")   # runs however the block ends

try:
    with batch("2026-01-21") as rows:
        rows.append(106)
        raise RuntimeError("sink down")
except RuntimeError as error:
    print("raised:", error)

print(log)

class Batch:
    def __init__(self, name):
        self.name = name

    def __enter__(self):
        log.append(f"open {self.name}")
        return []

    def __exit__(self, exc_type, exc, tb):
        log.append(f"close {self.name}")
        return False                  # do not hide the error

row = {"order_id": 104, "amount": 15.5}
print(row.get("amuont"))

row["amuont"]

from dataclasses import dataclass

@dataclass
class Order:
    order_id: int
    status: str
    amount: float | None

order = Order(104, "paid", 15.5)
order.amuont

print(order)
print(order == Order(104, "paid", 15.5))

from dataclasses import field

@dataclass
class Bad:
    warnings: list = []

@dataclass
class CheckedOrder:
    order_id: int
    warnings: list = field(default_factory=list)

x = CheckedOrder(105)
y = CheckedOrder(106)
x.warnings.append("no amount")
print(x.warnings, y.warnings)

print(Order("104", "paid", "15,5"))

{order: 1}

@dataclass(frozen=True)
class ItemKey:
    order_id: int
    product: str

items = [ItemKey(101, "Pen"), ItemKey(101, "Notebook"), ItemKey(101, "Pen")]
print(len(set(items)), ItemKey(101, "Pen") == items[2])
print(items[0])

from typing import Protocol

class Sink(Protocol):
    def write(self, rows: list) -> int: ...

class FakeSink:                      # does not mention Sink
    def __init__(self):
        self.rows = []

    def write(self, rows):
        self.rows.extend(rows)
        return len(rows)

def flush(sink: Sink, rows):
    return sink.write(rows)

print(flush(FakeSink(), [105, 106]))

from abc import ABC, abstractmethod

class BaseSink(ABC):
    @abstractmethod
    def write(self, rows): ...

class HalfSink(BaseSink):
    pass

HalfSink()