import this   # prints the Zen of Python — a philosophical README for the language

# The most important lines:
# "Beautiful is better than ugly."
# "Explicit is better than implicit."
# "Readability counts."
# "There should be one-- and preferably only one --obvious way to do it."
# "If the implementation is hard to explain, it's a bad idea."

# Five lines of Python that would take 50 in Java:
import pandas as pd

df = pd.read_csv("deals.csv")
summary = df.groupby("region")["revenue"].agg(["sum", "mean", "count"])
summary.sort_values("sum", ascending=False).to_csv("regional_summary.csv")

# Read CSV, group by region, compute sum/mean/count, sort, write — four lines.
# The equivalent in Java: 3 imports, a class, a main(), explicit file handling,
# a CSV parser, a sorting comparator, and a writer. ~60 lines.

# Live demo pattern: validate a customer's API during a discovery call
import httpx
import json

def probe_endpoint(url: str, token: str) -> None:
    """Hit an endpoint and report status, latency, and field presence."""
    with httpx.Client(timeout=10) as client:
        resp = client.get(url, headers={"Authorization": f"Bearer {token}"})
    print(f"Status : {resp.status_code}")
    print(f"Latency: {resp.elapsed.total_seconds() * 1000:.0f}ms")
    if resp.is_success:
        data = resp.json()
        print(f"Fields : {list(data.keys())}")
    else:
        print(f"Error  : {resp.text[:200]}")

# probe_endpoint("https://api.customer.io/v2/customers", token="...")

# ── 1. Install uv (macOS / Linux) ──────────────────────────────────
curl -LsSf https://astral.sh/uv/install.sh | sh

# Windows (PowerShell):
# powershell -c "irm https://astral.sh/uv/install.ps1 | iex"

# ── 2. Install a specific Python version ───────────────────────────
uv python install 3.13

# ── 3. Start a new project ─────────────────────────────────────────
uv init my-analysis          # creates directory, pyproject.toml, .python-version
cd my-analysis

# ── 4. Add dependencies ────────────────────────────────────────────
uv add pandas httpx           # resolves + installs + writes to pyproject.toml + lockfile

# ── 5. Run your script ─────────────────────────────────────────────
uv run main.py                # uses the project's managed venv automatically

[project]
name = "my-analysis"
version = "0.1.0"
requires-python = ">=3.13"    # minimum Python version
dependencies = [
    "pandas>=2.2",            # added by `uv add pandas`
    "httpx>=0.27",
]

[tool.uv]
# uv-specific settings (dev dependencies, source overrides, etc.)

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

# Add a production dependency:
uv add requests

# Add a development-only dependency (not shipped in production):
uv add --dev pytest ruff mypy

# Remove a dependency:
uv remove requests

# Sync environment to match lockfile (e.g. after a git pull):
uv sync

# Run a one-off script with temporary dependencies (no project needed):
uv run --with pandas python -c "import pandas; print(pandas.__version__)"

# Show the resolved dependency tree:
uv tree

# Update a specific package to the latest compatible version:
uv lock --upgrade-package pandas

# Ruff is the 2026 standard linter AND formatter (replaces flake8 + black + isort)
uv add --dev ruff

# Check for issues:
ruff check .

# Auto-fix what can be fixed:
ruff check --fix .

# Format code (replaces black):
ruff format .

# The ruff config goes in pyproject.toml:
# [tool.ruff]
# line-length = 88
# select = ["E", "F", "I"]  # pycodestyle, pyflakes, isort

import dis   # the disassembler module — peek at bytecode

def add(a, b):
    return a + b

dis.dis(add)   # print the bytecode instructions for add()

>>> 2 + 2          # Read the expression, Evaluate it...
4                  # ...Print the result, then Loop
>>> name = "Ada"   # assignment produces no output
>>> f"Hi, {name}"  # an f-string: builds text from the value
'Hi, Ada'
>>> _              # underscore holds the last result — useful shortcut
'Hi, Ada'
>>> import dis; dis.dis(lambda x: x * 2)   # explore bytecode live
  ...
>>> type(42), type("hello"), type([1,2])
(<class 'int'>, <class 'str'>, <class 'list'>)
>>> help(str.split)   # inline documentation for any object

import py_compile
import marshal
import pathlib

# Compile a .py file manually to a .pyc in __pycache__
py_compile.compile("main.py")

# List what's in __pycache__:
for p in pathlib.Path("__pycache__").iterdir():
    print(p.name, p.stat().st_size, "bytes")

# main.cpython-313.pyc — the bytecode for Python 3.13
# The filename encodes the Python version so switching versions
# doesn't accidentally load stale bytecode.

x = 42           # name 'x' binds to the int object 42
x = "hello"     # x re-binds to a str object; 42 is now unreferenced
y = x            # y binds to the SAME object x points at

print(id(x))               # memory address of the object
print(id(y))               # same address — one object, two names
print(id(x) == id(y))     # True
print(x is y)             # True — 'is' checks identity (same object)

# Rebind x — y is unaffected:
x = "world"
print(y)                  # still "hello" — the object didn't change

a = [1, 2, 3]
b = a            # b points at THE SAME list object

b.append(4)       # mutates the shared object
print(a)         # [1, 2, 3, 4] — a sees the change
print(a is b)    # True

# To get an INDEPENDENT copy, copy explicitly:
c = a.copy()       # or: list(a), or a[:]
c.append(99)
print(a)         # [1, 2, 3, 4] — unchanged
print(c)         # [1, 2, 3, 4, 99]
print(a is c)    # False — different objects now

import copy

original = [[1, 2], [3, 4]]
shallow  = original.copy()    # outer list is new; inner lists are shared
deep     = copy.deepcopy(original)

original[0].append(99)

print(original)   # [[1, 2, 99], [3, 4]]
print(shallow)    # [[1, 2, 99], [3, 4]] — inner list shared, sees change
print(deep)       # [[1, 2], [3, 4]]    — fully independent

value = 42
print(type(value))          # <class 'int'>

value = "hello"             # re-bind to a str — perfectly valid
print(type(value))          # <class 'str'>

# Strong typing: operations must make sense
try:
    result = "age: " + 7     # TypeError — can't add str and int
except TypeError as e:
    print(e)                 # can only concatenate str (not "int") to str

result = "age: " + str(7)  # fix: explicit conversion
print(result)               # age: 7

# isinstance() for runtime type checks:
print(isinstance(value, str))         # True
print(isinstance(value, (str, int)))  # True — accepts a tuple of types

a = [1, 2, 3]
b = [1, 2, 3]   # same value, different object
print(a == b)   # True  — same value
print(a is b)  # False — different objects

# None is a singleton — exactly one object exists
x = None
print(x is None)    # True  — correct idiom
print(x == None)   # True  — works, but not idiomatic; avoids __eq__ override

# Small integer caching — CPython caches -5 to 256
p = 256; q = 256
print(p is q)       # True  (cached object)
r = 257; s = 257
print(r is s)       # False or True (implementation-dependent!)
# ⚠ Never use 'is' to compare integers — this is a known interview trap

