orchestrata_ultimate.gateway.exceptions

Reference › orchestrata_ultimate › gateway › exceptions

exceptions is the gateway's flat exception hierarchy, rooted at GatewayError. Its subclasses are AuthenticationError (failed credentials), ConnectionLifecycleError, RPCError, RingBufferError, PubSubError and InsecureJWTSecretError (raised by config when the JWT secret is missing or is the development default outside a dev or test context). RPCError carries an rpc_code attribute, which defaults to the JSON-RPC internal-error code, so a handler can raise it with a specific code and the dispatcher in rpc turns it into an error response. Almost every module and handler in the package imports this module. See gateway.

At a glance

Class diagram

Dependency diagram

API

Classes

GatewayError

Bases: Exception · Line 6

Base exception for all gateway errors.

AuthenticationError

Bases: GatewayError · Line 10

Token authentication failed.

ConnectionLifecycleError

Bases: GatewayError · Line 14

WebSocket connection lifecycle error.

RPCError

Bases: GatewayError · Line 18

RPC method invocation error.

RingBufferError

Bases: GatewayError · Line 41

Redis ring buffer operation error.

PubSubError

Bases: GatewayError · Line 45

Pub/sub operation error.

InsecureJWTSecretError

Bases: GatewayError · Line 49

Raised when JWT_SECRET is missing/default outside dev/test.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/gateway/exceptions.py

"""Custom exceptions for orchestrata-gateway."""

from orchestrata_ultimate.core.base.traces import traces


class GatewayError(Exception):
    """Base exception for all gateway errors."""


class AuthenticationError(GatewayError):
    """Token authentication failed."""


class ConnectionLifecycleError(GatewayError):
    """WebSocket connection lifecycle error."""


class RPCError(GatewayError):
    """RPC method invocation error.

    Attributes:
        rpc_code: JSON-RPC error code for the response.
    """

    @traces(spec="014", story="US-011", reqs=["REQ-014-180"])
    def __init__(
        self,
        message: str,
        rpc_code: int = -32603,
    ) -> None:
        """Create an RPC error with an optional JSON-RPC code.

        Args:
            message: Human-readable error description.
            rpc_code: JSON-RPC error code (default: internal error).
        """
        super().__init__(message)
        self.rpc_code = rpc_code


class RingBufferError(GatewayError):
    """Redis ring buffer operation error."""


class PubSubError(GatewayError):
    """Pub/sub operation error."""


class InsecureJWTSecretError(GatewayError):
    """Raised when ``JWT_SECRET`` is missing/default outside dev/test.

    SPEC-099 (P0): a deploy that forgets to set ``JWT_SECRET`` must fail
    closed instead of silently accepting the well-known ``"dev-secret"``
    default, which would let anyone forge valid tokens.
    """