Python try-except-finally Explained — How to Handle Errors Like a Pro

# basic_exception_anatomy.py
# Demonstrates every block of a try-except-else-finally structure
# with a real file-reading scenario so the purpose is obvious.

def read_user_config(filepath: str) -> dict:
    """
    Reads a simple key=value config file and returns it as a dict.
    Shows exactly which block handles which scenario.
    """
    config_file = None  # declare outside so finally can reference it safely

    try:
        # --- TRY: only the operation that can actually fail ---
        config_file = open(filepath, 'r', encoding='utf-8')
        raw_contents = config_file.read()  # could raise IOError

    except FileNotFoundError:
        # --- EXCEPT: respond to a SPECIFIC, anticipated failure ---
        # FileNotFoundError is a subclass of OSError — catch the most specific one
        print(f"[CONFIG] File '{filepath}' not found. Using default settings.")
        return {}  # graceful fallback — caller gets an empty dict, not a crash

    except PermissionError:
        # --- EXCEPT: a different failure needs a different response ---
        print(f"[CONFIG] No read permission for '{filepath}'. Check file permissions.")
        return {}

    else:
        # --- ELSE: runs ONLY when try succeeded with zero exceptions ---
        # Perfect place for processing that depends on the operation working
        print(f"[CONFIG] File read successfully. Parsing {len(raw_contents)} characters.")
        parsed = {}
        for line in raw_contents.splitlines():
            line = line.strip()
            if '=' in line and not line.startswith('#'):  # skip comments
                key, _, value = line.partition('=')
                parsed[key.strip()] = value.strip()
        return parsed

    finally:
        # --- FINALLY: ALWAYS runs — the perfect cleanup zone ---
        # Even if except hits a return, this block still executes before the
        # function actually returns to the caller.
        if config_file is not None:  # guard: only close if we managed to open
            config_file.close()
            print("[CONFIG] File handle closed.")


# --- Run scenario 1: file doesn't exist ---
print("=== Scenario 1: Missing file ===")
result = read_user_config("settings.cfg")
print(f"Result: {result}")

print()

# --- Run scenario 2: create a real file and read it ---
import os
test_filepath = "test_settings.cfg"
with open(test_filepath, 'w') as f:
    f.write("# App configuration\ntheme = dark\nfont_size = 14\ntimeout = 30\n")

print("=== Scenario 2: Valid file ===")
result = read_user_config(test_filepath)
print(f"Result: {result}")

os.remove(test_filepath)  # tidy up after ourselves

# database_connection_cleanup.py
# Simulates a database connection lifecycle to prove WHY finally
# is non-negotiable for resource cleanup.
# Uses sqlite3 (built-in) so this runs anywhere with zero setup.

import sqlite3

def fetch_active_users(db_path: str, min_login_count: int) -> list:
    """
    Fetches users who have logged in more than `min_login_count` times.
    Demonstrates that finally runs even when an exception is re-raised.
    """
    connection = None  # must be None so finally doesn't crash if open() fails

    try:
        print("[DB] Opening connection...")
        connection = sqlite3.connect(db_path)
        cursor = connection.cursor()

        # Create a demo table if it doesn't exist
        cursor.execute("""
            CREATE TABLE IF NOT EXISTS users (
                id INTEGER PRIMARY KEY,
                username TEXT NOT NULL,
                login_count INTEGER DEFAULT 0
            )
        """)
        # Seed some data
        cursor.executemany(
            "INSERT OR IGNORE INTO users (id, username, login_count) VALUES (?, ?, ?)",
            [(1, 'alice', 42), (2, 'bob', 3), (3, 'carol', 17)]
        )
        connection.commit()

        # This is the operation that could realistically fail in production
        # (e.g., table doesn't exist, column name wrong, DB locked)
        cursor.execute(
            "SELECT username, login_count FROM users WHERE login_count > ?",
            (min_login_count,)  # parameterized query — never string-format SQL
        )
        rows = cursor.fetchall()
        return rows  # finally STILL runs before this return reaches the caller

    except sqlite3.OperationalError as db_error:
        # Catch database-specific errors separately so we can log context
        print(f"[DB] Query failed — OperationalError: {db_error}")
        raise  # re-raise so the caller knows something went wrong
                # finally STILL runs even though we're re-raising

    except Exception as unexpected_error:
        # Catch-all for truly unexpected issues — log, then re-raise
        print(f"[DB] Unexpected error: {type(unexpected_error).__name__}: {unexpected_error}")
        raise

    finally:
        # This block is GUARANTEED to execute — whether we returned,
        # re-raised, or completed normally. The connection WILL be closed.
        if connection is not None:
            connection.close()
            print("[DB] Connection closed.")


