Python Exception Handling Explained — try, except, finally and Real-World Patterns

def read_config_file(file_path: str) -> dict:
    """
    Reads a JSON config file and returns its contents as a dictionary.
    Demonstrates try / except / else / finally working together.
    """
    config_file = None

    try:
        # ATTEMPT: open the file — this can raise FileNotFoundError
        config_file = open(file_path, 'r')

        import json
        # ATTEMPT: parse JSON — this can raise json.JSONDecodeError
        config_data = json.load(config_file)

    except FileNotFoundError:
        # Fires ONLY when the file doesn't exist at that path
        print(f"[ERROR] Config file not found at: {file_path}")
        return {}  # Return a safe default instead of crashing

    except json.JSONDecodeError as parse_error:
        # Fires ONLY when the file exists but contains invalid JSON
        # 'as parse_error' gives us access to the error details
        print(f"[ERROR] Config file has invalid JSON: {parse_error.msg}")
        return {}

    else:
        # Runs ONLY if the try block succeeded with zero exceptions
        # Perfect place for 'success path' logic — keeps it separate from error handling
        print(f"[OK] Config loaded successfully. Keys found: {list(config_data.keys())}")
        return config_data

    finally:
        # Runs ALWAYS — whether an exception happened or not
        # Critical for cleanup: close the file so we don't leak file handles
        if config_file and not config_file.closed:
            config_file.close()
            print("[CLEANUP] File handle closed.")


# --- Test case 1: file exists and is valid JSON ---
# Assume 'settings.json' contains: {"debug": true, "port": 8080}
result = read_config_file("settings.json")
print(f"Result: {result}\n")

# --- Test case 2: file does not exist ---
result = read_config_file("missing.json")
print(f"Result: {result}")

# --- Define a custom exception hierarchy for a payment module ---

class PaymentError(Exception):
    """
    Base exception for all payment-related failures.
    Any caller catching PaymentError will catch all subclasses too.
    """
    def __init__(self, message: str, transaction_id: str):
        # Always call super().__init__() so Python's exception machinery works correctly
        super().__init__(message)
        self.transaction_id = transaction_id  # Attach useful context data


class PaymentDeclinedError(PaymentError):
    """
    Raised when the card issuer declines the charge.
    Carries a decline_code so the caller knows WHY — not just that it failed.
    """
    def __init__(self, transaction_id: str, decline_code: str):
        message = f"Payment declined (code: {decline_code})"
        super().__init__(message, transaction_id)
        self.decline_code = decline_code
        self.is_retryable = False  # Declined cards won't succeed on retry


class PaymentGatewayTimeoutError(PaymentError):
    """
    Raised when the payment gateway doesn't respond in time.
    Unlike a decline, this one IS safe to retry.
    """
    def __init__(self, transaction_id: str, timeout_seconds: int):
        message = f"Gateway timed out after {timeout_seconds}s"
        super().__init__(message, transaction_id)
        self.is_retryable = True  # Timeout might be temporary — caller can retry


# --- Simulate a payment processing function ---

def process_payment(amount: float, card_token: str, transaction_id: str) -> str:
    """
    Attempts to charge a card. Raises specific PaymentError subclasses on failure.
    """
    # Simulate a declined card scenario
    if card_token == "DECLINED_CARD":
        raise PaymentDeclinedError(
            transaction_id=transaction_id,
            decline_code="insufficient_funds"
        )

    # Simulate a gateway timeout
    if card_token == "SLOW_GATEWAY":
        raise PaymentGatewayTimeoutError(
            transaction_id=transaction_id,
            timeout_seconds=30
        )

    return f"Success: charged ${amount:.2f}"


# --- Caller code: handle each failure type differently ---

def checkout(amount: float, card_token: str):
    transaction_id = "TXN-20240815-001"

    try:
        result = process_payment(amount, card_token, transaction_id)
        print(f"[PAYMENT] {result}")

    except PaymentDeclinedError as error:
        # We know exactly why it failed AND that retrying is pointless
        print(f"[DECLINED] Transaction {error.transaction_id} failed: {error}")
        print(f"  Decline reason: {error.decline_code}")
        print(f"  Retryable: {error.is_retryable}")
        # In real code: show user a 'check your card details' message

    except PaymentGatewayTimeoutError as error:
        # Different response — we might queue this for automatic retry
        print(f"[TIMEOUT] Transaction {error.transaction_id} timed out: {error}")
        print(f"  Retryable: {error.is_retryable}")
        # In real code: push transaction_id onto a retry queue

    except PaymentError as error:
        # Catch-all for any other payment failure we didn't specifically handle
        print(f"[PAYMENT ERROR] Unexpected failure for {error.transaction_id}: {error}")


