04. Error Handling, File I/O & Context Managers

def parse_config(path: str) -> dict:
    try:
        with open(path) as f:
            data = json.load(f)
    except FileNotFoundError:
        print(f"Config not found: {path}")
        return {}
    except json.JSONDecodeError as e:
        print(f"Invalid JSON at line {e.lineno}: {e.msg}")
        return {}
    else:
        # Runs ONLY if no exception was raised in try
        # Keep code here that shouldn't be protected by except
        print(f"Loaded {len(data)} keys from config")
        return data
    finally:
        # Always runs — cleanup goes here
        print("Config parse attempt complete")

# Catch several types with one handler
try:
    value = int(user_input)
except (ValueError, TypeError) as e:
    print(f"Bad input: {e}")

# Separate handlers for different types
try:
    result = remote_api_call()
except ConnectionError:
    retry()
except TimeoutError:
    use_cached_value()
except Exception as e:
    # Catch-all — last resort only
    log.error(f"Unexpected: {e}")
    raise  # Re-raise after logging

BaseException
 +-- SystemExit
 +-- KeyboardInterrupt
 +-- GeneratorExit
 +-- Exception
      +-- StopIteration
      +-- ArithmeticError
      |    +-- ZeroDivisionError
      |    +-- OverflowError
      +-- LookupError
      |    +-- IndexError
      |    +-- KeyError
      +-- OSError
      |    +-- FileNotFoundError
      |    +-- PermissionError
      |    +-- IsADirectoryError
      +-- ValueError
      +-- TypeError
      +-- AttributeError
      +-- RuntimeError
      |    +-- RecursionError
      +-- ImportError
           +-- ModuleNotFoundError

def withdraw(account: str, amount: float) -> float:
    if amount <= 0:
        raise ValueError(f"Amount must be positive, got {amount}")
    if amount > get_balance(account):
        raise InsufficientFundsError(account, amount)
    return process_withdrawal(account, amount)

# Re-raise the current exception (preserves traceback)
try:
    dangerous_operation()
except OSError:
    log.error("Operation failed")
    raise  # Re-raises the original OSError

# Exception chaining with `from` — sets __cause__
try:
    data = json.loads(raw)
except json.JSONDecodeError as e:
    raise ConfigError(f"Invalid config format") from e
    # Traceback shows: "The above exception was the direct cause of..."

# Suppress chaining with `from None`
try:
    value = mapping[key]
except KeyError:
    raise PublicAPIError(f"Unknown field: {key}") from None
    # Hides the internal KeyError from the user-facing traceback

class AppError(Exception):
    """Base for all application errors."""
    pass

class InsufficientFundsError(AppError):
    def __init__(self, account: str, amount: float):
        self.account = account
        self.amount = amount
        super().__init__(
            f"Account {account} has insufficient funds for {amount:.2f}"
        )

class RateLimitError(AppError):
    def __init__(self, retry_after: int):
        self.retry_after = retry_after
        super().__init__(f"Rate limited. Retry after {retry_after}s")

# Callers can now branch on attributes
try:
    api_request()
except RateLimitError as e:
    time.sleep(e.retry_after)
    api_request()  # Retry

# LBYL — the Java/C way (not Pythonic)
if key in mapping:
    value = mapping[key]
else:
    value = default

# EAFP — the Python way
try:
    value = mapping[key]
except KeyError:
    value = default

# Even better — use the API designed for it
value = mapping.get(key, default)

# LBYL — race condition if file is deleted between check and open
import os
if os.path.exists(path):
    f = open(path)  # File could vanish here

# EAFP — atomic, no race condition
try:
    f = open(path)
except FileNotFoundError:
    handle_missing()

# Always use `with` — it closes the file automatically
with open("data.txt", "r", encoding="utf-8") as f:
    content = f.read()          # Entire file as one string

with open("data.txt", encoding="utf-8") as f:
    lines = f.readlines()       # List of lines (includes \n)

with open("data.txt", encoding="utf-8") as f:
    lines = f.read().splitlines()  # List of lines (strips \n)

# Best for large files — iterate line by line (lazy, memory-efficient)
with open("huge.log", encoding="utf-8") as f:
    for line in f:
        process(line.rstrip("\n"))

# Write (creates or truncates)
with open("output.txt", "w", encoding="utf-8") as f:
    f.write("line one\n")
    f.write("line two\n")

# Append
with open("log.txt", "a", encoding="utf-8") as f:
    f.write(f"{timestamp}: event occurred\n")

# Write multiple lines at once
lines = ["alpha\n", "beta\n", "gamma\n"]
with open("output.txt", "w", encoding="utf-8") as f:
    f.writelines(lines)  # Does NOT add newlines — you must include them

# Binary mode
with open("image.png", "rb") as f:
    header = f.read(8)  # Read first 8 bytes

with open("copy.png", "wb") as f:
    f.write(header)

