VALID_PRIORITIES = {"low", "normal", "high", "critical"}

class InvalidPriorityError(Exception):
    """Raised when a priority value isn't in the allowed set."""
    pass

def parse_ticket_age(raw_value):
    """Convert raw input to a non-negative integer age in hours."""
    try:
        age = int(raw_value)
    except ValueError:
        raise ValueError(f"'{raw_value}' is not a whole number")
    if age < 0:
        raise ValueError(f"Age cannot be negative: {age}")
    return age

def parse_priority(raw_value):
    """Validate a priority string against the allowed set."""
    priority = raw_value.strip().lower()
    if priority not in VALID_PRIORITIES:
        raise InvalidPriorityError(
            f"'{raw_value}' is not valid. Choose from: {', '.join(sorted(VALID_PRIORITIES))}"
        )
    return priority

def build_ticket(age_input, priority_input):
    """Validate both fields, returning a clean ticket dict or raising."""
    age      = parse_ticket_age(age_input)
    priority = parse_priority(priority_input)
    return {"age_hours": age, "priority": priority}

# ── Simulated batch of user submissions (some invalid) ──
submissions = [
    ("5",    "high"),
    ("abc",  "normal"),     # bad age
    ("12",   "urgent"),     # bad priority
    ("-3",   "low"),        # negative age
    ("20",   "CRITICAL"),
]

valid_tickets, errors = [], []
for age_in, prio_in in submissions:
    try:
        ticket = build_ticket(age_in, prio_in)
    except (ValueError, InvalidPriorityError) as e:
        errors.append(f"Rejected (age={age_in!r}, priority={prio_in!r}): {e}")
    else:
        valid_tickets.append(ticket)

print(f"Accepted: {len(valid_tickets)}  Rejected: {len(errors)}\n")
for t in valid_tickets:
    print(f"  ✓ {t}")
print()
for err in errors:
    print(f"  ✗ {err}")

import random
import time

class DatabaseConnectionError(Exception):
    """Raised when all connection attempts to the database fail."""
    pass

def attempt_connection(host, fail_chance=0.6):
    """Simulate a flaky connection attempt. Raises ConnectionError on failure."""
    if random.random() < fail_chance:
        raise ConnectionError(f"Could not reach {host}")
    return f"connection_to_{host}"

def connect_with_retry(host, max_attempts=4, delay_seconds=0.3):
    """
    Try to connect, retrying on ConnectionError up to max_attempts times.

    Raises:
        DatabaseConnectionError: if every attempt fails.
    """
    last_error = None
    for attempt in range(1, max_attempts + 1):
        try:
            conn = attempt_connection(host)
        except ConnectionError as e:
            last_error = e
            print(f"  Attempt {attempt}/{max_attempts} failed: {e}")
            time.sleep(delay_seconds)
        else:
            print(f"  Attempt {attempt}/{max_attempts} succeeded")
            return conn
    raise DatabaseConnectionError(
        f"Failed to connect to {host} after {max_attempts} attempts: {last_error}"
    )

# ── Run ──
random.seed(7)   # reproducible for this demo
try:
    print("Connecting to ora-prod-01...")
    conn = connect_with_retry("ora-prod-01")
    print(f"Connected: {conn}")
except DatabaseConnectionError as e:
    print(f"FATAL: {e}")
finally:
    print("Connection routine finished.")

import json
import os

class ConfigError(Exception):
    """Raised for any problem loading or validating deployment config."""
    pass

REQUIRED_KEYS = ["app_name", "db_host", "replicas"]

def load_deployment_config(path):
    """
    Load and validate deployment config.

    Raises:
        ConfigError: for any missing file, bad JSON, or missing required key.
                     The original cause is preserved via 'from'.
    """
    if not os.path.isfile(path):
        raise ConfigError(f"Config file not found: {path}")

    try:
        with open(path, "r", encoding="utf-8") as f:
            config = json.load(f)
    except json.JSONDecodeError as e:
        raise ConfigError(f"Config file is not valid JSON: {e}") from e

    missing = [k for k in REQUIRED_KEYS if k not in config]
    if missing:
        raise ConfigError(f"Config missing required keys: {missing}")

    if not isinstance(config["replicas"], int) or config["replicas"] < 1:
        raise ConfigError(f"'replicas' must be a positive integer, got {config['replicas']!r}")

    return config

