mxaccessgw/clients/python/src/mxgateway/errors.py

"""Typed exception model for MXAccess Gateway Python clients."""

from __future__ import annotations

from typing import Any

import grpc

from .generated import mxaccess_gateway_pb2 as pb


class MxGatewayError(Exception):
    """Base class for client wrapper errors."""

    def __init__(
        self,
        message: str,
        *,
        protocol_status: pb.ProtocolStatus | None = None,
        raw_reply: Any | None = None,
    ) -> None:
        """Initialize with a message and the optional raw protocol context."""
        super().__init__(message)
        self.protocol_status = protocol_status
        self.raw_reply = raw_reply


class MxGatewayTransportError(MxGatewayError):
    """Transport-level gRPC failure."""


class MxGatewayAuthenticationError(MxGatewayTransportError):
    """Authentication failure reported by gRPC."""


class MxGatewayAuthorizationError(MxGatewayTransportError):
    """Authorization failure reported by gRPC."""


class MxGatewaySessionError(MxGatewayError):
    """Gateway session failure."""


class MxGatewayWorkerError(MxGatewayError):
    """Gateway worker process or protocol failure."""


class MxGatewayCommandError(MxGatewayError):
    """Command failure that preserves the raw protobuf reply."""


class MxAccessError(MxGatewayCommandError):
    """MXAccess HRESULT or status failure."""


def map_rpc_error(operation: str, error: grpc.RpcError) -> MxGatewayTransportError:
    """Map a generated gRPC exception to the client exception hierarchy."""

    code = error.code() if hasattr(error, "code") else None
    details = error.details() if hasattr(error, "details") else str(error)
    message = f"{operation} failed: {details}"

    if code == grpc.StatusCode.UNAUTHENTICATED:
        return MxGatewayAuthenticationError(message)
    if code == grpc.StatusCode.PERMISSION_DENIED:
        return MxGatewayAuthorizationError(message)

    return MxGatewayTransportError(message)


def ensure_protocol_success(
    operation: str,
    protocol_status: pb.ProtocolStatus | None,
    raw_reply: Any | None = None,
) -> Any | None:
    """Raise typed gateway errors for non-OK protocol statuses."""

    code = (
        protocol_status.code
        if protocol_status is not None
        else pb.PROTOCOL_STATUS_CODE_UNSPECIFIED
    )

    if code in (
        pb.PROTOCOL_STATUS_CODE_OK,
        pb.PROTOCOL_STATUS_CODE_MXACCESS_FAILURE,
    ):
        return raw_reply

    message_text = protocol_status.message if protocol_status else ""
    message = f"{operation} failed: {message_text or pb.ProtocolStatusCode.Name(code)}"

    if code in (
        pb.PROTOCOL_STATUS_CODE_SESSION_NOT_FOUND,
        pb.PROTOCOL_STATUS_CODE_SESSION_NOT_READY,
    ):
        raise MxGatewaySessionError(
            message,
            protocol_status=protocol_status,
            raw_reply=raw_reply,
        )

    if code in (
        pb.PROTOCOL_STATUS_CODE_WORKER_UNAVAILABLE,
        pb.PROTOCOL_STATUS_CODE_TIMEOUT,
        pb.PROTOCOL_STATUS_CODE_CANCELED,
        pb.PROTOCOL_STATUS_CODE_PROTOCOL_VIOLATION,
    ):
        raise MxGatewayWorkerError(
            message,
            protocol_status=protocol_status,
            raw_reply=raw_reply,
        )

    raise MxGatewayCommandError(
        message,
        protocol_status=protocol_status,
        raw_reply=raw_reply,
    )


def ensure_mxaccess_success(operation: str, reply: pb.MxCommandReply) -> pb.MxCommandReply:
    """Raise `MxAccessError` when MXAccess returned HRESULT or status failure."""

    status = reply.protocol_status
    if status.code == pb.PROTOCOL_STATUS_CODE_MXACCESS_FAILURE:
        raise MxAccessError(
            _mxaccess_message(operation, reply),
            protocol_status=status,
            raw_reply=reply,
        )

    if reply.HasField("hresult") and reply.hresult < 0:
        raise MxAccessError(
            _mxaccess_message(operation, reply),
            protocol_status=status,
            raw_reply=reply,
        )

    for mx_status in reply.statuses:
        if _is_mxaccess_status_failure(mx_status):
            raise MxAccessError(
                _mxaccess_message(operation, reply),
                protocol_status=status,
                raw_reply=reply,
            )

    return reply


def _is_mxaccess_status_failure(mx_status: pb.MxStatusProxy) -> bool:
    """Return ``True`` only for a populated MXAccess status reporting failure.

    MXAccess uses ``success == 0`` as the failure flag, but ``0`` is also the
    proto3 scalar default. The gateway emits placeholder ``MxStatusProxy``
    entries with ``success`` unset for null ``MXSTATUS_PROXY`` COM entries
    (see ``MxStatusProxyConverter.ConvertMany``); such an entry has
    ``category`` of ``UNSPECIFIED`` or ``UNKNOWN``. Treating it as a failure
    would raise ``MxAccessError`` for a reply that carries no real failure,
    so failure is keyed on ``success == 0`` together with a populated,
    non-OK status category.
    """

    if mx_status.success != 0:
        return False
    return mx_status.category not in (
        pb.MX_STATUS_CATEGORY_UNSPECIFIED,
        pb.MX_STATUS_CATEGORY_UNKNOWN,
        pb.MX_STATUS_CATEGORY_OK,
    )


def _mxaccess_message(operation: str, reply: pb.MxCommandReply) -> str:
    status_text = reply.protocol_status.message or "MXAccess command failed"
    hresult = reply.hresult if reply.HasField("hresult") else None
    return (
        f"{operation} failed: {status_text}; "
        f"session={reply.session_id}; correlation={reply.correlation_id}; "
        f"hresult={hresult}; statuses={len(reply.statuses)}"
    )