# Without with — error-prone
f = open("data.txt")
try:
    data = f.read()
finally:
    f.close()

# With with — clean and safe
with open("data.txt") as f:
    data = f.read()
# f is closed here, guaranteed

# Multiple context managers (Python 3.10+ parenthesized form)
with (
    open("input.txt") as src,
    open("output.txt", "w") as dst,
):
    dst.write(src.read())

from pathlib import Path

# Creating paths
config = Path("/etc/myapp/config.json")
home = Path.home()
cwd = Path.cwd()

# Joining — use / operator (yes, really)
data_dir = Path("project") / "data" / "raw"
log_file = data_dir / "app.log"

# Path components
p = Path("/home/user/docs/report.tar.gz")
p.name          # "report.tar.gz"
p.stem          # "report.tar"
p.suffix        # ".gz"
p.suffixes      # [".tar", ".gz"]
p.parent        # Path("/home/user/docs")
p.parts         # ("/", "home", "user", "docs", "report.tar.gz")

# Querying
p.exists()
p.is_file()
p.is_dir()
p.stat().st_size  # File size in bytes

# Globbing
for py_file in Path("src").rglob("*.py"):
    print(py_file)

# Reading and writing (convenience methods)
text = Path("config.json").read_text(encoding="utf-8")
Path("output.txt").write_text("hello\n", encoding="utf-8")
raw = Path("image.png").read_bytes()

# Creating directories
Path("logs/2024/03").mkdir(parents=True, exist_ok=True)

class Timer:
    def __enter__(self):
        self.start = time.perf_counter()
        return self  # Value bound by `as`

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.elapsed = time.perf_counter() - self.start
        print(f"Elapsed: {self.elapsed:.3f}s")
        return False  # Don't suppress exceptions

with Timer() as t:
    heavy_computation()
print(f"Took {t.elapsed:.3f}s")

from contextlib import contextmanager

@contextmanager
def temporary_env(key: str, value: str):
    old = os.environ.get(key)
    os.environ[key] = value
    try:
        yield  # Control returns to the `with` block here
    finally:
        if old is None:
            del os.environ[key]
        else:
            os.environ[key] = old

with temporary_env("DEBUG", "1"):
    run_debug_mode()
# Original env restored here

@contextmanager
def managed_db_connection(url: str):
    conn = create_connection(url)
    try:
        yield conn
    except Exception:
        conn.rollback()
        raise
    else:
        conn.commit()
    finally:
        conn.close()

with managed_db_connection("postgres://...") as conn:
    conn.execute("INSERT INTO users ...")
# Auto-committed if no error, rolled back if exception, always closed

import json
import csv
from pathlib import Path

# JSON round-trip
def load_json(path: Path) -> dict:
    return json.loads(path.read_text(encoding="utf-8"))

def save_json(path: Path, data: dict) -> None:
    path.write_text(
        json.dumps(data, indent=2, ensure_ascii=False) + "\n",
        encoding="utf-8",
    )

# CSV reading — DictReader gives you dicts, not lists
with open("users.csv", newline="", encoding="utf-8") as f:
    reader = csv.DictReader(f)
    users = [row for row in reader]  # List of {"name": ..., "email": ...}

# CSV writing
with open("output.csv", "w", newline="", encoding="utf-8") as f:
    writer = csv.DictWriter(f, fieldnames=["name", "email"])
    writer.writeheader()
    writer.writerows(users)

import tempfile
from pathlib import Path

# Temp file — auto-deleted when context exits
with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=True) as tmp:
    tmp.write('{"key": "value"}')
    tmp.flush()
    # Use tmp.name to pass the path to other tools
    process_file(tmp.name)
# File is gone here

# Temp directory
with tempfile.TemporaryDirectory() as tmpdir:
    p = Path(tmpdir) / "scratch.txt"
    p.write_text("temporary data")
    # Entire directory is deleted on exit

from contextlib import contextmanager
from pathlib import Path
import tempfile
import os

@contextmanager
def atomic_write(target: Path, mode: str = "w", **kwargs):
    """Write to a temp file, then rename to target.
    If anything fails, the original file is untouched."""
    tmp_fd, tmp_path = tempfile.mkstemp(
        dir=target.parent, suffix=".tmp"
    )
    try:
        with open(tmp_fd, mode, **kwargs) as f:
            yield f
        os.replace(tmp_path, target)  # Atomic on POSIX
    except BaseException:
        os.unlink(tmp_path)  # Clean up temp file
        raise

# Usage
with atomic_write(Path("config.json"), encoding="utf-8") as f:
    json.dump(config_data, f, indent=2)
# If json.dump raises, config.json is never corrupted

from contextlib import suppress

# Instead of try/except/pass
with suppress(FileNotFoundError):
    Path("cache.tmp").unlink()

# Equivalent to:
try:
    Path("cache.tmp").unlink()
except FileNotFoundError:
    pass