Logging, Assertions and Debugging

import logging

# Configuring global logging threshold and formatting output schema
logging.basicConfig(
    level=logging.DEBUG,  # Enables DEBUG, INFO, WARNING, ERROR, CRITICAL
    format="%(asctime)s | [%(levelname)s] | %(name)s | %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S"
)

logger = logging.getLogger("EnterpriseEngine")

print("=== Triggering Log Events Across All 5 Severity Levels ===")
logger.debug("Database query parameters: host=127.0.0.1, port=5432")
logger.info("Server initialized and listening on port 8080.")
logger.warning("Cache miss threshold reached (82% misses). Performance degraded.")
logger.error("Failed to serialize transaction payload to JSON!")
logger.critical("Database connection pool exhausted! System shutting down.")

=== Triggering Log Events Across All 5 Severity Levels ===
2026-07-26 19:05:00 | [DEBUG] | EnterpriseEngine | Database query parameters: host=127.0.0.1, port=5432
2026-07-26 19:05:00 | [INFO] | EnterpriseEngine | Server initialized and listening on port 8080.
2026-07-26 19:05:00 | [WARNING] | EnterpriseEngine | Cache miss threshold reached (82% misses). Performance degraded.
2026-07-26 19:05:00 | [ERROR] | EnterpriseEngine | Failed to serialize transaction payload to JSON!
2026-07-26 19:05:00 | [CRITICAL] | EnterpriseEngine | Database connection pool exhausted! System shutting down.

import logging
from pathlib import Path

log_file_path = Path.cwd() / "application_audit.log"

# Setup custom logger object
logger = logging.getLogger("FileAuditLogger")
logger.setLevel(logging.INFO)

# Create File Handler for persistent file storage
file_handler = logging.FileHandler(log_file_path, mode="a", encoding="utf-8")
file_formatter = logging.Formatter("%(asctime)s [%(levelname)s] %(message)s")
file_handler.setFormatter(file_formatter)

# Attach handler to logger (preventing duplicate handlers)
if not logger.handlers:
    logger.addHandler(file_handler)

def execute_risky_operation(val_a: float, val_b: float):
    logger.info(f"Executing division task: ({val_a} / {val_b})")
    try:
        result = val_a / val_b
        logger.info(f"Operation completed cleanly. Result: {result:.2f}")
        return result
    except ZeroDivisionError:
        # logger.exception automatically appends full traceback info!
        logger.exception("CRITICAL_FAILURE: Division by zero encountered inside operation block!")
        return None

# Testing file logging with traceback capturing
execute_risky_operation(100.0, 2.0)
execute_risky_operation(50.0, 0.0)

print(f"Audit log saved to: {log_file_path.name}")
print("=== Read Back File Contents ===")
with open(log_file_path, mode="r", encoding="utf-8") as f:
    print(f.read().strip())

Audit log saved to: application_audit.log
=== Read Back File Contents ===
2026-07-26 19:05:05 [INFO] Executing division task: (100.0 / 2.0)
2026-07-26 19:05:05 [INFO] Operation completed cleanly. Result: 50.00
2026-07-26 19:05:05 [INFO] Executing division task: (50.0 / 0.0)
2026-07-26 19:05:05 [ERROR] CRITICAL_FAILURE: Division by zero encountered inside operation block!
Traceback (most recent call last):
  File "main.py", line 18, in execute_risky_operation
    result = val_a / val_b
ZeroDivisionError: division by zero

def calculate_discounted_price(original_price: float, discount_pct: float) -> float:
    """Calculates discounted net price with defensive internal assertion checks."""
    # Internal developer invariant assertions (Catching developer programming bugs)
    assert original_price >= 0.0, f"Invariant Broken: Price cannot be negative! Got: {original_price}"
    assert 0.0 <= discount_pct <= 1.0, f"Invariant Broken: Discount rate must be between 0.0 and 1.0! Got: {discount_pct}"
    
    final_price = original_price * (1.0 - discount_pct)
    
    # Post-condition assertion
    assert final_price <= original_price, "Invariant Broken: Final price exceeded original base price!"
    return round(final_price, 2)

# Valid Invocation
valid_price = calculate_discounted_price(100.0, 0.20)
print(f"Calculated Discounted Price: ${valid_price:.2f}")

# Invalid Invocation (Triggers AssertionError)
try:
    bad_price = calculate_discounted_price(100.0, 1.50)  # Invalid 150% discount!
