"""Typed errors mapped from grpc.RpcError and Connect failures."""
import grpc
[docs]
class OloError(Exception):
[docs]
def __init__(
self,
message: str,
code: int | None = None,
details: str | None = None,
*,
reason: str | None = None,
):
"""Create an error with optional gRPC/Connect code, details, and reason."""
super().__init__(message)
self.code = code
self.details = details
self.reason = reason
[docs]
class OloAmbiguousNamespace(OloError): ...
[docs]
class OloAmbiguousPlanningGroup(OloError): ...
[docs]
class OloTimeoutError(OloError): ...
[docs]
class OloInvalidArgument(OloError): ...
[docs]
class OloNotFound(OloError): ...
[docs]
class OloUnavailable(OloError): ...
[docs]
class OloFailedPrecondition(OloError): ...
[docs]
class OloInternalError(OloError): ...
[docs]
class OloUnauthenticated(OloError): ...
[docs]
class OloPermissionDenied(OloError): ...
_STATUS_TO_ERROR: dict[grpc.StatusCode, type[OloError]] = {
grpc.StatusCode.DEADLINE_EXCEEDED: OloTimeoutError,
grpc.StatusCode.INVALID_ARGUMENT: OloInvalidArgument,
grpc.StatusCode.NOT_FOUND: OloNotFound,
grpc.StatusCode.UNAVAILABLE: OloUnavailable,
grpc.StatusCode.FAILED_PRECONDITION: OloFailedPrecondition,
grpc.StatusCode.INTERNAL: OloInternalError,
grpc.StatusCode.UNAUTHENTICATED: OloUnauthenticated,
grpc.StatusCode.PERMISSION_DENIED: OloPermissionDenied,
}
_CONNECT_CODE_TO_ERROR: dict[str, type[OloError]] = {
"canceled": OloError,
"unknown": OloError,
"invalid_argument": OloInvalidArgument,
"deadline_exceeded": OloTimeoutError,
"not_found": OloNotFound,
"already_exists": OloError,
"permission_denied": OloPermissionDenied,
"resource_exhausted": OloError,
"failed_precondition": OloFailedPrecondition,
"aborted": OloError,
"out_of_range": OloInvalidArgument,
"unimplemented": OloError,
"internal": OloInternalError,
"unavailable": OloUnavailable,
"data_loss": OloInternalError,
"unauthenticated": OloUnauthenticated,
}
[docs]
def wrap_rpc_error(exc: grpc.RpcError) -> OloError:
"""Map a grpc.RpcError to a typed OloError."""
status = exc.code()
code = status.value[0] if status is not None else None
cls = _STATUS_TO_ERROR.get(status, OloError) if status is not None else OloError
return cls(exc.details() or str(exc), code=code)
[docs]
def wrap_connect_error(
code: str,
message: str,
*,
reason: str | None = None,
) -> OloError:
"""Map a Connect error code string to a typed OloError."""
normalized = (code or "unknown").strip().lower()
cls = _CONNECT_CODE_TO_ERROR.get(normalized, OloError)
return cls(message or normalized, reason=reason)
__all__ = [
"OloError",
"OloAmbiguousNamespace",
"OloAmbiguousPlanningGroup",
"OloTimeoutError",
"OloInvalidArgument",
"OloNotFound",
"OloUnavailable",
"OloFailedPrecondition",
"OloInternalError",
"OloUnauthenticated",
"OloPermissionDenied",
"OloPlatformNotConfigured",
"wrap_rpc_error",
"wrap_connect_error",
]