print("=== Test 1: Declined card ===")
checkout(99.99, "DECLINED_CARD")

print("\n=== Test 2: Gateway timeout ===")
checkout(49.99, "SLOW_GATEWAY")

print("\n=== Test 3: Success ===")
checkout(25.00, "VALID_TOKEN")

import sqlite3
from contextlib import contextmanager

# --- Custom exception for our data layer ---

class DatabaseError(Exception):
    """Raised when a database operation fails at the application level."""
    pass


# --- Context manager for safe database transactions ---

@contextmanager
def managed_transaction(db_path: str):
    """
    A context manager that handles the full lifecycle of a database connection:
    - Opens the connection
    - Commits if the block succeeds
    - Rolls back if any exception occurs
    - Always closes the connection

    Usage:
        with managed_transaction('mydb.sqlite') as cursor:
            cursor.execute(...)
    """
    connection = sqlite3.connect(db_path)
    cursor = connection.cursor()

    try:
        # Yield the cursor to the 'with' block — execution pauses here
        yield cursor

        # If we reach this line, the 'with' block completed without exceptions
        connection.commit()
        print("[DB] Transaction committed.")

    except Exception as db_exception:
        # Something went wrong inside the 'with' block — roll back every change
        connection.rollback()
        print(f"[DB] Transaction rolled back due to: {db_exception}")

        # EXCEPTION CHAINING: wrap the low-level sqlite error in our domain error.
        # 'raise ... from db_exception' preserves the original traceback as __cause__.
        # Callers see a clean DatabaseError, but the full sqlite context is still there
        # for debugging when they inspect the traceback.
        raise DatabaseError(
            f"Failed to complete database operation: {db_exception}"
        ) from db_exception

    finally:
        # Runs regardless — closes connection even if commit or rollback failed
        connection.close()
        print("[DB] Connection closed.")


# --- Application-level function using the context manager ---

def save_user(db_path: str, username: str, email: str) -> None:
    """
    Saves a new user record. If anything fails, the whole transaction rolls back.
    """
    try:
        with managed_transaction(db_path) as cursor:
            # Create table if needed (idempotent)
            cursor.execute("""
                CREATE TABLE IF NOT EXISTS users (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    username TEXT UNIQUE NOT NULL,
                    email TEXT NOT NULL
                )
            """)

            # This INSERT will fail if username already exists (UNIQUE constraint)
            cursor.execute(
                "INSERT INTO users (username, email) VALUES (?, ?)",
                (username, email)
            )
            print(f"[APP] User '{username}' queued for insert.")

    except DatabaseError as error:
        # We catch our domain-level error here
        print(f"[APP] Could not save user: {error}")
        # The original sqlite3 error is still accessible at error.__cause__
        print(f"[APP] Root cause: {error.__cause__}")


# --- Test it ---
DB_FILE = ":memory:"  # In-memory SQLite — no file needed, perfect for demos

print("=== Insert first user ===")
save_user(DB_FILE, "alice", "alice@example.com")
# Note: :memory: databases are fresh per-connection, so duplicate test
# needs a persistent file. We'll simulate a constraint violation differently.

print("\n=== Simulate a broken operation ===")
try:
    with managed_transaction(DB_FILE) as cursor:
        cursor.execute("SELECT * FROM nonexistent_table")  # This will fail
except DatabaseError as error:
    print(f"[APP] Caught at top level: {error}")

import json
from typing import Optional

# --- Simulate a simplified web request/response cycle ---

class HttpResponse:
    def __init__(self, status_code: int, body: dict):
        self.status_code = status_code
        self.body = body

    def __repr__(self):
        return f"HttpResponse({self.status_code}, {self.body})"


# --- Low-level parser — only handles what it knows about ---

def parse_request_body(raw_body: str) -> dict:
    """
    Parses a JSON string into a dict.
    DOES NOT catch exceptions — it has no idea what to do with a parse failure.
    Lets json.JSONDecodeError propagate to whoever called it.
    """
    # No try/except here — this function's job is parsing, not error handling
    return json.loads(raw_body)


def validate_user_payload(payload: dict) -> None:
    """
    Checks that required fields are present.
    Raises ValueError with a descriptive message — doesn't catch anything.
    """
    required_fields = {"username", "email", "password"}
    missing = required_fields - payload.keys()

    if missing:
        # Raise with enough detail for the handler to build a useful error response
        raise ValueError(f"Missing required fields: {sorted(missing)}")