except AssertionError as assert_err:
    print(f"ASSERTION CAUGHT -> {assert_err}")

Calculated Discounted Price: $80.00
ASSERTION CAUGHT -> Invariant Broken: Discount rate must be between 0.0 and 1.0! Got: 1.5

def process_inventory_batch(items: list):
    """
    Demonstrates programmatic breakpoint invocation simulation.
    In terminal execution, calling breakpoint() pauses execution and opens pdb shell.
    """
    total_value = 0.0
    for idx, item in enumerate(items, start=1):
        # Simulating conditional breakpoint during developer debugging
        if item.get("price", 0) < 0:
            print(f"[DEBUGGER] Anomaly detected at item index #{idx}! Pre-breakpoint state logged.")
            # Real code: breakpoint()  <-- Opens interactive (Pdb) prompt
        
        total_value += item.get("price", 0) * item.get("quantity", 1)
        
    return total_value

sample_batch = [
    {"name": "Mouse", "price": 25.0, "quantity": 4},
    {"name": "Corrupted_Item", "price": -100.0, "quantity": 1},  # Invalid price!
    {"name": "Monitor", "price": 300.0, "quantity": 2}
]

batch_total = process_inventory_batch(sample_batch)
print(f"Processed Batch Net Total: ${batch_total:.2f}")

[DEBUGGER] Anomaly detected at item index #2! Pre-breakpoint state logged.
Processed Batch Net Total: $600.00

import logging

logging.basicConfig(level=logging.INFO, format="[%(levelname)s] %(message)s")
logger = logging.getLogger("DiagnosticDispatcher")

def dispatch_diagnostic_event(event_payload: tuple | dict):
    """
    Routes diagnostic payloads to logging & assertions using match-case.
    Demonstrates evaluating complex diagnostic schemas dynamically.
    """
    match event_payload:
        case ("AUDIT_OK", service, duration_ms) if duration_ms <= 200:
            logger.info(f"Service '{service}' healthy -> Response time: {duration_ms}ms")
            return "STATUS_NORMAL"
            
        case ("AUDIT_OK", service, duration_ms) if duration_ms > 200:
            logger.warning(f"Service '{service}' LATENCY ALERT -> Response time: {duration_ms}ms (>200ms threshold)")
            return "STATUS_LATENCY_WARN"
            
        case {"status": "CRITICAL_FAILURE", "module": mod_name, "code": err_code}:
            logger.critical(f"MODULE '{mod_name}' CRASHED! Code: {err_code}. Escalating to On-Call Engineer...")
            return "STATUS_ESCALATED"
            
        case ("TEST_ASSERT", test_val) if isinstance(test_val, int):
            try:
                assert test_val > 0, f"Diagnostic Assertion Failed: {test_val} is not positive!"
                logger.info("Diagnostic Assertion Passed successfully.")
                return "STATUS_ASSERT_PASS"
            except AssertionError as err:
                logger.error(f"DIAGNOSTIC_FAIL: {err}")
                return "STATUS_ASSERT_FAIL"
            
        case _:
            logger.error(f"Unrecognized diagnostic payload schema -> {event_payload}")
            return "STATUS_UNKNOWN"

# Testing the diagnostic dispatcher
print("Result 1:", dispatch_diagnostic_event(("AUDIT_OK", "AuthService", 45)))
print("Result 2:", dispatch_diagnostic_event(("AUDIT_OK", "PaymentGateway", 450)))
print("Result 3:", dispatch_diagnostic_event({"status": "CRITICAL_FAILURE", "module": "DB_Cluster", "code": 500}))
print("Result 4:", dispatch_diagnostic_event(("TEST_ASSERT", -5)))

[INFO] Service 'AuthService' healthy -> Response time: 45ms
Result 1: STATUS_NORMAL
[WARNING] Service 'PaymentGateway' LATENCY ALERT -> Response time: 450ms (>200ms threshold)
Result 2: STATUS_LATENCY_WARN
[CRITICAL] MODULE 'DB_Cluster' CRASHED! Code: 500. Escalating to On-Call Engineer...
Result 3: STATUS_ESCALATED
[ERROR] DIAGNOSTIC_FAIL: Diagnostic Assertion Failed: -5 is not positive!
Result 4: STATUS_ASSERT_FAIL