"""Domain-specific error hierarchy. Each domain defines its own error subclasses with HTTP status + code. The FastAPI exception handler (in core.errors) returns a consistent ErrorResponse envelope. Why domain errors over a generic AppError: - OpenAPI schema documents each error code (clients can switch on code). - Self-documenting: a wallet domain error says "WalletNotFound" not "AppError with code 404". - Each domain owns its error vocabulary. New errors don't require touching the global error catalog. - The legacy code has scattered HTTPException raises. Domain errors replace these with typed exceptions that the handler converts. Migration: existing code that raises AppError/NotFoundError/AuthError keeps working (those are base classes). New code should use domain-specific subclasses. """ from __future__ import annotations from typing import Any, Optional class AppError(Exception): """Base for all RMI errors. Each subclass declares its HTTP status.""" status_code: int = 500 code: str = "internal_error" def __init__( self, message: str = "", *, details: Optional[dict[str, Any]] = None, ) -> None: super().__init__(message) self.message = message or self.code self.details = details or {} def to_dict(self) -> dict[str, Any]: return { "code": self.code, "message": self.message, "details": self.details, } # ── Generic categories (subdomain-agnostic) ─────────────────────────── class NotFoundError(AppError): status_code = 404 code = "not_found" class AuthError(AppError): status_code = 401 code = "unauthorized" class ForbiddenError(AppError): status_code = 403 code = "forbidden" class RateLimitError(AppError): status_code = 429 code = "rate_limited" class ValidationError(AppError): status_code = 400 code = "validation_error" class ConflictError(AppError): status_code = 409 code = "conflict" class UpstreamError(AppError): """External API failed (chain RPC, databus provider, etc).""" status_code = 502 code = "upstream_error" # ── Domain-specific errors ──────────────────────────────────────────── class WalletError(AppError): """Base for all wallet-domain errors.""" code = "wallet_error" class WalletNotFoundError(WalletError, NotFoundError): code = "wallet_not_found" def __init__(self, address: str, chain: str = "unknown") -> None: super().__init__( f"Wallet {address[:12]}... not found", details={"address": address, "chain": chain}, ) class InsufficientFundsError(WalletError): status_code = 402 code = "insufficient_funds" def __init__(self, required: float, available: float, asset: str = "native") -> None: super().__init__( f"Insufficient {asset}: need {required}, have {available}", details={"required": required, "available": available, "asset": asset}, ) class TokenError(AppError): """Base for all token-domain errors.""" code = "token_error" class TokenNotScannedError(TokenError, NotFoundError): code = "token_not_scanned" def __init__(self, address: str, chain: str = "unknown") -> None: super().__init__( f"Token {address[:12]}... has not been scanned", details={"address": address, "chain": chain}, ) class HoneypotDetectedError(TokenError): status_code = 422 code = "honeypot_detected" def __init__(self, address: str, chain: str = "unknown") -> None: super().__init__( f"Token {address[:12]}... is a honeypot — cannot trade", details={"address": address, "chain": chain}, ) class ScanError(TokenError, UpstreamError): code = "scan_failed" def __init__(self, address: str, reason: str) -> None: super().__init__( f"Scan failed for {address[:12]}...: {reason}", details={"address": address, "reason": reason}, ) class AlertError(AppError): """Base for alert-domain errors.""" code = "alert_error" class AlertNotFoundError(AlertError, NotFoundError): code = "alert_not_found" def __init__(self, alert_id: str) -> None: super().__init__( f"Alert {alert_id} not found", details={"alert_id": alert_id}, ) class AlertQuotaExceededError(AlertError, RateLimitError): code = "alert_quota_exceeded" def __init__(self, limit: int, used: int) -> None: super().__init__( f"Alert quota exceeded: {used}/{limit}", details={"limit": limit, "used": used}, ) class PaymentError(AppError): """Base for x402 payment errors.""" code = "payment_error" class PaymentRequiredError(PaymentError): status_code = 402 code = "payment_required" def __init__(self, tool: str, price_usd: float, chain: str = "solana") -> None: super().__init__( f"Payment required for {tool}: ${price_usd}", details={"tool": tool, "price_usd": price_usd, "chain": chain}, ) class PaymentFailedError(PaymentError, UpstreamError): code = "payment_failed" def __init__(self, tx_hash: str | None, reason: str) -> None: super().__init__( f"Payment failed: {reason}", details={"tx_hash": tx_hash, "reason": reason}, ) class RAGError(AppError): """Base for RAG errors.""" code = "rag_error" class RAGSearchError(RAGError, UpstreamError): code = "rag_search_failed" def __init__(self, query: str, reason: str) -> None: super().__init__( f"RAG search failed: {reason}", details={"query": query[:100], "reason": reason}, ) # ── Helpers ────────────────────────────────────────────────────────── def domain_error_response(error: AppError) -> dict[str, Any]: """Convert an AppError to the standard error envelope.""" return { "code": error.code, "message": error.message, "details": error.details, "status": error.status_code, } # ── FastAPI exception handlers ────────────────────────────────────── def register_error_handlers(app: Any, debug: bool = False) -> None: """Register exception handlers on the FastAPI app. Handlers: - AppError subclasses → domain_error_response, status from class - StarletteHTTPException → standard {error, code, request_id} envelope - ValueError → 400 validation error - Exception → 500 with optional traceback (dev only) """ import traceback import uuid from fastapi import Request from fastapi.responses import JSONResponse from starlette.exceptions import HTTPException as StarletteHTTPException @app.exception_handler(AppError) async def app_error_handler(request: Request, exc: AppError): request_id = getattr(request.state, "request_id", str(uuid.uuid4())) body = domain_error_response(exc) body["request_id"] = request_id return JSONResponse(status_code=exc.status_code, content=body) @app.exception_handler(StarletteHTTPException) async def http_exception_handler(request: Request, exc: StarletteHTTPException): request_id = getattr(request.state, "request_id", str(uuid.uuid4())) return JSONResponse( status_code=exc.status_code, content={ "code": exc.status_code, "message": str(exc.detail), "details": {}, "request_id": request_id, }, ) @app.exception_handler(ValueError) async def value_error_handler(request: Request, exc: ValueError): request_id = getattr(request.state, "request_id", str(uuid.uuid4())) return JSONResponse( status_code=400, content={ "code": "validation_error", "message": str(exc), "details": {}, "request_id": request_id, }, ) @app.exception_handler(Exception) async def unhandled_exception_handler(request: Request, exc: Exception): request_id = getattr(request.state, "request_id", str(uuid.uuid4())) body = { "code": "internal_error", "message": "Internal server error" if not debug else str(exc), "details": {}, "request_id": request_id, } if debug: body["details"]["traceback"] = traceback.format_exc().split("\n") return JSONResponse(status_code=500, content=body)