# Arithmetic operators
print(7 + 2)     # 9   — add
print(7 - 2)     # 5   — subtract
print(7 * 2)     # 14  — multiply
print(7 / 2)     # 3.5 — true division ALWAYS returns float
print(7 // 2)    # 3   — floor division: rounds toward negative infinity
print(7 % 2)     # 1   — modulo (remainder)
print(7 ** 2)    # 49  — exponentiation
print(-7 // 2)   # -4  — floor division: rounds DOWN, not toward zero!

# Arbitrary precision in action:
print(2 ** 100)  # 1267650600228229401496703205376 — no overflow

# Integer bit operations (useful for flags and binary data):
print(0b1010 & 0b1100)  # 8  — bitwise AND
print(0b1010 | 0b1100)  # 14 — bitwise OR
print(1 << 4)            # 16 — left shift (multiply by 2^4)

# The classic float trap:
print(0.1 + 0.2)           # 0.30000000000000004
print(0.1 + 0.2 == 0.3)    # False!

# Workaround 1: math.isclose() for approximate comparisons
import math
print(math.isclose(0.1 + 0.2, 0.3))    # True — within relative tolerance

# Workaround 2: Decimal for exact arithmetic (use for money)
from decimal import Decimal, ROUND_HALF_UP

price    = Decimal("19.99")    # always pass a STRING to Decimal
tax_rate = Decimal("0.08")
tax      = price * tax_rate
total    = price + tax
total_rounded = total.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)

print(f"Tax:   ${tax}")          # 1.5992 — exact
print(f"Total: ${total_rounded}") # 21.59

# Workaround 3: store integer cents
price_cents = 1999    # $19.99 stored as cents
tax_cents   = price_cents * 8 // 100  # integer division
print(f"Total: ${(price_cents + tax_cents) / 100:.2f}")   # $21.59

# String literals
s1 = "Hello, world"
s2 = 'same thing'
s3 = """multi
line
string"""
path = r"C:\Users\Ada\Documents"   # raw: backslashes are literal

# f-string format mini-language:
name, revenue, pct = "Acme", 1_234_567.89, 0.1234

print(f"{name:<12}")            # "Acme        " (left-align, width 12)
print(f"${revenue:,.2f}")       # "$1,234,567.89"
print(f"{pct:.1%}")             # "12.3%"
print(f"{name=}")               # "name='Acme'" — debug form (3.8+)
print(f"{2 ** 10:,}")           # "1,024" (any expression in braces)

s = "  Hello, Solutions World  "

# Cleaning:
print(s.strip())               # "Hello, Solutions World"
print(s.lower())               # "  hello, solutions world  "
print(s.upper())               # "  HELLO, SOLUTIONS WORLD  "

# Searching and replacing:
print(s.strip().startswith("Hello"))   # True
print("Solutions" in s)          # True (membership test)
print(s.replace("World", "Team"))  # new string — immutable!

# Splitting and joining:
words = "Acme,Globex,Initech".split(",")   # ["Acme", "Globex", "Initech"]
print(" | ".join(words))               # "Acme | Globex | Initech"

# Slicing (same syntax as lists):
text = "solutions"
print(text[0:4])     # "solu"
print(text[-4:])     # "ions"
print(text[::-1])    # "snoitulos"  (reversed)

text  = "café"                       # str — Unicode text
bdata = text.encode("utf-8")      # str → bytes: b'caf\xc3\xa9'
back  = bdata.decode("utf-8")     # bytes → str: "café"

print(type(bdata))   # <class 'bytes'>
print(bdata)         # b'caf\xc3\xa9'
print(back)          # café

# File modes:
# open("f.txt", "r")  → yields str (text mode, decodes via locale)
# open("f.bin", "rb") → yields bytes (binary mode, no decoding)

# Rule: use UTF-8 everywhere unless you have a specific reason not to.
# Always specify encoding explicitly when opening text files:
with open("data.txt", "r", encoding="utf-8") as f:
    content = f.read()

# Function annotations: parameters and return type
def discount(price: float, pct: float) -> float:
    """Return price after applying a percentage discount."""
    return price * (1 - pct / 100)

# Variable annotations:
name: str = "Acme"
count: int = 0
active: bool = True

# Modern composite types (Python 3.9+ syntax, no imports needed):
names: list[str]          = ["Acme", "Globex"]
scores: dict[str, int]    = {"Ada": 95}
point: tuple[int, int]    = (10, 20)
tags: set[str]            = {"enterprise", "saas"}

# Union types — Python 3.10+ syntax:
def find_deal(deal_id: int) -> dict | None:
    """Return a deal dict, or None if not found."""
    ...

# Any value (opt out of type checking — use sparingly):
from typing import Any
def log(value: Any) -> None: ...

from typing import TypedDict
from dataclasses import dataclass, field

# TypedDict: typed schema for dict-shaped data
class DealRecord(TypedDict):
    id: int
    name: str
    mrr: float
    closed: bool

def process_deal(deal: DealRecord) -> str:
    return f"{deal['name']}: ${deal['mrr']:,.2f}"

# @dataclass: typed class with auto-generated __init__, __repr__, __eq__
@dataclass
class Deal:
    id: int
    name: str
    mrr: float
    closed: bool = False
    tags: list[str] = field(default_factory=list)  # correct mutable default!

    def display(self) -> str:
        return f"Deal({self.name}, ${self.mrr:,.2f})"

d = Deal(id=1, name="Acme", mrr=4999.99)
print(d)            # Deal(id=1, name='Acme', mrr=4999.99, closed=False, tags=[])
print(d.display())   # Deal(Acme, $4,999.99)

from typing import Callable, Final, Literal, TypeVar
from collections.abc import Sequence, Iterator

# Callable: annotate functions-as-arguments
def apply_transform(data: list[float], fn: Callable[[float], float]) -> list[float]:
    return [fn(x) for x in data]

# Final: a name that should never be rebound
MAX_RETRIES: Final[int] = 3

# Literal: constrain to specific values
def set_status(status: Literal["open", "closed", "pending"]) -> None: ...

# TypeVar: generic functions
T = TypeVar("T")
def first(items: Sequence[T]) -> T:
    return items[0]

# Sequence: read-only, accepts list, tuple, str — prefer over list[T] when
# you only need to iterate (more flexible for callers)
def total_mrr(deals: Sequence[Deal]) -> float:
    return sum(d.mrr for d in deals)

score = 82

if score >= 90:
    grade = "A"
elif score >= 80:       # checked only if the first was False
    grade = "B"
elif score >= 70:
    grade = "C"
else:
    grade = "F"

print(grade)   # B

# Comparison operators:
# ==   !=   <   >   <=   >=   is   is not   in   not in

# Logical operators combine conditions:
age, member = 25, True
if age >= 18 and member:    # both must be True
    print("full access")
if not member or age < 13:  # either suffices
    print("restricted")

# Chained comparisons (unique to Python — reads like maths):
if 0 <= score <= 100:
    print("valid score")    # equivalent to score >= 0 and score <= 100

# Idiomatic truthiness checks:
items = []
if not items:             # ✓ idiomatic: "if the list is empty"
    print("no items")

name = ""
if not name:              # ✓ idiomatic: "if name is blank"
    print("name required")

value = None
if value is None:          # ✓ explicit None check — preferred
    print("missing value")

# Watch out: 0 is falsy, but 0 is a valid value!
count = 0
if not count:              # BUG: this fires for 0 AND None AND "" !
    print("no count")
if count is None:          # correct if you mean "not yet set"
    print("count not set")

# Short-circuit evaluation:
# 'and': stops at first falsy value (returns it)
# 'or' : stops at first truthy value (returns it)
default = None
label   = default or "unknown"    # "unknown" — common default pattern
print(label)

import re

# Walrus operator: assign inside a condition
data = "error: connection refused on port 8080"
if m := re.search(r"port (\d+)", data):
    print(f"Port: {m.group(1)}")   # Port: 8080
# Without walrus: m = re.search(...); if m: ...

# Guard clauses vs nested if:
def process_deal_nested(deal: dict | None) -> str:
    if deal is not None:
        if deal.get("mrr"):
            if deal["mrr"] > 0:
                return f"Valid: ${deal['mrr']:,.2f}"
    return "invalid"   # hard to see this at the bottom of nesting

def process_deal_guards(deal: dict | None) -> str:
    if deal is None:        # guard: bail early
        return "invalid"
    if not deal.get("mrr"):  # guard: bail early
        return "invalid"
    if deal["mrr"] <= 0:    # guard: bail early
        return "invalid"
    return f"Valid: ${deal['mrr']:,.2f}"   # happy path at the end

# Conditional expression:
deals = 5
status = "open" if deals > 0 else "none"  # "open"

# Short-circuit 'or' for defaults:
timeout = None
effective_timeout = timeout or 30    # 30  (None is falsy)
timeout2 = 0
effective2 = timeout2 or 30          # BUG: 30, not 0 — 0 is falsy!
# Fix when 0 is valid:
effective3 = timeout2 if timeout2 is not None else 30   # 0

# Short-circuit 'and' for safe access:
user = {"name": "Ada"}
email = user and user.get("email")   # None — user exists but no email key
no_user = None
no_email = no_user and no_user.get("email")  # None, no AttributeError

print(status, effective_timeout, effective2, effective3)

clients = ["Acme", "Globex", "Initech"]

# ── Basic iteration: items directly, no index needed ──
for client in clients:
    print(client.upper())

# ── enumerate: index + item in one step ──
for i, client in enumerate(clients, start=1):
    print(f"{i}. {client}")   # 1. Acme  /  2. Globex  /  3. Initech

# ── zip: walk two sequences in lockstep ──
mrrs = [9900, 14500, 2100]
for client, mrr in zip(clients, mrrs):
    print(f"{client}: ${mrr:,}")

# ── zip with unequal lengths (zip_longest for full coverage) ──
from itertools import zip_longest
for c, m in zip_longest(clients, [100, 200], fillvalue=0):
    print(c, m)   # Initech gets fillvalue=0

deal_mrrs = {"Acme": 9900, "Globex": 14500, "Initech": 2100}

# Iterate keys:
for name in deal_mrrs:
    print(name)

# Iterate key-value pairs — most common:
for name, mrr in deal_mrrs.items():
    print(f"{name}: ${mrr:,}")

# Sort by value descending:
for name, mrr in sorted(deal_mrrs.items(), key=lambda kv: kv[1], reverse=True):
    print(f"{name}: ${mrr:,}")

# Reverse a list without modifying it:
for client in reversed(list(deal_mrrs.keys())):
    print(client)

# while loop with break:
attempts = 0
while attempts < 3:
    pwd = input("password: ")
    if pwd == "secret":
        print("welcome")
        break
    attempts += 1
else:
    print("locked out")   # runs only if while exhausted without break

# for/else — elegant search pattern:
targets = ["Acme", "Globex", "Initech"]
for target in targets:
    if target.startswith("G"):
        print(f"Found: {target}")
        break
else:
    print("No match found")   # only if no break was hit

# continue: skip current iteration
for n in range(10):
    if n % 2 == 0:
        continue     # skip even numbers
    print(n)        # 1, 3, 5, 7, 9

import itertools

# chain: iterate multiple iterables as one
q1 = ["Jan", "Feb", "Mar"]
q2 = ["Apr", "May", "Jun"]
for month in itertools.chain(q1, q2):
    print(month, end=" ")   # Jan Feb Mar Apr May Jun

# islice: take the first N items from any iterable (no list needed)
import itertools
def naturals():
    n = 1
    while True:
        yield n; n += 1

print(list(itertools.islice(naturals(), 5)))   # [1, 2, 3, 4, 5]

# groupby: group consecutive elements by a key
events = [
    {"region": "EMEA", "deal": "Acme"},
    {"region": "EMEA", "deal": "Globex"},
    {"region": "AMER", "deal": "Initech"},
]
events_sorted = sorted(events, key=lambda e: e["region"])
for region, group in itertools.groupby(events_sorted, key=lambda e: e["region"]):
    deals = [e["deal"] for e in group]
    print(f"{region}: {deals}")
# AMER: ['Initech']
# EMEA: ['Acme', 'Globex']

def http_label(status: int) -> str:
    match status:
        case 200 | 201 | 204:      # OR pattern — matches any of these literals
            return "success"
        case 301 | 302:
            return "redirect"
        case 400:
            return "bad request"
        case 401 | 403:
            return "auth error"
        case 404:
            return "not found"
        case status if 500 <= status < 600:  # guard clause
            return f"server error ({status})"
        case _:                   # wildcard — the default case
            return "unknown"

print(http_label(200))    # success
print(http_label(503))    # server error (503)
print(http_label(418))    # unknown

def handle_event(event: dict) -> str:
    match event:
        # Dict pattern: match keys and bind values to names
        case {"type": "click", "x": x, "y": y}:
            return f"click at ({x}, {y})"
        case {"type": "key", "key": k} if k.isupper():
            return f"uppercase key: {k}"   # guard: only uppercase keys
        case {"type": "key", "key": k}:
            return f"key pressed: {k}"
        case {"type": t}:                       # bind unknown type
            return f"unknown event type: {t}"
        case _:
            return "malformed event"

print(handle_event({"type": "click", "x": 10, "y": 5}))   # click at (10, 5)
print(handle_event({"type": "key", "key": "ENTER"}))      # uppercase key: ENTER
print(handle_event({"type": "scroll"}))                    # unknown event type: scroll

def parse_command(tokens: list[str]) -> str:
    match tokens:
        case []:
            return "empty command"
        case ["quit"]:
            return "exiting"
        case ["get", resource]:               # exactly 2 elements
            return f"GET {resource}"
        case ["set", key, value]:             # exactly 3 elements
            return f"SET {key} = {value}"
        case [cmd, *args]:                    # first element + rest captured
            return f"unknown cmd {cmd!r} with {len(args)} args"

print(parse_command(["get", "users"]))           # GET users
print(parse_command(["set", "timeout", "30"]))  # SET timeout = 30
print(parse_command(["delete", "a", "b", "c"]))# unknown cmd 'delete' with 3 args

from dataclasses import dataclass

@dataclass
class UpgradeEvent:
    account_id: int
    new_plan: str

@dataclass
class ChurnEvent:
    account_id: int
    reason: str

@dataclass
class PaymentEvent:
    account_id: int
    amount: float

def handle(event) -> str:
    match event:
        case UpgradeEvent(account_id=aid, new_plan=plan):
            return f"Account {aid} upgraded to {plan}"
        case ChurnEvent(account_id=aid, reason=r):
            return f"Account {aid} churned: {r}"
        case PaymentEvent(account_id=aid, amount=amt) if amt > 10_000:
            return f"Large payment from {aid}: ${amt:,.2f}"
        case PaymentEvent(account_id=aid, amount=amt):
            return f"Payment from {aid}: ${amt:,.2f}"

print(handle(UpgradeEvent(42, "enterprise")))
print(handle(PaymentEvent(99, 15000)))

def full_signature(
    pos_only_a, pos_only_b, /,   # positional-only (zone 1)
    normal_c, normal_d,          # positional-or-keyword (zone 2)
    *args,                       # var-positional (zone 3)
    kw_only_e, kw_only_f=10,     # keyword-only (zone 4)
    **kwargs                     # var-keyword (zone 5)
):
    print(f"pos_only : {pos_only_a}, {pos_only_b}")
    print(f"normal   : {normal_c}, {normal_d}")
    print(f"args     : {args}")
    print(f"kw_only  : {kw_only_e}, {kw_only_f}")
    print(f"kwargs   : {kwargs}")

full_signature(1, 2, 3, 4, 5, 6, kw_only_e=99, color="red")

# This would FAIL – pos_only_a cannot be passed by name:
# full_signature(pos_only_a=1, pos_only_b=2, normal_c=3, normal_d=4, kw_only_e=9)

# ❌ Classic bug
def append_item_bad(item, lst=[]):
    lst.append(item)
    return lst

print(append_item_bad("a"))   # ['a']
print(append_item_bad("b"))   # ['a', 'b']  ← shared state!
print(append_item_bad("c"))   # ['a', 'b', 'c']

# ✅ Correct pattern: sentinel None, create inside body
def append_item_good(item, lst=None):
    if lst is None:
        lst = []
    lst.append(item)
    return lst

print(append_item_good("a"))  # ['a']
print(append_item_good("b"))  # ['b']  ← fresh list every time

# Inspect the shared default object
print(append_item_bad.__defaults__)   # (['a', 'b', 'c'],)

x = "global"          # G

def outer():
    x = "enclosing"   # E (from inner's perspective)

    def inner():
        x = "local"   # L
        print(x)      # → 'local'

    inner()
    print(x)          # → 'enclosing'

outer()
print(x)              # → 'global'
print(len)            # → built-in len function (B layer)

# global: reach up to module scope
counter = 0

def increment():
    global counter       # without this, Python creates a local 'counter'
    counter += 1

increment(); increment()
print(counter)   # 2

# nonlocal: reach up to nearest enclosing function scope (not global)
def make_counter():
    count = 0

    def tick():
        nonlocal count   # bind to enclosing 'count', not module
        count += 1
        return count

    return tick

c = make_counter()
print(c(), c(), c())   # 1 2 3

# nonlocal vs global: key difference
x = 100
def outer2():
    x = 200
    def inner2():
        nonlocal x       # reaches outer2's x=200, NOT module x=100
        x += 1
        print(x)         # 201
    inner2()
    print(x)             # 201

outer2()
print(x)                 # still 100 – module x untouched

import math

# Store in a variable
my_sqrt = math.sqrt
print(my_sqrt(16))   # 4.0

# Store in a list / dict (dispatch table)
ops = {
    "add": lambda a, b: a + b,
    "mul": lambda a, b: a * b,
    "pow": pow,
}
print(ops["pow"](2, 10))   # 1024

# Higher-order: pass as argument
def apply_twice(fn, value):
    return fn(fn(value))

print(apply_twice(lambda x: x * 3, 2))   # 18

# Higher-order: return from function
def multiplier(n):
    def mult(x):
        return x * n
    return mult

triple = multiplier(3)
print(triple(7))   # 21
print(type(triple))   # <class 'function'>

from typing import Optional

def parse_price(raw: str, currency: str = "USD") -> Optional[float]:
    """Convert a price string like '$12.99' to a float."""
    cleaned = raw.lstrip("$£€")
    try:
        return float(cleaned)
    except ValueError:
        return None

# Annotations are stored on the function object
print(parse_price.__annotations__)
# {'raw': <class 'str'>, 'currency': <class 'str'>, 'return': typing.Optional[float]}

# Inspect module gives richer introspection
import inspect
sig = inspect.signature(parse_price)
for name, param in sig.parameters.items():
    print(f"  {name}: default={param.default!r}, annotation={param.annotation}")

from functools import partial

def power(base, exponent):
    return base ** exponent

# Freeze the exponent, create specialised functions
square = partial(power, exponent=2)
cube   = partial(power, exponent=3)

print(square(5))   # 25
print(cube(4))     # 64

# Practical: freeze a logging function's level
import logging
logging.basicConfig(level=logging.DEBUG)
debug = partial(logging.log, logging.DEBUG)
warn  = partial(logging.log, logging.WARNING)

debug("Cache miss for key=%s", "user:42")
warn("Rate limit reached for IP=%s", "10.0.0.1")

# partial preserves the original function for introspection
print(square.func)       # <function power at ...>
print(square.keywords)   # {'exponent': 2}

sales = [
    {"product": "Widget", "revenue": 120, "region": "APAC"},
    {"product": "Gadget", "revenue": 450, "region": "EMEA"},
    {"product": "Doohickey", "revenue": 90, "region": "APAC"},
    {"product": "Thingamajig", "revenue": 310, "region": "EMEA"},
]

# List comprehension: transform
revenues = [row["revenue"] for row in sales]
print(revenues)   # [120, 450, 90, 310]

# Filter: only high-value
big_deals = [row["product"] for row in sales if row["revenue"] > 200]
print(big_deals)   # ['Gadget', 'Thingamajig']

# Dict comprehension: product → revenue map
rev_map = {row["product"]: row["revenue"] for row in sales}
print(rev_map)     # {'Widget': 120, 'Gadget': 450, ...}

# Set comprehension: unique regions
regions = {row["region"] for row in sales}
print(regions)     # {'APAC', 'EMEA'} (order varies)

# Generator expression (lazy – parens, not brackets)
total = sum(row["revenue"] for row in sales)
print(total)       # 970

# Flatten a 2-D matrix
matrix = [[1, 2, 3], [4, 5, 6], [7, 8, 9]]
flat = [cell for row in matrix for cell in row]
print(flat)   # [1, 2, 3, 4, 5, 6, 7, 8, 9]

# Transpose (rows become columns)
transposed = [[row[i] for row in matrix] for i in range(3)]
print(transposed)   # [[1, 4, 7], [2, 5, 8], [3, 6, 9]]

# Cartesian product (pairs of sizes and colours)
sizes   = ["S", "M", "L"]
colours = ["red", "blue"]
skus = [f"{s}-{c}" for s in sizes for c in colours]
print(skus)
# ['S-red', 'S-blue', 'M-red', 'M-blue', 'L-red', 'L-blue']

import re

log_lines = [
    "2024-01-15 ERROR disk full",
    "2024-01-15 INFO  backup started",
    "2024-01-15 ERROR connection timeout",
    "2024-01-15 DEBUG heartbeat",
]

# Without walrus: regex match twice (once to filter, once to extract)
errors_bad = [re.search(r"ERROR (.+)", line).group(1)
              for line in log_lines
              if re.search(r"ERROR (.+)", line)]   # double work

# With walrus: compute once, reuse in same expression
errors_good = [m.group(1)
               for line in log_lines
               if (m := re.search(r"ERROR (.+)", line))]

print(errors_good)   # ['disk full', 'connection timeout']

# Walrus also works for side-effectful computations
import random
random.seed(42)
results = [y for x in range(5) if (y := random.randint(1, 10)) > 5]
print(results)   # [7, 10, 8] (values already computed, not recomputed)

# What Python's for-loop actually does:
lst = [10, 20, 30]
it = iter(lst)              # calls lst.__iter__()
print(next(it))             # 10  (calls it.__next__())
print(next(it))             # 20
print(next(it))             # 30
# print(next(it))           # → StopIteration

# Writing a custom iterator
class CountUp:
    """Yields integers from start up to stop (inclusive)."""
    def __init__(self, start, stop):
        self.current = start
        self.stop    = stop

    def __iter__(self):
        return self       # iterator is its own iterable

    def __next__(self):
        if self.current > self.stop:
            raise StopIteration
        value = self.current
        self.current += 1
        return value

for n in CountUp(1, 5):
    print(n, end=" ")   # 1 2 3 4 5

import itertools

# product – Cartesian product (like nested for-loops)
for a, b in itertools.product("AB", [1, 2]):
    print(f"({a},{b})", end=" ")   # (A,1) (A,2) (B,1) (B,2)
print()

# combinations / permutations
print(list(itertools.combinations("ABCD", 2)))
# [('A','B'),('A','C'),('A','D'),('B','C'),('B','D'),('C','D')]

# groupby – consecutive groups (data must be pre-sorted by key)
events = [("APAC","login"),("APAC","purchase"),("EMEA","login"),("EMEA","view")]
for region, group in itertools.groupby(events, key=lambda e: e[0]):
    actions = [g[1] for g in group]
    print(f"{region}: {actions}")
# APAC: ['login', 'purchase']
# EMEA: ['login', 'view']

# takewhile / dropwhile
nums = [2, 4, 6, 7, 8, 10]
print(list(itertools.takewhile(lambda n: n % 2 == 0, nums)))  # [2, 4, 6]
print(list(itertools.dropwhile(lambda n: n % 2 == 0, nums)))  # [7, 8, 10]

# chain – concatenate iterables without building a list
a = range(3)
b = range(5, 8)
print(list(itertools.chain(a, b)))   # [0, 1, 2, 5, 6, 7]

# islice – lazy slicing of any iterator
inf_squares = (n*n for n in itertools.count(1))
print(list(itertools.islice(inf_squares, 6)))   # [1, 4, 9, 16, 25, 36]

from functools import reduce

# Good use: short, single-expression key function
records = [("Alice", 32), ("Bob", 25), ("Carol", 28)]
records.sort(key=lambda r: r[1])    # sort by age
print(records)   # [('Bob', 25), ('Carol', 28), ('Alice', 32)]

# Good use: inline with map/filter/reduce
prices = [9.99, 24.50, 3.75, 149.00]
discounted = list(map(lambda p: round(p * 0.9, 2), prices))
print(discounted)   # [8.99, 22.05, 3.38, 134.1]

total = reduce(lambda acc, p: acc + p, prices, 0)
print(f"Total: ${total:.2f}")   # Total: $187.24

# ❌ Anti-pattern: assigning lambda to a name
# bad:  double = lambda x: x * 2
# good: def double(x): return x * 2
# Named lambdas lose traceback clarity and can't have docstrings.

def make_adder(n):
    def adder(x):
        return x + n       # 'n' is a free variable
    return adder

add5  = make_adder(5)
add10 = make_adder(10)
print(add5(3))    # 8
print(add10(3))   # 13

# Each closure has its own cell holding a different 'n'
print(add5.__closure__)           # (<cell at 0x...>,)
print(add5.__closure__[0].cell_contents)    # 5
print(add10.__closure__[0].cell_contents)   # 10

# Cell objects are shared: mutating nonlocal updates the cell
def counter():
    count = 0
    def inc():
        nonlocal count
        count += 1
        return count
    return inc

c = counter()
print(c(), c(), c())   # 1 2 3
# Inspect: the cell still holds the latest count
print(c.__closure__[0].cell_contents)  # 3

import time
import functools

def timer(fn):
    @functools.wraps(fn)       # preserves __name__, __doc__, __annotations__
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        result = fn(*args, **kwargs)
        elapsed = time.perf_counter() - start
        print(f"{fn.__name__} took {elapsed:.4f}s")
        return result
    return wrapper

@timer
def slow_sum(n):
    """Sum integers 0..n."""
    return sum(range(n))

print(slow_sum(10_000_000))
# slow_sum took 0.3211s
# 49999995000000

# The @ syntax is purely shorthand:
# @timer         is identical to:
# slow_sum = timer(slow_sum)

# functools.wraps matters:
print(slow_sum.__name__)   # 'slow_sum'  (not 'wrapper'!)
print(slow_sum.__doc__)    # 'Sum integers 0..n.'

import time, functools, random

def retry(max_attempts=3, delay=1.0, exceptions=(Exception,)):
    """Decorator factory: retry on specified exceptions."""
    def decorator(fn):                      # level 2 – receives the function
        @functools.wraps(fn)
        def wrapper(*args, **kwargs):       # level 3 – called at runtime
            last_exc = None
            for attempt in range(1, max_attempts + 1):
                try:
                    return fn(*args, **kwargs)
                except exceptions as exc:
                    last_exc = exc
                    print(f"Attempt {attempt}/{max_attempts} failed: {exc}")
                    if attempt < max_attempts:
                        time.sleep(delay * attempt)   # exponential-ish backoff
            raise last_exc
        return wrapper
    return decorator     # decorator factory returns decorator

# Usage
@retry(max_attempts=4, delay=0.1, exceptions=(ValueError, IOError))
def flaky_api_call(endpoint):
    if random.random() < 0.7:
        raise ValueError(f"Transient error on {endpoint}")
    return f"Success from {endpoint}"

random.seed(99)
print(flaky_api_call("/health"))

import functools, time

class RateLimiter:
    """Allow at most `calls` calls per `period` seconds."""
    def __init__(self, calls, period):
        self.calls  = calls
        self.period = period
        self.history = []

    def __call__(self, fn):
        @functools.wraps(fn)
        def wrapper(*args, **kwargs):
            now = time.monotonic()
            # Purge old entries outside the window
            self.history = [t for t in self.history if now - t < self.period]
            if len(self.history) >= self.calls:
                raise RuntimeError(
                    f"Rate limit exceeded: {self.calls} calls/{self.period}s"
                )
            self.history.append(now)
            return fn(*args, **kwargs)
        return wrapper

@RateLimiter(calls=3, period=1.0)
def fetch_quote(symbol):
    return f"Price of {symbol}: ${round(100 + hash(symbol) % 50, 2)}"

for _ in range(3):
    print(fetch_quote("AAPL"))

try:
    fetch_quote("AAPL")   # 4th call within 1 second
except RuntimeError as e:
    print(f"Caught: {e}")

import functools

def bold(fn):
    @functools.wraps(fn)
    def wrapper(*a, **kw): return f"<b>{fn(*a, **kw)}</b>"
    return wrapper

def italic(fn):
    @functools.wraps(fn)
    def wrapper(*a, **kw): return f"<i>{fn(*a, **kw)}</i>"
    return wrapper

@bold       # applied second (outer)
@italic     # applied first (inner)
def greet(name):
    return f"Hello, {name}"

print(greet("Alice"))   # <b><i>Hello, Alice</i></b>

# Execution order reads bottom-up for application,
# top-down for actual invocation:
# greet = bold(italic(greet))
# call: bold.wrapper → italic.wrapper → original greet

import functools

# @cache (Python 3.9+) is lru_cache(maxsize=None) – unbounded
@functools.cache
def fib(n):
    if n < 2: return n
    return fib(n-1) + fib(n-2)

print(fib(50))            # 12586269025  (instant)
print(fib.cache_info())   # hits=..., misses=51, maxsize=None, currsize=51

# @lru_cache(maxsize=N) – bounded LRU eviction
@functools.lru_cache(maxsize=128)
def heavy_query(region, year):
    # simulate DB round-trip
    import time; time.sleep(0.01)
    return f"data_{region}_{year}"

heavy_query("APAC", 2023)
heavy_query("APAC", 2023)   # cache hit
print(heavy_query.cache_info())
# CacheInfo(hits=1, misses=1, maxsize=128, currsize=1)

heavy_query.cache_clear()   # manual invalidation

class Temperature:
    def __init__(self, celsius):
        self._celsius = celsius

    @property
    def celsius(self):
        """Getter – accessed like an attribute."""
        return self._celsius

    @celsius.setter
    def celsius(self, value):
        if value < -273.15:
            raise ValueError(f"Below absolute zero: {value}")
        self._celsius = value

    @celsius.deleter
    def celsius(self):
        del self._celsius

    @property
    def fahrenheit(self):
        return self._celsius * 9/5 + 32

    @classmethod
    def from_fahrenheit(cls, f):
        """Alternative constructor – receives the class, not an instance."""
        return cls((f - 32) * 5/9)

    @staticmethod
    def is_valid(value):
        """Utility with no access to class or instance."""
        return value >= -273.15

t = Temperature(100)
print(t.celsius)       # 100
print(t.fahrenheit)    # 212.0
t.celsius = 37
print(t.celsius)       # 37

body = Temperature.from_fahrenheit(98.6)
print(round(body.celsius, 1))   # 37.0

print(Temperature.is_valid(-300))   # False

def count_up(start, stop):
    """Generator: yields integers from start to stop, inclusive."""
    current = start
    while current <= stop:
        print(f"  >> about to yield {current}")
        yield current
        print(f"  >> resumed after yield {current}")
        current += 1
    print("  >> generator exhausted")

gen = count_up(1, 3)
print(type(gen))           # <class 'generator'>
print("Calling next()...")
print(next(gen))           # >> about to yield 1  |  1
print(next(gen))           # >> resumed ...  |  >> about to yield 2  |  2
for val in gen:            # drives remaining values
    print(val)             # 3

# Second call creates a fresh generator – they don't restart
gen2 = count_up(1, 3)
print(list(gen2))   # [1, 2, 3]

import itertools

# Infinite generator – never exhausted
def integers(start=0):
    n = start
    while True:
        yield n
        n += 1

# Pipeline stages: each is a generator consuming another generator
def evens(src):
    for n in src:
        if n % 2 == 0:
            yield n

def squared(src):
    for n in src:
        yield n * n

def take(n, src):
    for _ in range(n):
        yield next(src)

# Compose the pipeline (nothing executes yet)
pipeline = take(5, squared(evens(integers(1))))

# Pull values through on demand
print(list(pipeline))   # [4, 16, 36, 64, 100]
#   integers(1)  → 1,2,3,4,5,6,7,8,9,10,...
#   evens        → 2,4,6,8,10,...
#   squared      → 4,16,36,64,100,...
#   take(5)      → [4,16,36,64,100]  (stops pulling)

def flatten(nested):
    """Recursively flatten arbitrarily nested lists."""
    for item in nested:
        if isinstance(item, list):
            yield from flatten(item)   # delegate to recursive call
        else:
            yield item

data = [1, [2, [3, 4], 5], [6, 7], 8]
print(list(flatten(data)))   # [1, 2, 3, 4, 5, 6, 7, 8]

# Without yield from, you'd need:
# for sub_item in flatten(item):
#     yield sub_item
# yield from is cleaner AND passes send() / throw() / close() through
# to the sub-generator (critical for coroutine chaining)

# yield from also captures the sub-generator's StopIteration value
def sub():
    yield 1
    yield 2
    return "sub_done"          # StopIteration value

def delegator():
    result = yield from sub()  # 'result' gets "sub_done"
    print(f"sub returned: {result}")
    yield 99

gen = delegator()
print(next(gen))   # 1
print(next(gen))   # 2
print(next(gen))   # sub returned: sub_done  |  99

def running_average():
    """Coroutine that accepts values via send() and yields the average."""
    total = 0.0
    count = 0
    value = yield   # prime: first next() runs to here, returns None
    while True:
        total += value
        count += 1
        value = yield total / count   # send average back, receive next value

avg = running_average()
next(avg)          # prime the coroutine (advance to first yield)
print(avg.send(10))    # 10.0
print(avg.send(20))    # 15.0
print(avg.send(30))    # 20.0
print(avg.send(40))    # 25.0

# throw() injects an exception at the yield point
def resilient():
    try:
        while True:
            value = yield
            print(f"Got: {value}")
    except ValueError as e:
        print(f"Caught inside generator: {e}")
        yield "handled"

r = resilient()
next(r)
r.send(42)               # Got: 42
print(r.throw(ValueError, "bad input"))   # Caught... | 'handled'

# close() raises GeneratorExit at the yield point
gen = (x for x in range(100))
print(next(gen))   # 0
gen.close()        # generator is now exhausted / closed

# This for loop:
for x in [1, 2, 3]:
    print(x)

# Is exactly equivalent to:
_iter = iter([1, 2, 3])    # calls list.__iter__()
while True:
    try:
        x = next(_iter)    # calls _iter.__next__()
        print(x)
    except StopIteration:
        break

# Generators satisfy the iterator protocol natively:
# gen.__iter__() returns self
# gen.__next__() resumes and returns the next yielded value
# So generators can be used directly in for loops, zip(), map() etc.

import sys, tracemalloc

N = 1_000_000

# Eager list
tracemalloc.start()
result_list = [n * n for n in range(N)]
_, list_peak = tracemalloc.get_traced_memory()
tracemalloc.stop()

# Lazy generator
tracemalloc.start()
result_gen = (n * n for n in range(N))
_, gen_peak = tracemalloc.get_traced_memory()
tracemalloc.stop()

print(f"List peak:      {list_peak / 1_048_576:.2f} MB")
print(f"Generator peak: {gen_peak / 1_048_576:.4f} MB")
print(f"Ratio: {list_peak // gen_peak}x more memory for list")

import csv, gzip, itertools
from pathlib import Path
from typing import Iterator

def read_gzipped_csv(path: Path) -> Iterator[dict]:
    """Stream rows from a gzipped CSV without loading it into memory."""
    with gzip.open(path, "rt", encoding="utf-8") as f:
        reader = csv.DictReader(f)
        for row in reader:
            yield row

def parse_numerics(rows: Iterator[dict]) -> Iterator[dict]:
    """Cast revenue and quantity to correct types."""
    for row in rows:
        row["revenue"]  = float(row.get("revenue",  0) or 0)
        row["quantity"] = int(row.get("quantity", 0)   or 0)
        yield row

def filter_region(rows: Iterator[dict], region: str) -> Iterator[dict]:
    for row in rows:
        if row.get("region") == region:
            yield row

def aggregate_by_product(rows: Iterator[dict]) -> dict:
    totals = {}
    for row in rows:
        pid = row["product_id"]
        totals[pid] = totals.get(pid, 0.0) + row["revenue"]
    return totals

# Compose pipeline – nothing runs yet
files = Path("data").glob("sales_*.csv.gz")   # e.g., 50 files @ 1 GB each
raw      = itertools.chain.from_iterable(read_gzipped_csv(f) for f in files)
parsed   = parse_numerics(raw)
apac     = filter_region(parsed, "APAC")
result   = aggregate_by_product(apac)   # pulls the pipeline

# Peak memory stays near the size of one CSV row, never the full dataset
print(sorted(result.items(), key=lambda x: x[1], reverse=True)[:5])

from functools import wraps

def coroutine(fn):
    """Decorator to auto-prime a generator coroutine."""
    @wraps(fn)
    def wrapper(*args, **kwargs):
        gen = fn(*args, **kwargs)
        next(gen)   # advance to first yield
        return gen
    return wrapper

@coroutine
def grep(pattern):
    """Coroutine that receives lines and prints matches."""
    import re
    while True:
        line = yield
        if re.search(pattern, line):
            print(f"  MATCH: {line.rstrip()}")

@coroutine
def broadcast(*targets):
    """Fan-out: send each received value to multiple coroutines."""
    while True:
        value = yield
        for target in targets:
            target.send(value)

# Build a coroutine pipeline
errors  = grep(r"ERROR")
timeouts = grep(r"timeout")
router   = broadcast(errors, timeouts)

lines = [
    "2024-01-01 INFO  service started\n",
    "2024-01-01 ERROR disk full\n",
    "2024-01-01 ERROR connection timeout\n",
    "2024-01-01 DEBUG heartbeat ok\n",
]

for line in lines:
    router.send(line)

router.close()

import sys

lst = []
last_size = 0
for i in range(16):
    lst.append(i)
    size = sys.getsizeof(lst)
    if size != last_size:
        print(f"len={len(lst):2d}  allocated capacity changed  bytes={size}")
        last_size = size

nums = [3, 1, 4, 1, 5, 9, 2, 6]

# ── O(1) ─────────────────────────────────────────────
nums[3]             # 1        — pointer math: base + 3*8
nums[-1]            # 6        — len-1 then pointer math
nums.append(7)      # extend at end, amortized O(1)
nums.pop()          # 7        — shrink at end, O(1)
len(nums)           # 8        — stored as a field, not counted

# ── O(k) — proportional to slice length ──────────────
nums[2:5]           # [4, 1, 5]  — copies 3 pointers into new list
nums[::-1]          # reversed  — copies all n pointers

# ── O(n) — touch every element ───────────────────────
nums.insert(0, 99)  # shifts EVERY element right by one
nums.pop(0)         # shifts EVERY element left by one
3 in nums           # linear scan — compare one by one
nums.index(5)       # linear scan for first match
nums.count(1)       # linear scan counting matches
nums.reverse()      # in-place reversal

# ── O(n log n) ────────────────────────────────────────
nums.sort()                    # in-place Timsort (stable)
sorted(nums)                   # returns new sorted list
nums.sort(key=lambda x: -x)    # sort with key function

# BUG: the default [] is created once and reused
def add_tag(tag, tags=[]):     # WRONG — shared across calls
    tags.append(tag)
    return tags

add_tag("python")    # ["python"]
add_tag("data")      # ["python", "data"]  — unexpected!
add_tag("api")       # ["python", "data", "api"]

# CORRECT: None sentinel, fresh list each call
def add_tag(tag, tags=None):
    if tags is None:
        tags = []
    tags.append(tag)
    return tags

add_tag("python")    # ["python"]
add_tag("data")      # ["data"]  — independent, correct

import sys

# Tuples as typed records
coord   = (51.5074, -0.1278)         # (lat, lon) — structured record
rgb     = (255, 128, 0)              # immutable colour
db_row  = ("ada@x.io", "engineer", 36)

# Hashable: tuples can be dict keys and set members
visit_counts = {}
visit_counts[(51.5074, -0.1278)] = 42   # coordinates as key
visit_counts[(40.7128, -74.0060)] = 18

# Try the same with a list — TypeError
try:
    d = {[1, 2]: "value"}
except TypeError as e:
    print(e)    # unhashable type: 'list'

# Memory: tuples are smaller (no over-allocation)
lst = [1, 2, 3, 4, 5]
tup = (1, 2, 3, 4, 5)
print(sys.getsizeof(lst))   # 104 bytes  (capacity headroom)
print(sys.getsizeof(tup))   #  80 bytes  (exact fit)

# Extended unpacking — the most powerful unpack form
first, *middle, last = (10, 20, 30, 40, 50)
# first=10  middle=[20, 30, 40]  last=50

# Discard with _ by convention
_, important, _ = ("ignore", "keep_this", "ignore")

# Pythonic swap — no temp variable needed
x, y = 5, 10
x, y = y, x          # x=10, y=5  (uses tuple packing/unpacking)

# Function returning multiple values is always a tuple
def divmod_custom(a, b):
    return a // b, a % b   # packs into a tuple

quotient, remainder = divmod_custom(17, 5)
# quotient=3  remainder=2

data = [10, 20, 30, 40, 50, 60, 70, 80, 90]

# Slicing
data[2:5]      # [30, 40, 50]   — indices 2, 3, 4
data[:4]       # [10, 20, 30, 40]  — first 4
data[-3:]      # [70, 80, 90]   — last 3
data[1::2]     # [20, 40, 60, 80]  — every other starting at 1
data[::-1]     # [90, 80, ... 10]  — full reversal (new list)

# Slice assignment — in-place replacement, works even with different length
data[1:3] = [200, 300, 350]    # replace 2 elements with 3

# List comprehension vs manual loop — identical output, comp preferred
squares_loop = []
for n in range(10):
    squares_loop.append(n ** 2)

squares_comp = [n ** 2 for n in range(10)]
# [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]

# Filtered comprehension — real-world: parse valid prices from messy data
raw = ["19.99", "N/A", "5.00", "", "12.49", "SOLD"]
prices = [float(p) for p in raw if p.replace(".", "", 1).isdigit()]
# [19.99, 5.0, 12.49]

# Generator expression — no intermediate list, lazy evaluation
# Use when you pipe directly into sum/max/any/all/join
total = sum(p * 1.2 for p in prices)          # sum without a list
has_expensive = any(p > 15 for p in prices)   # short-circuits at first True

# Memory comparison: comprehension vs generator for 10M items
import sys
comp_mem = sys.getsizeof([x for x in range(100)])    # ~904 bytes for 100 items
gen_mem  = sys.getsizeof(x for x in range(100))      # ~208 bytes regardless of size
print(f"list: {comp_mem}  generator: {gen_mem}")

from collections import deque

# BFS graph traversal — deque is the canonical frontier
def bfs(graph, start):
    visited  = {start}
    frontier = deque([start])    # O(1) popleft — critical for BFS
    order    = []
    while frontier:
        node = frontier.popleft()
        order.append(node)
        for nbr in graph.get(node, []):
            if nbr not in visited:
                visited.add(nbr)
                frontier.append(nbr)
    return order

graph = {"A": ["B", "C"], "B": ["D"], "C": ["D", "E"], "D": [], "E": []}
bfs(graph, "A")   # ['A', 'B', 'C', 'D', 'E']

# Fixed-size sliding window — maxlen auto-evicts oldest
def moving_average(values, n):
    window = deque(maxlen=n)
    result = []
    for v in values:
        window.append(v)
        if len(window) == n:
            result.append(sum(window) / n)
    return result

moving_average([1, 3, 5, 7, 9, 11], n=3)
# [3.0, 5.0, 7.0, 9.0]  — each is avg of last 3 values

# Task queue (producer-consumer)
tasks = deque()
tasks.append("send_email")        # producer adds to right
tasks.append("process_payment")
tasks.popleft()   # "send_email"  — consumer takes from left, O(1)

user = {"name": "Ada", "role": "engineer", "active": True}

# Basic access
user["name"]                      # "Ada"              O(1)
user.get("email")                 # None               O(1) — safe, no KeyError
user.get("email", "unknown")      # "unknown"          O(1) — with default
"role" in user                    # True               O(1) key membership
"salary" in user                  # False

# Mutation
user["email"] = "ada@x.io"        # insert or overwrite   O(1)
del user["active"]                # delete by key          O(1)
user.pop("role", None)            # delete + return, safe if missing

# Iteration — dicts preserve insertion order (Python 3.7+)
for key in user:                  # iterates keys
    print(key)
for key, val in user.items():     # iterates (key, value) pairs
    print(f"{key}: {val}")
list(user.keys())                 # ['name', 'email']
list(user.values())               # ['Ada', 'ada@x.io']

# Merge and update (Python 3.9+)
defaults = {"timeout": 30, "retries": 3}
config   = {"timeout": 60, "debug": True}
merged = defaults | config        # {'timeout': 60, 'retries': 3, 'debug': True}
# right side wins on collision

# dict comprehension
squares = {n: n**2 for n in range(6)}
# {0: 0, 1: 1, 2: 4, 3: 9, 4: 16, 5: 25}

# Invert a dict (assuming unique values)
inv = {v: k for k, v in squares.items()}

from collections import defaultdict

# PATTERN 1: Frequency counting
events = ["click", "view", "click", "purchase", "view", "click"]

# Manual with .get()
freq = {}
for e in events:
    freq[e] = freq.get(e, 0) + 1
# {"click": 3, "view": 2, "purchase": 1}

# Idiomatic with defaultdict
freq2 = defaultdict(int)
for e in events:
    freq2[e] += 1   # missing key auto-initializes to int() = 0

# Best: Counter (covered in 5.4)
from collections import Counter
Counter(events)     # Counter({'click': 3, 'view': 2, 'purchase': 1})


# PATTERN 2: Grouping (one-to-many)
orders = [
    {"id": 1, "region": "EU", "amount": 120},
    {"id": 2, "region": "US", "amount": 85},
    {"id": 3, "region": "EU", "amount": 200},
    {"id": 4, "region": "US", "amount": 310},
]

by_region = defaultdict(list)
for order in orders:
    by_region[order["region"]].append(order["amount"])
# {"EU": [120, 200], "US": [85, 310]}

# PATTERN 3: Replacing an O(n) loop with an O(1) dict lookup
# Two-sum: find two indices whose values sum to target
def two_sum(nums, target):
    seen = {}                        # value -> index
    for i, n in enumerate(nums):
        complement = target - n
        if complement in seen:       # O(1) lookup
            return [seen[complement], i]
        seen[n] = i
    return []

two_sum([2, 7, 11, 15], 9)   # [0, 1]  — O(n) total, not O(n²)

# Set creation
active_users = {"alice", "bob", "carol"}    # literal syntax
ids = set([1, 2, 2, 3, 3, 3])              # deduplication: {1, 2, 3}
empty = set()                               # NOT {} — that's an empty dict!

# Membership — O(1)
"alice" in active_users    # True
"dave"  in active_users    # False

# Modification
active_users.add("dave")
active_users.discard("bob")    # remove if present, no error if missing
active_users.remove("bob")     # removes; raises KeyError if missing

# Set algebra — real-world: user overlap analysis
paying     = {"alice", "carol", "eve"}
trial      = {"bob", "carol", "frank"}

paying & trial          # intersection: {"carol"}       — both groups
paying | trial          # union: all users in either
paying - trial          # difference: paying but NOT trial: {"alice", "eve"}
trial  - paying         # trial but NOT paying: {"bob", "frank"}
paying ^ trial          # symmetric difference: in one but not both

# Subset / superset checks
{"alice"} <= paying     # True  — issubset
paying >= {"alice"}     # True  — issuperset

# Frozenset: immutable set (hashable — can be a dict key or set member)
cache_key = frozenset(["read", "write"])
permissions = {cache_key: True}

# Cleaning and normalizing
s = "  Ada Lovelace  "
s.strip()           # "Ada Lovelace"    remove leading/trailing whitespace
s.lstrip()          # "Ada Lovelace  "  left only
s.rstrip()          # "  Ada Lovelace"  right only
s.strip().lower()   # "ada lovelace"    chain methods freely

# Splitting and joining
"a,b,c".split(",")          # ["a", "b", "c"]
"  a  b  c  ".split()       # ["a", "b", "c"]  no arg: split on ANY whitespace
"-".join(["2026", "06", "27"])   # "2026-06-27"
"\n".join(["line1", "line2"])    # "line1\nline2"

# Searching and checking
"hello world".find("world")   # 6      (-1 if not found)
"hello world".index("world")  # 6      (raises ValueError if not found)
"hello world".count("l")      # 3
"report.pdf".endswith(".pdf")  # True
"config.yaml".startswith("con")  # True
"hello world".replace("l", "L")  # "heLLo worLd"

# Testing string content
"abc123".isalpha()    # False  (has digits)
"abc".isalpha()       # True
"123".isdigit()       # True
"abc123".isalnum()    # True
"  ".isspace()        # True

# Case manipulation
"ada lovelace".title()   # "Ada Lovelace"
"Ada".upper()            # "ADA"
"ADA".lower()            # "ada"
"Ada".swapcase()         # "aDA"

# Padding and alignment
"42".zfill(5)                 # "00042"   zero-pad on left
"left".ljust(10)              # "left      "
"right".rjust(10)             # "     right"
"center".center(12, "-")      # "---center---"

name  = "Ada"
qty   = 3
price = 19.5
score = 0.9234

# Basic expression embedding — any valid Python expression
f"Hello, {name}!"                      # "Hello, Ada!"
f"{qty} × {price} = {qty * price}"     # "3 × 19.5 = 58.5"
f"{name.upper()!r}"                    # "'ADA'"  (!r adds repr())
f"{name!s} {name!a}"                   # str() and ascii() conversion

# Number formatting — the format spec after :
f"{price:.2f}"           # "19.50"   2 decimal places
f"{price:08.2f}"         # "00019.50"  width 8, zero-padded
f"{score:.1%}"           # "92.3%"   percentage
f"{1_000_000:,}"         # "1,000,000"  thousands separator
f"{255:#010x}"           # "0x000000ff"  hex with prefix, padded

# Alignment
f"{'left':<10}|"         # "left      |"   left-align in width 10
f"{'right':>10}|"        # "     right|"   right-align
f"{'center':^10}|"       # "  center  |"   center
f"{'fill':*^10}|"        # "**fill****|"   fill with *

# Debug form (Python 3.8+): variable=value
f"{price=}"              # "price=19.5"
f"{qty * price = :.2f}"  # "qty * price = 58.50"

# Multi-line f-string
report = (
    f"Customer: {name}\n"
    f"  Items:  {qty}\n"
    f"  Total:  ${qty * price:.2f}\n"
)

# Conditional expression inside f-string
f"Status: {'active' if score > 0.9 else 'inactive'}"
# "Status: active"

# str (text) vs bytes (binary)
text  = "Hello, 世界"           # str: Unicode, abstract characters
raw   = text.encode("utf-8")    # bytes: b'Hello, \xe4\xb8\x96\xe7\x95\x8c'
back  = raw.decode("utf-8")     # str again: "Hello, 世界"

len(text)   # 9  characters
len(raw)    # 13  bytes (CJK characters take 3 bytes each in UTF-8)

# ALWAYS specify encoding when opening files
with open("data.txt", "w", encoding="utf-8") as f:
    f.write("Hello, 世界\n")

with open("data.txt", encoding="utf-8") as f:
    content = f.read()

# Common encoding pitfall: platform default != utf-8 on Windows
# open("file.txt") on Windows might use cp1252 — always be explicit

# Practical: check and normalize encoding in a data pipeline
def safe_decode(data: bytes, fallback="replace") -> str:
    """Try UTF-8 first, fall back gracefully."""
    try:
        return data.decode("utf-8")
    except UnicodeDecodeError:
        return data.decode("latin-1", errors=fallback)

import timeit

# SLOW: O(n²) — each += copies the whole string built so far
def slow_concat(n):
    result = ""
    for i in range(n):
        result += str(i) + ","
    return result

# FAST: O(n) — collect, then join once
def fast_concat(n):
    parts = []
    for i in range(n):
        parts.append(str(i))
    return ",".join(parts)

# FASTEST: generator expression into join
def fastest_concat(n):
    return ",".join(str(i) for i in range(n))

n = 10_000
slow = timeit.timeit(lambda: slow_concat(n), number=100)
fast = timeit.timeit(lambda: fast_concat(n), number=100)

print(f"slow: {slow:.3f}s   fast: {fast:.3f}s   speedup: {slow/fast:.1f}x")
# typical: slow: 1.8s   fast: 0.12s   speedup: 15x

import re

log_line = "2026-06-27 14:32:01 ERROR user_id=1042 msg='Payment failed'"

# re.search: find first match anywhere in string
match = re.search(r"\d{4}-\d{2}-\d{2}", log_line)
if match:
    print(match.group())    # "2026-06-27"

# re.findall: return all non-overlapping matches as list
re.findall(r"\d+", log_line)    # ["2026", "06", "27", "14", "32", "01", "1042"]

# Named groups: extract structured data
pattern = re.compile(
    r"(?P<date>\d{4}-\d{2}-\d{2}) "
    r"(?P<time>\d{2}:\d{2}:\d{2}) "
    r"(?P<level>\w+) "
    r"user_id=(?P<uid>\d+)"
)
m = pattern.search(log_line)
m.group("date")     # "2026-06-27"
m.group("level")    # "ERROR"
m.group("uid")      # "1042"
m.groupdict()       # {'date': '2026-06-27', 'time': '14:32:01', 'level': 'ERROR', 'uid': '1042'}

# re.sub: replace with regex
cleaned = re.sub(r"\s+", " ", "  too   many   spaces  ").strip()
# "too many spaces"

# Compile patterns you reuse — avoids recompiling on every call
EMAIL_RE = re.compile(r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}")
emails = EMAIL_RE.findall("contact ada@x.io or bob@y.com for help")
# ['ada@x.io', 'bob@y.com']

from collections import Counter

# Construct from any iterable
word_freq = Counter("the quick brown fox jumps over the lazy fox".split())
# Counter({'the': 2, 'fox': 2, 'quick': 1, ...})

word_freq.most_common(3)    # [('the', 2), ('fox', 2), ('quick', 1)]
word_freq["the"]            # 2
word_freq["missing"]        # 0  — Counter never raises KeyError

# Arithmetic on Counters
a = Counter({"python": 3, "java": 2, "rust": 1})
b = Counter({"python": 1, "go": 4, "java": 1})
a + b    # Counter({'go': 4, 'python': 4, 'java': 3, 'rust': 1})
a - b    # Counter({'python': 2, 'rust': 1})  — positive only
a & b    # Counter({'python': 1, 'java': 1})  — min of each
a | b    # Counter({'go': 4, 'python': 3, 'java': 2, 'rust': 1})  — max

# Real-world: count HTTP status codes in access log
import re
log_lines = [
    "GET /api/users 200", "POST /api/orders 201",
    "GET /api/users 200", "DELETE /api/item 404",
    "GET /api/orders 500", "GET /api/users 200",
]
status_codes = Counter(
    m.group() for line in log_lines
    if (m := re.search(r"\d{3}$", line))
)
print(status_codes)     # Counter({'200': 3, '201': 1, '404': 1, '500': 1})
print(status_codes.most_common(2))   # [('200', 3), ('201', 1)]

from collections import defaultdict, ChainMap

# Grouping: one-to-many mapping
events = [
    ("alice", "login"), ("bob", "view"), ("alice", "purchase"),
    ("carol", "login"), ("bob", "logout"), ("alice", "logout"),
]

by_user = defaultdict(list)
for user, action in events:
    by_user[user].append(action)
# defaultdict(list, {'alice': ['login', 'purchase', 'logout'],
#                    'bob':   ['view', 'logout'],
#                    'carol': ['login']})

# Nested grouping: region → category → list of amounts
sales = [
    ("EU", "hardware", 200), ("US", "software", 150),
    ("EU", "software", 80),  ("US", "hardware", 320),
    ("EU", "hardware", 110),
]
grouped = defaultdict(lambda: defaultdict(list))
for region, cat, amount in sales:
    grouped[region][cat].append(amount)
# {"EU": {"hardware": [200, 110], "software": [80]}, ...}

# Summing: accumulate totals without init check
revenue = defaultdict(float)
for region, _, amount in sales:
    revenue[region] += amount
# defaultdict(float, {'EU': 390.0, 'US': 470.0})

# ChainMap: layered lookup (think: local scope → module scope → builtins)
defaults = {"color": "blue", "size": "M", "qty": 1}
user_prefs = {"size": "XL"}
session = {"qty": 3}

# First match wins — like variable scoping
config = ChainMap(session, user_prefs, defaults)
config["size"]    # "XL"   — from user_prefs
config["color"]   # "blue" — fallback to defaults
config["qty"]     # 3      — from session

from dataclasses import dataclass, field, KW_ONLY
from datetime import datetime

@dataclass
class LineItem:
    sku:   str
    qty:   int
    price: float

    @property
    def subtotal(self) -> float:
        return round(self.qty * self.price, 2)

@dataclass
class Order:
    id:         int
    customer:   str
    items:      list[LineItem] = field(default_factory=list)   # safe mutable default
    created_at: datetime       = field(default_factory=datetime.utcnow)
    _:          KW_ONLY        = field(init=False, repr=False)  # forces rest to kw-only
    paid:       bool           = False

    def total(self) -> float:
        return round(sum(i.subtotal for i in self.items), 2)

    def add_item(self, sku: str, qty: int, price: float) -> None:
        self.items.append(LineItem(sku, qty, price))

# Usage
order = Order(id=1, customer="Ada")
order.add_item("PEN-001", qty=3, price=2.50)
order.add_item("PAD-002", qty=1, price=12.99)
print(order.total())   # 20.49
print(order)
# Order(id=1, customer='Ada', items=[LineItem(...), LineItem(...)], paid=False)

# Frozen dataclass: immutable, hashable (can be a set member / dict key)
@dataclass(frozen=True)
class Point:
    x: float
    y: float

p = Point(1.0, 2.0)
# p.x = 3.0  # FrozenInstanceError
{p: "origin"}   # works because frozen=True makes it hashable

import heapq

# Min-heap: smallest element always at index 0
tasks = []
heapq.heappush(tasks, (3, "low priority task"))
heapq.heappush(tasks, (1, "urgent task"))
heapq.heappush(tasks, (2, "normal task"))

heapq.heappop(tasks)    # (1, 'urgent task')  — always min
heapq.heappop(tasks)    # (2, 'normal task')

# Top-k smallest/largest — O(n log k) — more efficient than full sort for small k
scores = [88, 42, 95, 67, 12, 73, 99, 56, 31, 84]
heapq.nlargest(3, scores)     # [99, 95, 88]
heapq.nsmallest(3, scores)    # [12, 31, 42]

# Merge sorted iterables — O(n log k) where k = number of iterables
import heapq
iter1 = [1, 4, 7]
iter2 = [2, 5, 8]
iter3 = [3, 6, 9]
list(heapq.merge(iter1, iter2, iter3))
# [1, 2, 3, 4, 5, 6, 7, 8, 9]

# Max-heap trick: negate the priority
events = [(3, "C"), (1, "A"), (4, "D"), (2, "B")]
max_heap = []
for priority, item in events:
    heapq.heappush(max_heap, (-priority, item))  # negate for max behavior
heapq.heappop(max_heap)   # (-4, 'D') — was highest priority

from pathlib import Path

# ── Writing ──────────────────────────────────────────────────────────────────
# "w" creates or truncates, "a" appends, "x" exclusive-create (fails if exists)
with open("notes.txt", "w", encoding="utf-8") as f:
    f.write("first line\n")
    f.write("second line\n")
    f.writelines(["three\n", "four\n"])   # write iterable of strings

# ── Reading ───────────────────────────────────────────────────────────────────
# Read entire file at once — fine for small files (< few MB)
with open("notes.txt", encoding="utf-8") as f:
    contents = f.read()          # entire file as one string

with open("notes.txt", encoding="utf-8") as f:
    lines = f.readlines()        # list of strings, each with \n

# ── Lazy line iteration — best for large files ───────────────────────────────
# Never loads the whole file; processes one line at a time from the OS buffer
line_count = 0
with open("notes.txt", encoding="utf-8") as f:
    for line in f:               # f is its own iterator
        line = line.rstrip("\n")
        line_count += 1

# ── Binary files ─────────────────────────────────────────────────────────────
# Use "rb"/"wb" for images, PDFs, any non-text data — no encoding arg
with open("image.png", "rb") as f:
    header = f.read(8)           # read first 8 bytes
    f.seek(0)                    # rewind to start
    data = f.read()              # read all bytes

# ── pathlib one-liners ────────────────────────────────────────────────────────
Path("notes.txt").read_text(encoding="utf-8")    # entire file as str
Path("notes.txt").write_text("content\n", encoding="utf-8")  # write + close
Path("image.png").read_bytes()                   # entire file as bytes

# Implementing the protocol from scratch — instructive, though
# @contextmanager is simpler for most cases (see below)

class ManagedFile:
    def __init__(self, path, mode="r", encoding="utf-8"):
        self.path     = path
        self.mode     = mode
        self.encoding = encoding
        self.file     = None

    def __enter__(self):
        self.file = open(self.path, self.mode, encoding=self.encoding)
        return self.file          # this becomes the `as f` variable

    def __exit__(self, exc_type, exc_val, exc_tb):
        if self.file:
            self.file.close()
        return False              # don't suppress exceptions

with ManagedFile("notes.txt") as f:
    content = f.read()
# f.close() guaranteed, even if f.read() raised an exception

from pathlib import Path

# Constructing paths
base = Path("data")
report = base / "reports" / "q2_2026.csv"   # OS-correct separators
config  = Path.home() / ".config" / "app" / "settings.json"
here    = Path(__file__).parent              # directory containing this script

# Decomposing paths
p = Path("/projects/app/data/report_2026.csv")
p.name         # "report_2026.csv"     — filename with extension
p.stem         # "report_2026"         — filename without extension
p.suffix       # ".csv"                — extension including dot
p.suffixes     # ['.csv']              — list (e.g., ['.tar', '.gz'])
p.parent       # Path('/projects/app/data')
p.parents[1]   # Path('/projects/app')
p.parts        # ('/', 'projects', 'app', 'data', 'report_2026.csv')

# Testing existence and type
p.exists()
p.is_file()
p.is_dir()
p.is_symlink()

# Creating directories
(base / "output").mkdir(parents=True, exist_ok=True)
# parents=True: creates intermediate dirs
# exist_ok=True: doesn't raise if already exists

# Listing files
for f in Path("data").iterdir():         # all entries in directory
    print(f.name)

for f in Path(".").glob("**/*.py"):      # recursive glob
    print(f)

for f in Path(".").rglob("*.csv"):       # rglob = recursive glob shorthand
    print(f.relative_to(Path(".")))      # path relative to base

# Reading/writing (convenience wrappers)
text = p.read_text(encoding="utf-8")
p.write_text("new content\n", encoding="utf-8")
raw  = p.read_bytes()
p.write_bytes(b"\x89PNG\r\n")

# Renaming and deleting
p.rename(p.with_suffix(".tsv"))         # change extension
p.unlink(missing_ok=True)               # delete file, OK if missing
Path("empty_dir").rmdir()               # delete empty directory

from contextlib import contextmanager, suppress
import time, tempfile, shutil
from pathlib import Path

# ── 1. Timer context manager ──────────────────────────────────────────────────
@contextmanager
def timer(label: str):
    start = time.perf_counter()
    try:
        yield                        # body runs here
    finally:                         # guaranteed even on exception
        elapsed = time.perf_counter() - start
        print(f"[{label}] {elapsed:.4f}s")

with timer("data load"):
    data = list(range(1_000_000))
# [data load] 0.0521s

# ── 2. Temporary directory that auto-cleans up ────────────────────────────────
@contextmanager
def temp_workspace():
    """Create a temp dir, yield its Path, then delete it on exit."""
    workspace = Path(tempfile.mkdtemp())
    try:
        yield workspace
    finally:
        shutil.rmtree(workspace, ignore_errors=True)

with temp_workspace() as ws:
    (ws / "output.txt").write_text("hello", encoding="utf-8")
    files = list(ws.iterdir())
    print(files)   # [PosixPath('/tmp/tmp.../output.txt')]
# workspace deleted here — always, even if exception was raised

# ── 3. Database transaction (pattern) ─────────────────────────────────────────
@contextmanager
def transaction(conn):
    """Commit on success, rollback on any exception."""
    try:
        yield conn
        conn.commit()
    except Exception:
        conn.rollback()
        raise                        # re-raise after cleanup

# with transaction(db_conn) as conn:
#     conn.execute("INSERT ...")
#     conn.execute("UPDATE ...")
# commits atomically; any error triggers rollback

# ── 4. contextlib.suppress — swallow specific exceptions ─────────────────────
with suppress(FileNotFoundError):
    Path("might_not_exist.txt").unlink()   # silently skipped if missing

def parse_age(raw: str) -> int:
    """
    Full try/except/else/finally anatomy:
    - try:     code that might fail
    - except:  runs ONLY if the try block raised that exception
    - else:    runs ONLY if the try block did NOT raise
    - finally: ALWAYS runs — even with return or re-raise
    """
    try:
        age = int(raw)
    except ValueError as exc:
        # Catch ONLY what you can handle
        # raise ... from exc preserves the original exception as __cause__
        raise ValueError(f"age must be an integer, got {raw!r}") from exc
    else:
        # Only reached if int(raw) succeeded
        if age < 0 or age > 150:
            raise ValueError(f"age {age} is outside valid range [0, 150]")
        return age
    finally:
        # Always runs — cleanup, logging, metrics
        print(f"parse_age({raw!r}) attempt completed")

# Multiple except clauses — most specific first
def fetch_record(record_id: int) -> dict:
    try:
        return database.get(record_id)
    except KeyError:
        raise LookupError(f"Record {record_id} not found") from None
    except ConnectionError as exc:
        raise RuntimeError("Database unreachable") from exc
    except Exception as exc:
        # Catch-all for truly unexpected errors — log, then re-raise
        logger.error("Unexpected error in fetch_record: %s", exc)
        raise

# Catching multiple types in one clause
try:
    result = risky_io_operation()
except (TimeoutError, ConnectionRefusedError) as exc:
    log.warning("Transient network error: %s", exc)
    result = fallback_value

# Define a domain exception hierarchy
class AppError(Exception):
    """Base for all application errors — callers can catch this broadly."""
    pass

class ValidationError(AppError):
    """Input data failed validation."""
    def __init__(self, field: str, value, message: str):
        super().__init__(f"{field}={value!r}: {message}")
        self.field   = field
        self.value   = value
        self.message = message

class NotFoundError(AppError):
    """Requested resource does not exist."""
    def __init__(self, resource: str, identifier):
        super().__init__(f"{resource} {identifier!r} not found")
        self.resource   = resource
        self.identifier = identifier

class ServiceError(AppError):
    """Downstream service call failed."""
    def __init__(self, service: str, cause: Exception):
        super().__init__(f"{service} failed: {cause}")
        self.service = service

# Usage: precise exception types let callers handle each case
def create_order(data: dict) -> dict:
    if not data.get("customer_id"):
        raise ValidationError("customer_id", data.get("customer_id"),
                               "is required")

    try:
        customer = customer_service.get(data["customer_id"])
    except RequestException as exc:
        raise ServiceError("CustomerService", exc) from exc

    if not customer:
        raise NotFoundError("Customer", data["customer_id"])

    return order_service.create(data)

# Callers can handle each case precisely
try:
    order = create_order(payload)
except ValidationError as e:
    return {"error": "invalid_input", "field": e.field}
except NotFoundError as e:
    return {"error": "not_found", "resource": e.resource}
except ServiceError as e:
    log.error("Dependency failure: %s", e)
    return {"error": "service_unavailable"}

# LBYL: check before every operation
def get_user_email_lbyl(data: dict, user_id: str) -> str | None:
    if not isinstance(data, dict):
        return None
    if user_id not in data:
        return None
    user = data[user_id]
    if not isinstance(user, dict):
        return None
    if "email" not in user:
        return None
    return user["email"]

# EAFP: just try it, handle failures
def get_user_email_eafp(data: dict, user_id: str) -> str | None:
    try:
        return data[user_id]["email"]
    except (KeyError, TypeError):
        return None

# EAFP is also race-condition safe:
# LBYL: file could be deleted between os.path.exists() check and open()
import os
# BAD:
if os.path.exists("config.json"):         # check
    with open("config.json") as f:        # open — race condition window!
        data = f.read()

# GOOD: EAFP handles the same case without the race
try:
    with open("config.json") as f:
        data = f.read()
except FileNotFoundError:
    data = "{}"    # default config

import json
from pathlib import Path
from datetime import datetime
from decimal import Decimal

# Basic serialization
record = {
    "user": "ada",
    "scores": [90, 85, 92],
    "active": True,
    "metadata": None,
}
text = json.dumps(record)                     # compact: no whitespace
text = json.dumps(record, indent=2)           # pretty-printed
text = json.dumps(record, indent=2, sort_keys=True)  # sorted keys

# Deserialization
obj  = json.loads(text)                       # string → Python object
obj  = json.loads(Path("data.json").read_text(encoding="utf-8"))

# File I/O — preferred over read_text + loads
with open("data.json", "w", encoding="utf-8") as f:
    json.dump(record, f, indent=2, ensure_ascii=False)  # non-ASCII preserved

with open("data.json", encoding="utf-8") as f:
    data = json.load(f)

# Custom encoder for non-JSON-native types
class AppEncoder(json.JSONEncoder):
    def default(self, obj):
        if isinstance(obj, datetime):
            return obj.isoformat()          # "2026-06-27T14:32:01"
        if isinstance(obj, Decimal):
            return float(obj)               # or str for exact precision
        if isinstance(obj, set):
            return sorted(list(obj))        # sets → sorted lists
        return super().default(obj)

complex_obj = {
    "created_at": datetime(2026, 6, 27, 14, 32, 1),
    "price":      Decimal("19.99"),
    "tags":       {"python", "api"},
}
json.dumps(complex_obj, cls=AppEncoder, indent=2)
# {
#   "created_at": "2026-06-27T14:32:01",
#   "price": 19.99,
#   "tags": ["api", "python"]
# }

import csv
from pathlib import Path

# ── Reading with DictReader ───────────────────────────────────────────────────
# newline="" is required on all platforms to prevent double-newline on Windows
with open("customers.csv", newline="", encoding="utf-8") as f:
    reader = csv.DictReader(f)
    # reader.fieldnames is populated from the header row
    for row in reader:
        print(row["name"], row["email"], row["spend"])
        # All values are strings — convert types yourself

# Handling missing columns gracefully
with open("customers.csv", newline="", encoding="utf-8") as f:
    reader = csv.DictReader(f)
    records = []
    for row in reader:
        records.append({
            "name":  row["name"].strip(),
            "email": row["email"].lower().strip(),
            "spend": float(row.get("spend", "0") or "0"),
        })

# ── Writing with DictWriter ───────────────────────────────────────────────────
fieldnames = ["id", "name", "email", "spend"]
with open("output.csv", "w", newline="", encoding="utf-8") as f:
    writer = csv.DictWriter(f, fieldnames=fieldnames)
    writer.writeheader()
    writer.writerow({"id": 1, "name": "Ada",  "email": "ada@x.io", "spend": 299.0})
    writer.writerow({"id": 2, "name": "Bob",  "email": "bob@y.io", "spend": 85.0})
    writer.writerows([                               # batch write
        {"id": 3, "name": "Carol", "email": "c@z.io", "spend": 412.0},
        {"id": 4, "name": "Dave",  "email": "d@w.io", "spend": 55.0},
    ])

# ── Tab-separated and custom delimiters ──────────────────────────────────────
with open("data.tsv", newline="", encoding="utf-8") as f:
    reader = csv.DictReader(f, delimiter="\t")  # TSV
    for row in reader:
        ...

# ── Reading from in-memory string (useful for testing) ───────────────────────
import io
csv_text = "name,score\nAda,95\nBob,82\n"
reader = csv.DictReader(io.StringIO(csv_text))
rows = list(reader)
# [{'name': 'Ada', 'score': '95'}, {'name': 'Bob', 'score': '82'}]

import pickle, json
from pathlib import Path

# pickle: full Python objects — TRUSTED DATA ONLY
data = {"model": some_sklearn_model, "created": datetime.now()}
Path("model.pkl").write_bytes(pickle.dumps(data))
loaded = pickle.loads(Path("model.pkl").read_bytes())  # ONLY safe for your own files

# NEVER do this with external data:
# user_data = request.body
# obj = pickle.loads(user_data)   # REMOTE CODE EXECUTION VULNERABILITY

# For ML models specifically: use joblib (wrapper around pickle with compression)
# import joblib; joblib.dump(model, "model.joblib"); joblib.load("model.joblib")

# For configs and API data: JSON is always the right answer
# For dataclasses → dict for JSON: use dataclasses.asdict()
from dataclasses import dataclass, asdict

@dataclass
class Config:
    host: str
    port: int
    debug: bool

cfg = Config(host="localhost", port=8080, debug=True)
json_str = json.dumps(asdict(cfg), indent=2)
# {"host": "localhost", "port": 8080, "debug": true}

# Restore: pass dict fields back as kwargs
cfg2 = Config(**json.loads(json_str))

from dataclasses import dataclass
from datetime import datetime

class BankAccount:
    bank_name    = "PyBank"              # class attribute: shared by all instances
    _interest    = 0.025                 # class attribute: convention = internal

    def __init__(self, owner: str, initial_balance: float = 0.0):
        self.owner     = owner           # instance attribute
        self._balance  = initial_balance # _ = "internal, not public API"
        self._history: list[dict] = []

    # ── Instance methods ──────────────────────────────────────────────────────
    def deposit(self, amount: float) -> None:
        if amount <= 0:
            raise ValueError(f"Deposit amount must be positive, got {amount}")
        self._balance += amount
        self._history.append({"type": "deposit", "amount": amount,
                               "ts": datetime.utcnow().isoformat()})

    def withdraw(self, amount: float) -> None:
        if amount <= 0:
            raise ValueError("Withdrawal must be positive")
        if amount > self._balance:
            raise ValueError(f"Insufficient funds: {self._balance:.2f} < {amount:.2f}")
        self._balance -= amount
        self._history.append({"type": "withdraw", "amount": amount,
                               "ts": datetime.utcnow().isoformat()})

    # ── Property: controlled read access ─────────────────────────────────────
    @property
    def balance(self) -> float:
        return round(self._balance, 2)

    @property
    def transaction_count(self) -> int:
        return len(self._history)

    # ── Class method: alternative constructor ────────────────────────────────
    @classmethod
    def from_dict(cls, data: dict) -> "BankAccount":
        """Create an account from a serialized dict (e.g., from a database row)."""
        account = cls(data["owner"], data.get("balance", 0.0))
        return account

    # ── Static method: utility with logical coupling to this class ────────────
    @staticmethod
    def validate_amount(amount) -> bool:
        """Returns True if amount is a valid positive number."""
        return isinstance(amount, (int, float)) and amount > 0

    def __repr__(self) -> str:
        return f"BankAccount(owner={self.owner!r}, balance={self.balance})"

# Usage
acc = BankAccount("Ada", 500.0)
acc.deposit(250)
acc.withdraw(100)
print(acc.balance)         # 650.0
print(acc)                 # BankAccount(owner='Ada', balance=650.0)

# Alternative constructor
acc2 = BankAccount.from_dict({"owner": "Bob", "balance": 200.0})

# Class attribute access
print(BankAccount.bank_name)    # "PyBank"
print(acc.bank_name)            # "PyBank"  — instances inherit class attrs

# Class attribute mutation trap
class Config:
    flags = []   # SHARED — dangerous if you append to it

c1 = Config()
c2 = Config()
c1.flags.append("debug")    # mutates the SHARED class attribute
c2.flags                    # ["debug"]  — c2 is affected!

# CORRECT: create per-instance in __init__
class Config:
    def __init__(self):
        self.flags = []      # each instance gets its own list

c1 = Config(); c2 = Config()
c1.flags.append("debug")
c2.flags                    # []  — unaffected

# Reading vs assigning class attributes through instances
class Counter:
    count = 0               # class attribute

c = Counter()
Counter.count += 1          # modifies the class attribute — all see it
c.count                     # 1

c.count = 99                # creates an INSTANCE attribute, shadowing the class one
Counter.count               # 1  — class attribute unchanged
c.count                     # 99 — instance attribute hides it

class Temperature:
    def __init__(self, celsius: float):
        self.celsius = celsius   # triggers the setter below if defined

    @property
    def celsius(self) -> float:
        return self._celsius

    @celsius.setter
    def celsius(self, value: float) -> None:
        if value < -273.15:
            raise ValueError(f"Temperature below absolute zero: {value}")
        self._celsius = float(value)

    @celsius.deleter
    def celsius(self) -> None:
        raise AttributeError("Temperature cannot be deleted")

    @property
    def fahrenheit(self) -> float:
        """Read-only computed property — no setter."""
        return self._celsius * 9/5 + 32

    @property
    def kelvin(self) -> float:
        return self._celsius + 273.15

t = Temperature(100)
t.fahrenheit      # 212.0   — computed, no stored state
t.kelvin          # 373.15

t.celsius = -10   # triggers setter, validates
t.celsius = -300  # ValueError: Temperature below absolute zero

# Property as a refactor safety net:
# Start with plain attribute: self.radius = r
# Later add validation: @property + setter
# ALL callers that wrote obj.radius += 1 continue to work unchanged