# --- Demo: normal execution ---
print("=== Query: users with more than 10 logins ===")
try:
    active_users = fetch_active_users(":memory:", min_login_count=10)
    print(f"[RESULT] Found {len(active_users)} user(s): {active_users}")
except Exception:
    print("[APP] Could not retrieve users.")

# api_client_with_retry.py
# A production-style pattern: custom exceptions, specific catching,
# retry logic, and clean separation of concerns.
# Uses urllib (built-in) — no external dependencies needed.

import urllib.request
import urllib.error
import time
import logging

# --- Setup structured logging (production code always uses logging, not print) ---
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s [%(levelname)s] %(message)s',
    datefmt='%H:%M:%S'
)
logger = logging.getLogger(__name__)


# --- Custom exception: carries domain context, not just a raw HTTP code ---
class ApiRequestError(Exception):
    """Raised when an API call fails after all retry attempts are exhausted."""
    def __init__(self, url: str, status_code: int, message: str):
        self.url = url
        self.status_code = status_code
        super().__init__(f"API request to '{url}' failed [{status_code}]: {message}")


class NetworkUnavailableError(Exception):
    """Raised when we can't reach the host at all — not the same as a bad response."""
    pass


def fetch_url_with_retry(
    url: str,
    max_attempts: int = 3,
    retry_delay_seconds: float = 1.0
) -> str:
    """
    Fetches a URL and returns the response body as a string.
    Retries on transient network errors, raises custom exceptions on hard failures.
    """
    last_exception = None

    for attempt_number in range(1, max_attempts + 1):
        try:
            logger.info(f"Attempt {attempt_number}/{max_attempts} → {url}")

            # urllib.request.urlopen raises HTTPError for 4xx/5xx,
            # URLError for network-level failures (DNS, timeout, refused)
            with urllib.request.urlopen(url, timeout=5) as response:
                # 'with' here handles closing the response — no finally needed
                body = response.read().decode('utf-8')

            # --- else equivalent: only reached if urlopen didn't raise ---
            logger.info(f"Success on attempt {attempt_number}. Body length: {len(body)} chars.")
            return body  # clean exit — finally in urlopen's context manager runs

        except urllib.error.HTTPError as http_err:
            # HTTPError: the server responded, but with an error status.
            # 4xx errors (bad request, not found, unauthorized) won't improve
            # with a retry — raise immediately as our custom domain exception.
            if 400 <= http_err.code < 500:
                logger.error(f"Client error {http_err.code} — not retrying.")
                raise ApiRequestError(url, http_err.code, str(http_err.reason))

            # 5xx errors might be transient — worth retrying
            logger.warning(f"Server error {http_err.code} on attempt {attempt_number}.")
            last_exception = http_err

        except urllib.error.URLError as network_err:
            # URLError: couldn't reach the server at all (DNS failure, refused, timeout)
            logger.warning(f"Network error on attempt {attempt_number}: {network_err.reason}")
            last_exception = network_err

        finally:
            # finally runs after EVERY iteration of the loop — including successful ones.
            # Only do cleanup here — logging retry delay is NOT cleanup, so it's outside.
            pass  # nothing to release here — 'with' above handled the response object

        # Only reached if an exception was caught and we're going to retry
        if attempt_number < max_attempts:
            logger.info(f"Waiting {retry_delay_seconds}s before retry...")
            time.sleep(retry_delay_seconds)

    # All attempts exhausted — decide what to raise based on what failed
    if isinstance(last_exception, urllib.error.URLError):
        raise NetworkUnavailableError(
            f"Could not connect to '{url}' after {max_attempts} attempts."
        ) from last_exception  # 'from' chains exceptions — preserves the traceback

    raise ApiRequestError(url, 0, f"Failed after {max_attempts} attempts.") from last_exception


# --- Caller code: handles only the exceptions it can act on ---
def main():
    test_url = "https://httpbin.org/get"  # reliable public echo endpoint

    try:
        response_body = fetch_url_with_retry(test_url, max_attempts=2)
        # Truncate for readable output
        preview = response_body[:120].replace('\n', ' ')
        print(f"\nResponse preview: {preview}...")

    except NetworkUnavailableError as e:
        logger.error(f"Network is down or URL unreachable: {e}")
        print("Tip: check your internet connection or firewall rules.")

    except ApiRequestError as e:
        logger.error(f"API returned an error: {e}")
        if e.status_code == 404:
            print("The endpoint doesn't exist — check the URL.")
        elif e.status_code == 401:
            print("Authentication failed — check your API key.")
        else:
            print(f"Unhandled API error [{e.status_code}] — escalate to on-call.")


if __name__ == "__main__":
    main()