# ── Test against three different broken inputs ──
test_files = ["missing_config.json", "bad_json_config.json", "incomplete_config.json"]

for path in test_files:
    try:
        cfg = load_deployment_config(path)
        print(f"✓ {path}: loaded OK — {cfg['app_name']}")
    except ConfigError as e:
        print(f"✗ {path}: {e}")

class InferenceError(Exception):
    """Raised when a single record cannot be processed by the model."""
    pass

def preprocess(record):
    """Validate and normalise a single record before inference."""
    if "text" not in record or not record["text"].strip():
        raise InferenceError("Record has no usable text field")
    if len(record["text"]) > 5000:
        raise InferenceError(f"Text too long ({len(record['text'])} chars, max 5000)")
    return record["text"].strip().lower()

def run_model(text):
    """Simulated model inference — fails on empty input after preprocessing."""
    if not text:
        raise InferenceError("Preprocessed text is empty")
    sentiment = "positive" if "good" in text or "great" in text else "neutral"
    return {"sentiment": sentiment, "length": len(text)}

def process_batch(records):
    """Run inference on a batch, isolating failures per-record."""
    results, failures = [], []
    for i, record in enumerate(records):
        try:
            text   = preprocess(record)
            output = run_model(text)
        except InferenceError as e:
            failures.append({"index": i, "reason": str(e)})
        else:
            results.append({"index": i, **output})
    return results, failures

# ── Run on a mixed batch ──
batch = [
    {"text": "This service is great and fast"},
    {"text": ""},                          # empty — will fail
    {"id": 99},                          # missing text field — will fail
    {"text": "Average response time"},
    {"text": "x" * 6000},                    # too long — will fail
]

results, failures = process_batch(batch)

print(f"Processed: {len(results)}/{len(batch)}  Failed: {len(failures)}/{len(batch)}\n")
for r in results:
    print(f"  ✓ [{r['index']}] {r['sentiment']} ({r['length']} chars)")
for f in failures:
    print(f"  ✗ [{f['index']}] {f['reason']}")

import random
from datetime import datetime

class HealthCheckError(Exception):
    """Base class for all health check failures."""
    pass

class ServerTimeoutError(HealthCheckError):
    """Server did not respond within the timeout window."""
    pass

class ServerAuthError(HealthCheckError):
    """Authentication to the server failed."""
    pass

def check_server(name):
    """Simulate a health check that can fail in different ways."""
    outcome = random.choice(["ok", "ok", "timeout", "auth", "ok"])
    if outcome == "timeout":
        raise ServerTimeoutError(f"{name} did not respond within 5s")
    if outcome == "auth":
        raise ServerAuthError(f"{name} rejected credentials")
    return {"name": name, "status": "healthy"}

def run_health_checks(server_names):
    """Check every server, isolating failures so all servers get checked."""
    healthy, problems = [], []
    for name in server_names:
        try:
            result = check_server(name)
        except ServerTimeoutError as e:
            problems.append({"server": name, "type": "TIMEOUT",  "detail": str(e)})
        except ServerAuthError as e:
            problems.append({"server": name, "type": "AUTH",     "detail": str(e)})
        except HealthCheckError as e:        # catches any future subclass we add later
            problems.append({"server": name, "type": "UNKNOWN", "detail": str(e)})
        else:
            healthy.append(result)
    return healthy, problems

# ── Run ──
random.seed(3)
fleet = ["web-01", "web-02", "db-01", "api-01", "cache-01"]

healthy, problems = run_health_checks(fleet)

print(f"Health Check — {datetime.now().strftime('%H:%M:%S')}")
print(f"Healthy: {len(healthy)}/{len(fleet)}\n")
for h in healthy:
    print(f"  ✓ {h['name']}: {h['status']}")
for p in problems:
    print(f"  ✗ {p['server']} [{p['type']}]: {p['detail']}")