Source code for heros.exceptions

"""Exceptions raised by HEROS, in particular those transported from a remote HERO."""
import builtins
import traceback
from typing import Any

#: Marker key of a structured error payload, so that a receiver can tell a serialized
#: exception apart from a plain string (as sent by protocol version <= 1.0).
ERROR_MARKER = "__heros_error__"




class HEROSError(Exception):
    """Base class of all HEROS errors."""





class RemoteHEROError(HEROSError):
    """An exception that was raised inside a remote HERO while serving a request.

    Args:
        message: the ``str`` of the remote exception.
        remote_type: class name of the remote exception, if the remote sent one.
        remote_traceback: formatted traceback from the remote process.
        endpoint: the endpoint that was queried.
    """

    def __init__(
        self,
        message: str,
        remote_type: str | None = None,
        remote_module: str | None = None,
        remote_traceback: str | None = None,
        endpoint: str | None = None,
    ):
        super().__init__(message)
        self.message = message
        self.remote_type = remote_type
        self.remote_module = remote_module
        self.remote_traceback = remote_traceback
        self.endpoint = endpoint



    @classmethod
    def from_exception(cls,
                       exc: BaseException, endpoint: str | None = None):
        return RemoteHEROError(
                               message=str(exc),
                               remote_type=type(exc).__name__,
                               remote_module=type(exc).__module__,
                               remote_traceback="".join(traceback.format_tb(exc.__traceback__)),
                               endpoint=endpoint
                               )




    @classmethod
    def from_payload(cls, payload: Any, endpoint: str | None = None):
        """Rebuild a remote exception from a serialized payload.

        All payloads that do not have the correct format (dict with error marker) are
        wrapped in a bare :class:`RemoteHEROError` so that old HEROs keep working.
        """

        if isinstance(payload, dict) and payload.get(ERROR_MARKER):
            new_cls = _error_class(payload.get("type"))
            return new_cls(
                payload.get("message", ""),
                remote_type=payload.get("type"),
                remote_traceback=payload.get("traceback"),
                endpoint=endpoint
            )
        return RemoteHEROError(str(payload), endpoint=endpoint)




    def __str__(self) -> str:
        head = f"{self.remote_type}: {self.message}" if self.remote_type else self.message
        if self.endpoint:
            head = f"{head} (raised in HERO at '{self.endpoint}')"
        if self.remote_traceback:
            head = f"{head}\n\n--- remote traceback ---\n{self.remote_traceback}"
        return head




    def serialize(self) -> dict[str, Any]:
        """Turn an exception into a payload that :func:`exception_from_payload` can rebuild."""
        return {
            ERROR_MARKER: True,
            "type": self.remote_type,
            "module": self.remote_module,
            "message": self.message,
            "traceback": self.remote_traceback
        }






class RemoteTransportError(HEROSError):
    """The query to a remote HERO failed at transport level (interrupted stream, ...)."""



_error_classes: dict[str, type] = {}
_mapping_blacklist = (RemoteHEROError, KeyboardInterrupt, SystemExit)




def _error_class(remote_type: str | None) -> type:
    """Build (and cache) an exception class that is both a ``RemoteHEROError`` and, where
    the remote exception is a builtin, that builtin.

    This lets user code keep writing ``except ValueError:`` around a remote call while the
    HEROS-specific information stays available on the exception object.
    """
    if not remote_type:
        return RemoteHEROError
    if remote_type not in _error_classes:
        base = getattr(builtins, remote_type, None)
        if isinstance(base, type) and issubclass(base, BaseException) and not issubclass(base, _mapping_blacklist):
            try:
                _error_classes[remote_type] = type(f"Remote{remote_type}", (RemoteHEROError, base), {})
            except TypeError:
                # incompatible layout: fall back to the plain remote error
                _error_classes[remote_type] = RemoteHEROError
        else:
            _error_classes[remote_type] = RemoteHEROError
    return _error_classes[remote_type]