# --- High-level handler — THIS is the right place to catch exceptions ---
# It has enough context to turn failures into proper HTTP responses.

def handle_register_request(raw_request_body: str) -> HttpResponse:
    """
    Handles a user registration API request.
    This is the ONLY layer that should catch exceptions — because it's the only
    layer that knows how to turn them into HTTP responses the client understands.
    """
    try:
        # Step 1: Parse — could raise json.JSONDecodeError
        payload = parse_request_body(raw_request_body)

        # Step 2: Validate — could raise ValueError
        validate_user_payload(payload)

        # Step 3: (In real code: save to DB, hash password, etc.)
        username = payload["username"]

    except json.JSONDecodeError:
        # We CAN handle this here: it means 400 Bad Request
        return HttpResponse(
            status_code=400,
            body={"error": "Request body must be valid JSON"}
        )

    except ValueError as validation_error:
        # We CAN handle this here: it means 422 Unprocessable Entity
        return HttpResponse(
            status_code=422,
            body={"error": str(validation_error)}
        )

    # Note: We do NOT catch Exception here.
    # If something unexpected blows up (DB is down, OOM), let it propagate
    # to the framework's top-level error handler, which will log it properly
    # and return a 500 without leaking stack traces to the client.

    else:
        return HttpResponse(
            status_code=201,
            body={"message": f"User '{username}' registered successfully"}
        )


# --- Run test cases ---

print("=== Valid registration request ===")
valid_body = '{"username": "bob", "email": "bob@example.com", "password": "s3cure!"}'
response = handle_register_request(valid_body)
print(response)

print("\n=== Malformed JSON ===")
response = handle_register_request("{this is not json}")
print(response)

print("\n=== Missing fields ===")
partial_body = '{"username": "carol"}'
response = handle_register_request(partial_body)
print(response)

import asyncio
import logging

# Configure the event loop to log unhandled exceptions
logging.basicConfig(level=logging.ERROR, format='%(asctime)s - %(levelname)s - %(message)s')

async def risky_io() -> str:
    """Simulates an I/O operation that fails randomly."""
    await asyncio.sleep(0.1)
    raise ConnectionError("Database connection refused")

async def safe_worker():
    """Properly awaits and catches exceptions."""
    try:
        result = await risky_io()
        print(f"Result: {result}")
    except ConnectionError as e:
        logging.exception(f"Worker failed with connection error: {e}")
        return None

async def fire_and_forget_worker():
    """Schedules a coroutine but never awaits it — exception is lost."""
    task = asyncio.create_task(risky_io())
    # Not awaited: task executes but its exception disappears into the loop's exception handler
    # This is almost always a bug
    await asyncio.sleep(0.2)  # Wait long enough for the task to fail
    # By now, the exception has been swallowed by the event loop

async def task_group_worker():
    """Uses TaskGroup to ensure all exceptions are surfaced."""
    try:
        async with asyncio.TaskGroup() as tg:
            task1 = tg.create_task(risky_io())
            task2 = tg.create_task(risky_io())  # This will trigger cancellation of task1
    except* ConnectionError as e:
        # Python 3.11+ exception groups allow catching multiple exceptions
        logging.exception(f"TaskGroup failed with {len(e.exceptions)} connection errors")

async def main():
    print("=== Safe worker (catches exception correctly) ===")
    await safe_worker()

    print("\n=== Fire-and-forget worker (exception lost) ===")
    await fire_and_forget_worker()
    print("[WARNING] No exception printed above — it was swallowed by the event loop.")

    print("\n=== TaskGroup worker (surfaces all exceptions) ===")
    await task_group_worker()

if __name__ == "__main__":
    asyncio.run(main())

# io.thecodeforge
# Bare except — the production incident starter pack
import sys

try:
    process_payment(order_id=42, amount=-50.00)
except:
    # Swallows KeyboardInterrupt, SystemExit, everything
    print("Payment failed")  # Logs nothing, re-raises nothing

# The correct pattern — catch only what you expect
try:
    process_payment(order_id=42, amount=-50.00)
except ValueError as e:
    logger.critical("Invalid payment amount: %s", e)
    raise  # Re-raise to prevent silent corruption

# io.thecodeforge
# Bad — loses traceback
import logging

try:
    user = fetch_user(user_id=1001)
except Exception as e:
    print(f"Failed: {e}")  # No stack trace, no context

# Correct — preserves evidence
try:
    user = fetch_user(user_id=1001)
except Exception:
    logger.exception("Failed to fetch user %s", 1001)
    raise  # Let the caller handle retry logic