Modules, Packages & Exception Handling

+-------------------------------------------------------------------------+
|                  THE CPYTHON IMPORT RESOLUTION PIPELINE                 |
+-------------------------------------------------------------------------+
|                                                                         |
|  User invokes: import foo.bar                                           |
|       |                                                                 |
|       v                                                                 |
|  [ Step 1: Cache Inspection ]                                           |
|       |--> Does 'foo.bar' exist in sys.modules?                         |
|       |       |-- YES --> Return sys.modules['foo.bar'] (O(1) Instant)   |
|       |-- NO                                                            |
|       v                                                                 |
|  [ Step 2: Meta Path Traversal ]                                        |
|       |--> Iterate through sys.meta_path finders:                       |
|       |    1. BuiltinImporter (for sys, math, builtins, etc.)          |
|       |    2. FrozenImporter (for frozen standard library modules)      |
|       |    3. PathFinder (for filesystem searches via sys.path)        |
|       |                                                                 |
|       |--> Finder calls find_spec("foo.bar", path, target)              |
|       |       |-- Found? --> Returns ModuleSpec object                  |
|       |-- Spec NOT found --> Raise ModuleNotFoundError                  |
|       v                                                                 |
|  [ Step 3: Module Object Creation ]                                     |
|       |--> ModuleSpec.loader.create_module(spec)                        |
|       |--> Instantiates types.ModuleType("foo.bar")                     |
|       |--> Sets __name__, __file__, __spec__, __loader__, __package__   |
|       v                                                                 |
|  [ Step 4: Cache Population (Early Insertion) ]                         |
|       |--> sys.modules['foo.bar'] = module_object                       |
|       |    (CRITICAL: Prevents infinite recursion in circular imports)  |
|       v                                                                 |
|  [ Step 5: Bytecode Compilation & Module Execution ]                    |
|       |--> Check for valid cached .pyc in __pycache__/                  |
|       |    - If absent/stale: compile .py source to PyCodeObject        |
|       |    - Write optimized .pyc with magic number + timestamp/hash    |
|       |--> ModuleSpec.loader.exec_module(module_object)                 |
|       |--> CPython executes top-level bytecode in module.__dict__       |
|       v                                                                 |
|  [ Step 6: Binding into Caller's Namespace ]                            |
|       |--> 'foo' is bound into caller's local/global scope              |
|                                                                         |
+-------------------------------------------------------------------------+

import sys

# Inspecting the global registry
print(type(sys.modules))  # <class 'dict'>
print('sys' in sys.modules)  # True
print('math' in sys.modules)  # False (if not yet imported)

import math
print('math' in sys.modules)  # True
print(sys.modules['math'] is math)  # True

+-------------------------------------------------------------------------+
|                  REGULAR PACKAGES VS. NAMESPACE PACKAGES                |
+---------------------+---------------------------+-----------------------+
| DIMENSION           | REGULAR PACKAGE           | NAMESPACE PACKAGE     |
|                     | (Traditional / Standard)  | (PEP 420 / Python 3.3+)|
+---------------------+---------------------------+-----------------------+
| Identifying Marker  | Must contain __init__.py  | NO __init__.py file   |
|                     | file in directory         | permitted             |
+---------------------+---------------------------+-----------------------+
| Filesystem Layout   | Confined to a single      | Can span multiple     |
|                     | physical directory        | disjoint directories, |
|                     |                           | zips, or site-packages|
+---------------------+---------------------------+-----------------------+
| Initialization Code | __init__.py executes upon | No initialization     |
|                     | first package import      | file; module namespace|
|                     |                           | is dynamically synthesized
+---------------------+---------------------------+-----------------------+
| ModuleSpec Type     | PathEntryFinder with      | _NamespacePath        |
|                     | Origin pointing to file   | (multi-path iterable) |
+---------------------+---------------------------+-----------------------+
| Primary Use Case    | Monolithic libraries,     | Monorepos, decoupled  |
|                     | standard domain packages  | company-wide plugins  |
+---------------------+---------------------------+-----------------------+

# mypackage/__init__.py
'''Package root initialization and public API re-exports.'''

from mypackage.client import APIClient
from mypackage.exceptions import APIError

__all__ = ['APIClient', 'APIError']

import sys
sys.path.extend(['/opt/cloud', '/opt/auth'])

import corp.storage.s3      # Resolves from /opt/cloud
import corp.identity.oauth   # Resolves from /opt/auth

print(corp.__path__)
# _NamespacePath(['/opt/cloud/corp', '/opt/auth/corp'])

from myapp.services.billing import process_invoice
import myapp.core.database as db

# Inside myapp/services/billing.py
from .calculator import compute_tax        # myapp.services.calculator
from ..core.database import get_connection # myapp.core.database
from ..models import Invoice               # myapp.models

python myapp/services/billing.py

python -m myapp.services.billing

+-------------------------------------------------------------------------+
|                  THE CIRCULAR IMPORT DEADLOCK TIMELINE                  |
+-------------------------------------------------------------------------+
|                                                                         |
|  [ File: service_a.py ]                  [ File: service_b.py ]         |
|                                                                         |
|  1. CPython starts executing             .                              |
|     service_a.py                         .                              |
|  2. sys.modules['service_a'] = ModuleA   .                              |
|  3. Line 1: from service_b import fn_b   .                              |
|        |                                 .                              |
|        +---- IMPORT SUSPENDS --------->  4. CPython starts executing    |
|              (service_a half-loaded)        service_b.py                |
|                                          5. sys.modules['service_b']    |
|                                             = ModuleB                   |
|                                          6. Line 1:                     |
|                                             from service_a import fn_a  |
|                                                |                        |
|                                                +-- Check sys.modules:   |
|                                                    'service_a' EXISTS!  |
|                                                    (ModuleA returned)   |
|                                          7. CPython inspects ModuleA    |
|                                             for symbol 'fn_a'           |
|                                          8. CRASH! AttributeError:      |
|                                             partially initialized       |
|                                             module 'service_a' has no   |
|                                             attribute 'fn_a'            |
|                                                                         |
+-------------------------------------------------------------------------+

+-------------------------------------------------------------------------+
|               STRATEGIES FOR RESOLVING CIRCULAR IMPORTS                 |
+---------------------------------+---------------------------------------+
| TECHNIQUE                       | ARCHITECTURAL TRADE-OFF               |
+---------------------------------+---------------------------------------+
| 1. Domain Decoupling            | Best Practice. Creates clean DAG.     |
|    (Extract Shared Interfaces)  | Requires thoughtful architecture.     |
+---------------------------------+---------------------------------------+
| 2. typing.TYPE_CHECKING Guard   | Solves 90% of type-hint cycles.       |
|                                 | Zero runtime performance cost.        |
+---------------------------------+---------------------------------------+
| 3. Import Module, Not Symbols   | Works if attribute is accessed lazily |
|    (import service_a)           | inside function bodies.               |
+---------------------------------+---------------------------------------+
| 4. Deferred / Local Import      | Functional workaround. Slightly hides |
|    (import inside function)     | module dependencies; minor overhead.  |
+---------------------------------+---------------------------------------+

# user_service.py
from typing import TYPE_CHECKING

if TYPE_CHECKING:
    from order_service import OrderService  # Never imported at runtime!

class UserService:
    def __init__(self, orders: 'OrderService') -> None:
        self.orders = orders

BEFORE (Cycle):
   [ AuthService ] <=======> [ UserModel ]

AFTER (Decoupled DAG):
   [ AuthService ]           [ UserModel ]
         \                         /
          v                       v
             [ Security / Crypto ]

+-------------------------------------------------------------------------+
|                  THE MODERN PYPROJECT.TOML ARCHITECTURE                 |
+-------------------------------------------------------------------------+
|                                                                         |
|  [build-system]                   <-- PEP 518 (Build Environment Setup) |
|  requires = ["hatchling>=1.18"]       Specifies what tools pip must     |
|  build-backend = "hatchling.build"    install in a temporary venv       |
|                                       to compile this package.          |
|                                                                         |
|  [project]                        <-- PEP 621 (Standardized Metadata)   |
|  name = "enterprise-telemetry"        Declarative, machine-readable,    |
|  version = "2.4.0"                    zero code execution required.     |
|  description = "Production metrics"                                     |
|  readme = "README.md"                                                   |
|  requires-python = ">=3.11"                                             |
|  dependencies = [                                                       |
|      "pydantic>=2.0",                                                   |
|      "httpx>=0.24.0",                                                   |
|  ]                                                                      |
|                                                                         |
|  [project.optional-dependencies]                                        |
|  test = ["pytest>=7.0", "coverage"]                                     |
|                                                                         |
|  [project.scripts]                                                      |
|  telemetry-cli = "enterprise_telemetry.cli:main"                        |
|                                                                         |
+-------------------------------------------------------------------------+

+-------------------------------------------------------------------------+
|                 SDIST (.tar.gz) VS. WHEEL (.whl - PEP 427)              |
+---------------------+---------------------------+-----------------------+
| DIMENSION           | SOURCE DISTRIBUTION       | BUILT WHEEL           |
|                     | (.tar.gz)                 | (.whl)                |
+---------------------+---------------------------+-----------------------+
| Format              | Compressed tarball of raw | Standard ZIP archive  |
|                     | source files + build spec | with pre-built files  |
+---------------------+---------------------------+-----------------------+
| Installation Speed  | Slow (Requires unpacking, | Ultra-Fast (Direct un-|
|                     | compilation, build venv)  | zip into site-packages|
+---------------------+---------------------------+-----------------------+
| C/C++ Extensions    | Must be compiled locally  | Pre-compiled binary;  |
|                     | on consumer's machine     | zero compiler needed  |
|                     | (requires GCC/Clang/MSVC) | on target machine     |
+---------------------+---------------------------+-----------------------+
| Platform Portability| Universal (pure source)   | Tagged by ABI & OS    |
|                     |                           | (e.g., manylinux, arm)|
+---------------------+---------------------------+-----------------------+

+-------------------------------------------------------------------------+
|                  CPYTHON EXCEPTION CLASS TAXONOMY                       |
+-------------------------------------------------------------------------+
|                                                                         |
|                              BaseException                              |
|                                    |                                    |
|      +-----------------------------+-----------------------------+      |
|      |                             |                             |      |
|  SystemExit                KeyboardInterrupt               Exception    |
|  (sys.exit())              (Ctrl+C SIGINT)                       |      |
|                                                                  |      |
|       +----------------------------+-----------------------------+      |
|       |                            |                             |      |
|  StandardError             RuntimeError                  LookupError    |
|       |                            |                             |      |
|  ValueError                 RecursionError           +-----------+----+ |
|  TypeError                                           |                | |
|  AttributeError                                  IndexError        KeyError
|  ImportError                                                            |
|       |                                                                 |
|  ModuleNotFoundError                                                    |
|                                                                         |
+-------------------------------------------------------------------------+

# CATASTROPHIC ANTI-PATTERN
try:
    process_transactions()
except:  # Or: except BaseException:
    logger.error("Something went wrong, continuing...")

try:
    process_transactions()
except Exception as err:
    # Safely catches all application-level errors while preserving
    # SystemExit, KeyboardInterrupt, and GeneratorExit
    logger.exception(f"Transaction failed: {err}")

try:
    int("invalid_number")
except ValueError as e:
    raise RuntimeError("Failed to parse configuration")

ValueError: invalid literal for int() with base 10: 'invalid_number'

During handling of the above exception, another exception occurred:

RuntimeError: Failed to parse configuration

try:
    raw = socket.recv(1024)
except TimeoutError as err:
    raise ServiceUnavailableError("Upstream billing gateway timed out") from err

TimeoutError: [Errno 110] Connection timed out

The above exception was the direct cause of the following exception:

ServiceUnavailableError: Upstream billing gateway timed out

try:
    data = internal_db.fetch(key)
except InternalDatabaseDriverError:
    raise KeyError(f"Item '{key}' not found") from None

import asyncio

async def worker(task_id: int):
    if task_id == 1:
        raise ValueError("Invalid payload in Worker 1")
    elif task_id == 2:
        raise ConnectionResetError("Connection dropped in Worker 2")
    return "Success"

async def main():
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(worker(1))
            tg.create_task(worker(2))
    except* ValueError as eg:
        for err in eg.exceptions:
            print(f"Handled validation error: {err}")
    except* ConnectionError as eg:
        for err in eg.exceptions:
            print(f"Handled network error: {err}")

# Both exceptions are caught, separated by type, and handled concurrently!

'''
Production-Grade Plugin Importer and Resilient Execution Engine.
Demonstrates custom import hooks, hierarchical exception handling,
and structured diagnostics without external dependencies.
'''

import sys
import types
import time
import logging
from typing import Dict, Any, List, Optional
from importlib.abc import MetaPathFinder, Loader
from importlib.machinery import ModuleSpec

# Configure structured logging
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger("Engine")


# =====================================================================
# 1. ENTERPRISE EXCEPTION HIERARCHY
# =====================================================================

class PlatformError(Exception):
    '''Root exception for all platform-level errors.'''
    def __init__(self, message: str, error_code: str, context: Optional[Dict[str, Any]] = None):
        super().__init__(message)
        self.message = message
        self.error_code = error_code
        self.context = context or {}
        self.timestamp = time.time()

    def to_dict(self) -> Dict[str, Any]:
        return {
            "error_type": self.__class__.__name__,
            "error_code": self.error_code,
            "message": self.message,
            "context": self.context,
            "timestamp": self.timestamp,
        }


class PluginImportError(PlatformError):
    '''Raised when a dynamic plugin fails resolution or syntax validation.'''
    def __init__(self, plugin_name: str, reason: str):
        super().__init__(
            message=f"Plugin '{plugin_name}' failed loading: {reason}",
            error_code="ERR_PLUGIN_IMPORT_FAILED",
            context={"plugin_name": plugin_name, "reason": reason}
        )


class PluginExecutionError(PlatformError):
    '''Raised when a plugin crashes during task execution.'''
    def __init__(self, plugin_name: str, task_name: str, reason: str):
        super().__init__(
            message=f"Plugin '{plugin_name}' failed executing task '{task_name}': {reason}",
            error_code="ERR_PLUGIN_EXECUTION_FAILED",
            context={"plugin_name": plugin_name, "task_name": task_name, "reason": reason}
        )


# =====================================================================
# 2. IN-MEMORY VIRTUAL PLUGIN IMPORT HOOK (PEP 302 / PEP 451)
# =====================================================================

class VirtualPluginLoader(Loader):
    '''Compiles and executes virtual in-memory Python source code.'''
    def __init__(self, code_source: str):
        self.code_source = code_source

    def create_module(self, spec: ModuleSpec) -> Optional[types.ModuleType]:
        # Return None to use default CPython module creation semantics
        return None

    def exec_module(self, module: types.ModuleType) -> None:
        try:
            # Compile source code into a code object
            code_obj = compile(self.code_source, module.__file__, "exec")
            # Execute bytecode inside the new module's __dict__ namespace
            exec(code_obj, module.__dict__)
        except Exception as err:
            raise PluginImportError(module.__name__, str(err)) from err


class VirtualPluginFinder(MetaPathFinder):
    '''
    Intercepts imports with prefix 'plugins.virtual.' and provides
    in-memory ModuleSpecs without reading from disk.
    '''
    def __init__(self):
        self.registry: Dict[str, str] = {}

    def register_plugin(self, full_name: str, python_code: str) -> None:
        self.registry[full_name] = python_code
        logger.info(f"Registered virtual plugin in memory: {full_name}")

    def find_spec(
        self,
        fullname: str,
        path: Optional[List[str]],
        target: Optional[types.ModuleType] = None
    ) -> Optional[ModuleSpec]:
        if fullname in self.registry:
            code = self.registry[fullname]
            loader = VirtualPluginLoader(code)
            # Synthesize ModuleSpec with origin set to virtual URI
            return ModuleSpec(
                name=fullname,
                loader=loader,
                origin=f"virtual://{fullname}"
            )
        return None  # Pass through to standard filesystem PathFinder


# =====================================================================
# 3. RESILIENT TASK EXECUTION ENGINE
# =====================================================================

class ResilientEngine:
    '''Orchestrates dynamic module execution with fault isolation.'''
    def __init__(self, finder: VirtualPluginFinder):
        self.finder = finder

    def run_task(self, module_name: str, function_name: str, payload: Dict[str, Any]) -> Any:
        logger.info(f"Initiating task execution: {module_name}.{function_name}()")
        
        # Step 1: Dynamic Import Resolution
        try:
            # Import dynamically via built-in import machinery
            mod = __import__(module_name, fromlist=[function_name])
        except Exception as import_err:
            raise PluginImportError(module_name, "Module resolution failed") from import_err

        # Step 2: Function Attribute Verification
        if not hasattr(mod, function_name):
            raise PluginExecutionError(
                module_name,
                function_name,
                f"Attribute '{function_name}' missing from module namespace"
            )

        task_func = getattr(mod, function_name)

        # Step 3: Sandboxed Execution with Telemetry
        try:
            start_time = time.perf_counter()
            result = task_func(payload)
            elapsed_ms = (time.perf_counter() - start_time) * 1000.0
            logger.info(f"Task {function_name} completed successfully in {elapsed_ms:.2f}ms")
            return result
        except PlatformError:
            # Re-raise known platform exceptions untouched
            raise
        except Exception as runtime_err:
            # Enrich diagnostic note (Python 3.11+)
            if hasattr(runtime_err, "add_note"):
                runtime_err.add_note(f"Engine Payload: {payload}")
                runtime_err.add_note(f"Execution Target: {module_name}.{function_name}")
            
            # Explicitly chain into domain error
            raise PluginExecutionError(
                module_name,
                function_name,
                str(runtime_err)
            ) from runtime_err


# =====================================================================
# 4. VERIFICATION HARNESS
# =====================================================================

def main():
    # Install our custom finder at the front of sys.meta_path
    plugin_finder = VirtualPluginFinder()
    sys.meta_path.insert(0, plugin_finder)

    # Register Plugin A: A healthy data transformation plugin
    plugin_finder.register_plugin(
        "plugins.virtual.transformer",
        '''
def process_data(payload):
    values = payload.get("values", [])
    total = sum(values)
    return {"status": "SUCCESS", "count": len(values), "sum": total}
'''
    )

    # Register Plugin B: A buggy plugin that causes division by zero
    plugin_finder.register_plugin(
        "plugins.virtual.analytics",
        '''
def compute_ratio(payload):
    x = payload["x"]
    y = payload["y"]
    return x / y
'''
    )

    engine = ResilientEngine(plugin_finder)

    # Execution 1: Test valid plugin
    try:
        res = engine.run_task(
            "plugins.virtual.transformer",
            "process_data",
            {"values": [10, 20, 30, 40]}
        )
        print("Result 1:", res)
    except PlatformError as e:
        logger.error(f"Execution failed: {e.to_dict()}")

    # Execution 2: Test fault-isolated plugin with chained exception
    try:
        res = engine.run_task(
            "plugins.virtual.analytics",
            "compute_ratio",
            {"x": 100, "y": 0}
        )
        print("Result 2:", res)
    except PlatformError as e:
        print("\n--- CAUGHT STRUCTURED DOMAIN EXCEPTION ---")
        print(f"Error Code: {e.error_code}")
        print(f"Message   : {e.message}")
        print(f"Underlying: {type(e.__cause__).__name__} -> {e.__cause__}")
        print(f"Full Telemetry Dict: {e.to_dict()}")

    # Execution 3: Test non-existent plugin
    try:
        engine.run_task("plugins.virtual.unknown", "run", {})
    except PlatformError as e:
        print("\n--- CAUGHT IMPORT FAILURE ---")
        print(f"Error Code: {e.error_code}")
        print(f"Message   : {e.message}")


if __name__ == "__main__":
    main()

+-------------------------------------------------------------------------+
|                  REQUESTS EXCEPTION WRAPPING PIPELINE                   |
+-------------------------------------------------------------------------+
|                                                                         |
|  User writes: requests.get("https://api.example.com")                   |
|       |                                                                 |
|       v                                                                 |
|  [ requests.sessions.Session.send() ]                                   |
|       |                                                                 |
|       v                                                                 |
|  [ requests.adapters.HTTPAdapter (Transport Layer) ]                    |
|       |                                                                 |
|       v                                                                 |
|  [ urllib3.connectionpool.urlopen() (Low-level Socket Engine) ]          |
|       |                                                                 |
|       |-- Low-level socket timeout occurs:                              |
|       |   urllib3 raises: urllib3.exceptions.TimeoutError               |
|       v                                                                 |
|  [ HTTPAdapter Exception Translation Boundary ]                         |
|       |                                                                 |
|       |-- HTTPAdapter catches urllib3.exceptions.TimeoutError           |
|       |-- Wraps into: requests.exceptions.ConnectTimeout                |
|       |   raise ConnectTimeout(e, request=request)                      |
|       v                                                                 |
|  [ Consumer Application ]                                               |
|       |                                                                 |
|       |-- Catches: requests.exceptions.RequestException                 |
|           (Consumer remains completely decoupled from urllib3)          |
|                                                                         |
+-------------------------------------------------------------------------+

class DatabaseSyncError(Exception):
    def __init__(self, table: str, record_id: int, sql_code: int):
        self.table = table
        self.record_id = record_id
        self.sql_code = sql_code
        super().__init__(f"Sync failed for {table}:{record_id} (Code {sql_code})")

class DatabaseSyncError(Exception):
    def __init__(self, message: str, table: str = "", record_id: int = 0, sql_code: int = 0):
        super().__init__(message)
        self.table = table
        self.record_id = record_id
        self.sql_code = sql_code

    def __reduce__(self):
        return (DatabaseSyncError, (str(self), self.table, self.record_id, self.sql_code))

# CATASTROPHIC ANTI-PATTERN
try:
    persist_user_audit_log(user_id, action)
except Exception:
    pass  # The developer assumes logging isn't critical

try:
    persist_user_audit_log(user_id, action)
except Exception:
    logger.warning("Audit log persistence failed", exc_info=True)

# FRAGILE ANTI-PATTERN
import sys
sys.path.append("/var/www/custom_libs")
sys.path.insert(0, "../relative_path")

# HAZARDOUS CONCURRENCY ANTI-PATTERN
async def worker_loop():
    while True:
        try:
            task = await queue.get()
            await process(task)
        except Exception:  # In Python < 3.8, asyncio.CancelledError inherited from Exception!
            logger.error("Error occurred")

from services.payment import process_payment

class Order:
    def __init__(self, order_id: str, amount: float):
        self.order_id = order_id
        self.amount = amount
        self.status = "PENDING"

    def checkout(self):
        self.status = process_payment(self)

from models.order import Order

def process_payment(order: Order) -> str:
    if order.amount <= 0:
        raise ValueError("Invalid amount")
    return "CHARGED"

from __future__ import annotations
from typing import TYPE_CHECKING

if TYPE_CHECKING:
    from models.order import Order  # Evaluated by static type checkers ONLY

def process_payment(order: Order) -> str:
    if order.amount <= 0:
        raise ValueError("Invalid amount")
    return "CHARGED"

from services.payment import process_payment

class Order:
    def __init__(self, order_id: str, amount: float):
        self.order_id = order_id
        self.amount = amount
        self.status = "PENDING"

    def checkout(self):
        self.status = process_payment(self)

from typing import Dict, Any, Tuple
import time

class BillingSystemError(Exception):
    '''Base exception for all billing domain operations.'''
    http_status: int = 500
    error_code: str = "ERR_INTERNAL_BILLING_ERROR"

    def __init__(
        self,
        client_message: str,
        internal_diagnostics: Dict[str, Any] = None,
        error_code: str = None,
        http_status: int = None
    ):
        super().__init__(client_message)
        self.client_message = client_message
        self.internal_diagnostics = internal_diagnostics or {}
        if error_code:
            self.error_code = error_code
        if http_status:
            self.http_status = http_status
        self.timestamp = time.time()

    def to_api_response(self) -> Tuple[Dict[str, Any], int]:
        '''Returns public-safe JSON payload and corresponding HTTP status code.'''
        payload = {
            "success": False,
            "error": {
                "code": self.error_code,
                "message": self.client_message,
                "timestamp": self.timestamp
            }
        }
        return payload, self.http_status


class InsufficientFundsError(BillingSystemError):
    http_status = 400
    error_code = "ERR_INSUFFICIENT_FUNDS"

    def __init__(self, requested: float, available: float, currency: str = "USD"):
        super().__init__(
            client_message=f"Transaction declined: insufficient {currency} balance.",
            internal_diagnostics={
                "requested_amount": requested,
                "available_balance": available,
                "deficit": requested - available,
                "currency": currency
            }
        )


class RateLimitExceededError(BillingSystemError):
    http_status = 429
    error_code = "ERR_RATE_LIMIT_EXCEEDED"

    def __init__(self, retry_after_seconds: int):
        self.retry_after = retry_after_seconds
        super().__init__(
            client_message=f"Rate limit exceeded. Please retry after {retry_after_seconds} seconds.",
            internal_diagnostics={"retry_after_seconds": retry_after_seconds}
        )

    def to_api_response(self) -> Tuple[Dict[str, Any], int]:
        payload, status = super().to_api_response()
        payload["error"]["retry_after"] = self.retry_after
        return payload, status


class PaymentGatewayTimeoutError(BillingSystemError):
    http_status = 504
    error_code = "ERR_GATEWAY_TIMEOUT"

    def __init__(self, gateway_name: str, elapsed_ms: float):
        super().__init__(
            client_message="Payment processing timed out. Please verify transaction status.",
            internal_diagnostics={"gateway": gateway_name, "latency_ms": elapsed_ms}
        )

err = InsufficientFundsError(requested=150.0, available=42.50)
payload, status = err.to_api_response()

# Status: 400
# Payload:
# {
#     "success": False,
#     "error": {
#         "code": "ERR_INSUFFICIENT_FUNDS",
#         "message": "Transaction declined: insufficient USD balance.",
#         "timestamp": 1698600000.0
#     }
# }
# Internal Diagnostics (preserved for logs, hidden from client):
# {'requested_amount': 150.0, 'available_balance': 42.5, 'deficit': 107.5, 'currency': 'USD'}

import asyncio

async def task_a():
    raise ValueError("Payload missing 'id'")

async def task_b():
    raise ConnectionResetError("Remote server closed peer socket")

async def task_c():
    raise KeyError("Missing authentication token")

async def run_batch():
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(task_a())
            tg.create_task(task_b())
            tg.create_task(task_c())
    except* ValueError as eg_val:
        print(f"Handler 1: Caught {len(eg_val.exceptions)} ValueError(s)")
    except* (ConnectionError, KeyError) as eg_other:
        print(f"Handler 2: Caught {len(eg_other.exceptions)} Network/Key Error(s)")

ExceptionGroup("unhandled errors in a TaskGroup", [
    ValueError("Payload missing 'id'"),
    ConnectionResetError("Remote server closed peer socket"),
    KeyError("Missing authentication token")
])

import sys
import types
from typing import Optional, List
from importlib.abc import MetaPathFinder, Loader
from importlib.machinery import ModuleSpec

def xor_decrypt(ciphertext: bytes, key: int) -> str:
    '''Decodes encrypted byte payload using XOR cipher.'''
    return bytes([b ^ key for b in ciphertext]).decode('utf-8')

def xor_encrypt(plaintext: str, key: int) -> bytes:
    '''Encodes plaintext string into encrypted bytes.'''
    return bytes([b ^ key for b in plaintext.encode('utf-8')])


class EncryptedModelLoader(Loader):
    '''Decodes and executes encrypted proprietary bytecode in memory.'''
    def __init__(self, encrypted_bytes: bytes, cipher_key: int):
        self.encrypted_bytes = encrypted_bytes
        self.cipher_key = cipher_key

    def create_module(self, spec: ModuleSpec) -> Optional[types.ModuleType]:
        # Return None to instruct CPython to instantiate default ModuleType
        return None

    def exec_module(self, module: types.ModuleType) -> None:
        # Decrypt source code in RAM (zero disk touch)
        decrypted_source = xor_decrypt(self.encrypted_bytes, self.cipher_key)
        
        # Compile source to code object
        virtual_filename = f"<vault://{module.__name__}>"
        code_obj = compile(decrypted_source, virtual_filename, "exec")
        
        # Execute code in module's dictionary
        exec(code_obj, module.__dict__)


class EncryptedModelFinder(MetaPathFinder):
    '''
    Intercepts imports under the 'secure_models.' namespace and loads
    them from the encrypted in-memory vault.
    '''
    def __init__(self, cipher_key: int):
        self.cipher_key = cipher_key
        # Vault stores model_name -> encrypted bytes
        self.vault: dict[str, bytes] = {}

    def register_model(self, model_name: str, source_code: str) -> None:
        fullname = f"secure_models.{model_name}"
        self.vault[fullname] = xor_encrypt(source_code, self.cipher_key)

    def find_spec(
        self,
        fullname: str,
        path: Optional[List[str]],
        target: Optional[types.ModuleType] = None
    ) -> Optional[ModuleSpec]:
        # Check if import matches our namespace
        if fullname.startswith("secure_models.") and fullname in self.vault:
            encrypted_data = self.vault[fullname]
            loader = EncryptedModelLoader(encrypted_data, self.cipher_key)
            return ModuleSpec(
                name=fullname,
                loader=loader,
                origin=f"vault://secure_models/{fullname.split('.')[-1]}"
            )
        return None  # Let standard finders handle other imports


# =====================================================================
# VERIFICATION HARNESS
# =====================================================================

if __name__ == "__main__":
    CIPHER_KEY = 0x5A
    finder = EncryptedModelFinder(cipher_key=CIPHER_KEY)
    
    # Store proprietary model in encrypted vault
    PROPRIETARY_CODE = '''
class FraudDetector:
    def predict(self, transaction_amount: float) -> str:
        if transaction_amount > 10000.0:
            return "FLAG_FOR_REVIEW"
        return "APPROVED"
'''
    finder.register_model("fraud_v1", PROPRIETARY_CODE)

    # Register finder on sys.meta_path
    sys.meta_path.insert(0, finder)

    # Consumer executes normal Python import!
    import secure_models.fraud_v1 as fraud_model

    print("Successfully imported encrypted module!")
    print("Module __name__  :", fraud_model.__name__)
    print("Module __file__  :", fraud_model.__file__)
    print("Module __loader__:", fraud_model.__loader__)

    # Instantiate the dynamically loaded class
    detector = fraud_model.FraudDetector()
    print("Inference 1 (500 USD)   :", detector.predict(500.0))
    print("Inference 2 (25000 USD) :", detector.predict(25000.0))