Source code for loom.core.sql.abc.errors

"""SQL subsystem errors built on the existing domain error hierarchy.

Inheriting from the existing domain types keeps the HTTP mapping unchanged:
:class:`~loom.core.errors.NotFound` maps to 404,
:class:`~loom.core.errors.Forbidden` to 403 and
:class:`~loom.core.errors.RuleViolation` to 422 through the current
``HttpErrorMapper`` without touching the public error API.
"""

from __future__ import annotations

from loom.core.errors import Forbidden, NotFound, RuleViolation


[docs] class UnknownConnectionError(NotFound): """Raised when a SQL connection name is not configured. Args: connection: Name of the missing connection. """ def __init__(self, connection: str) -> None: super().__init__("SqlConnection", id=connection)
[docs] class RoleNotAllowedError(Forbidden): """Raised when the caller role is not in the connection allowlist. Without a configured allowlist every caller-provided role is rejected (fail-closed policy). Args: role: Role requested by the caller. connection: Name of the connection that rejected the role. """ def __init__(self, role: str, *, connection: str) -> None: super().__init__(f"Role {role!r} is not allowed for SQL connection {connection!r}")
[docs] class RoleRequiredError(Forbidden): """Raised when no effective role can be resolved for a query. A query never runs with the full default roles of the connection user: without a caller role and without ``default_role`` execution is refused. Args: connection: Name of the connection lacking an effective role. """ def __init__(self, connection: str) -> None: super().__init__( f"SQL connection {connection!r} requires a role: the request carries " "no role and the connection defines no 'default_role'" )
[docs] class RolesNotBoundError(Forbidden): """Raised when the verified identity yields no allowed role. Covers every fail-closed outcome of identity binding — no verified claims, a missing, empty or wrongly typed roles claim, or an empty intersection with the connection allowlist. The message is deliberately uniform so the response never reveals which part of the token failed; the precise reason is logged server-side. Args: connection: Name of the connection that refused the request. """ def __init__(self, connection: str) -> None: super().__init__( f"SQL connection {connection!r} derives its roles from the verified " "identity, which carries no allowed role for this connection" )
[docs] class SqlExecutionError(RuleViolation): """Raised when the backend rejects a SQL statement. Carries a sanitized message (backend error code plus first line, without host, DSN or stack trace). Args: message: Sanitized backend rejection message. """ def __init__(self, message: str) -> None: super().__init__("sql", message)