Merge pull request #54 from daveades/feat/35-stellar-error

feat(errors): implement StellarError (#35)

Author: codebestia

src/shade/__init__.py

@@ -15,6 +15,8 @@
     RateLimitError,
     ShadeError,
     SignatureVerificationError,
+    StellarError,
+    wrap_stellar_errors,
 )
 from .models import (
     AssetBalance,
@@ -49,6 +51,7 @@
     "ShadeError",
     "SignatureVerificationError",
     "ShadeObject",
+    "StellarError",
     "SyncHTTPClient",
     "Transfer",
     "TransferStatus",
@@ -61,6 +64,7 @@
     "environment",
     "max_retries",
     "timeout",
+    "wrap_stellar_errors",
 ]
 
 class _ShadeModule(ModuleType):

src/shade/errors.py

@@ -4,7 +4,10 @@
 from __future__ import annotations
 
 import json
-from typing import Any, Optional
+from contextlib import contextmanager
+from typing import Any, Generator, List, Optional, Tuple
+
+from stellar_sdk.exceptions import BaseHorizonError, SdkError
 
 INVALID_REQUEST_STATUS_CODES = (400, 422)
 
@@ -201,6 +204,225 @@ class NetworkError(ShadeError):
     """Raised when the SDK cannot complete a network request."""
 
 
+OPERATION_SUCCESS_CODE = "op_success"
+
+# Human-readable descriptions for the Stellar result codes this SDK is most
+# likely to surface. Codes absent from these tables fall back to the raw code,
+# so an unrecognised code still reaches the caller intact.
+TRANSACTION_RESULT_CODE_DESCRIPTIONS: dict[str, str] = {
+    "tx_failed": "one or more operations in the transaction failed",
+    "tx_too_early": "the ledger closed before the transaction's minimum time bound",
+    "tx_too_late": "the ledger closed after the transaction's maximum time bound",
+    "tx_missing_operation": "the transaction contained no operations",
+    "tx_bad_seq": "the sequence number does not match the source account",
+    "tx_bad_auth": "too few valid signatures, or the wrong network was used",
+    "tx_bad_auth_extra": "the transaction carries signatures that were not needed",
+    "tx_insufficient_balance": "the fee would drop the source account below its minimum reserve",
+    "tx_insufficient_fee": "the fee offered is below the network minimum",
+    "tx_no_source_account": "the source account does not exist on the network",
+    "tx_internal_error": "Horizon reported an unknown internal error",
+    "tx_not_supported": "the network does not support this transaction",
+    "tx_fee_bump_inner_failed": "the inner transaction of the fee bump failed",
+}
+
+OPERATION_RESULT_CODE_DESCRIPTIONS: dict[str, str] = {
+    "op_malformed": "the operation is malformed",
+    "op_underfunded": "the source account does not hold enough of the asset",
+    "op_src_no_trust": "the source account has no trustline for the asset",
+    "op_src_not_authorized": "the source account is not authorized to send the asset",
+    "op_no_destination": "the destination account does not exist on the network",
+    "op_no_trust": "the destination account has no trustline for the asset",
+    "op_not_authorized": "the destination account is not authorized to hold the asset",
+    "op_line_full": "the transfer would exceed the destination account's trustline limit",
+    "op_no_issuer": "the issuer of the asset does not exist",
+    "op_low_reserve": "the resulting account would fall below its minimum reserve",
+    "op_bad_auth": "the operation was not signed by enough authorized signers",
+    "op_no_account": "the source account of the operation does not exist",
+    "op_not_supported": "the network does not support this operation",
+    "change_trust_no_issuer": "the issuer of the asset does not exist",
+    "change_trust_invalid_limit": "the requested trustline limit is invalid",
+    "change_trust_low_reserve": "the account cannot cover the reserve for a new trustline",
+    "change_trust_self_not_allowed": "an account cannot create a trustline to itself",
+}
+
+
+class StellarError(ShadeError):
+    """Raised when a Stellar/Horizon call fails.
+
+    Wraps the underlying ``stellar_sdk`` exception so callers keep access to the
+    raw error while still handling a single SDK exception type. Build one from a
+    caught ``stellar_sdk`` exception with :meth:`from_exception`, or let
+    :func:`wrap_stellar_errors` do it for a whole block.
+
+    Attributes:
+        stellar_result_code: The transaction-level Horizon result code (e.g.
+            ``"tx_failed"``, ``"tx_insufficient_fee"``). Falls back to the first
+            failing operation code when Horizon reported no transaction code, and
+            is ``None`` when the failure carried no result codes at all.
+        operation_result_codes: Per-operation result codes exactly as Horizon
+            ordered them, including any ``"op_success"`` entries. Empty when the
+            failure was not a rejected transaction.
+        original_error: The raw ``stellar_sdk`` exception, or ``None`` when the
+            error was constructed directly.
+    """
+
+    def __init__(
+        self,
+        message: str,
+        stellar_result_code: Optional[str] = None,
+        original_error: Optional[Exception] = None,
+        status_code: Optional[int] = None,
+        response_body: Optional[str] = None,
+        operation_result_codes: Optional[List[str]] = None,
+    ) -> None:
+        super().__init__(message, status_code, response_body)
+        self.stellar_result_code = stellar_result_code
+        self.original_error = original_error
+        self.operation_result_codes: List[str] = list(operation_result_codes or [])
+
+    @property
+    def failed_operation_code(self) -> Optional[str]:
+        """The first operation result code that is not ``"op_success"``.
+
+        Lets callers branch on the specific failure (``op_no_trust``,
+        ``op_underfunded``, …) without walking
+        :attr:`operation_result_codes` themselves.
+        """
+        for code in self.operation_result_codes:
+            if code != OPERATION_SUCCESS_CODE:
+                return code
+        return None
+
+    def __str__(self) -> str:
+        message = self.message
+        if self.stellar_result_code:
+            message = f"{message} (result code: {self.stellar_result_code})"
+        if self.status_code is None:
+            return message
+        return f"{message} (status code: {self.status_code})"
+
+    @classmethod
+    def from_exception(
+        cls,
+        exc: Exception,
+        message: Optional[str] = None,
+    ) -> "StellarError":
+        """Wrap a ``stellar_sdk`` exception, pulling out any Horizon result codes.
+
+        Args:
+            exc: The caught ``stellar_sdk`` exception.
+            message: Overrides the message derived from the result codes. Useful
+                for adding operation context the exception cannot know about
+                (e.g. "Failed to submit payout txn_123").
+        """
+        transaction_code, operation_codes = _stellar_result_codes(exc)
+        status_code, response_body = _horizon_context(exc)
+        return cls(
+            message or _stellar_failure_message(exc, transaction_code, operation_codes),
+            stellar_result_code=transaction_code
+            or next((c for c in operation_codes if c != OPERATION_SUCCESS_CODE), None),
+            original_error=exc,
+            status_code=status_code,
+            response_body=response_body,
+            operation_result_codes=operation_codes,
+        )
+
+
+@contextmanager
+def wrap_stellar_errors(message: Optional[str] = None) -> Generator[None, None, None]:
+    """Re-raise any ``stellar_sdk`` failure inside the block as :class:`StellarError`.
+
+    The Stellar integration layer wraps its Horizon and Soroban calls with this
+    so callers only ever have to catch :class:`~shade.errors.ShadeError`::
+
+        with wrap_stellar_errors("Failed to submit payment"):
+            server.submit_transaction(transaction)
+
+    Args:
+        message: Overrides the derived message on the raised ``StellarError``.
+            The result codes and original exception are attached either way.
+    """
+    try:
+        yield
+    except SdkError as exc:
+        raise StellarError.from_exception(exc, message=message) from exc
+
+
+def _stellar_result_codes(exc: Exception) -> Tuple[Optional[str], List[str]]:
+    """Return ``(transaction_code, operation_codes)`` from a Horizon error.
+
+    Horizon reports these under ``extras.result_codes``. Every level is
+    type-checked rather than assumed, so a malformed or partial error body
+    degrades to "no codes" instead of raising while building an exception.
+    """
+    extras = getattr(exc, "extras", None)
+    if not isinstance(extras, dict):
+        return None, []
+    result_codes = extras.get("result_codes")
+    if not isinstance(result_codes, dict):
+        return None, []
+
+    transaction = result_codes.get("transaction")
+    operations = result_codes.get("operations")
+    return (
+        transaction if isinstance(transaction, str) else None,
+        [code for code in operations if isinstance(code, str)]
+        if isinstance(operations, list)
+        else [],
+    )
+
+
+def _horizon_context(exc: Exception) -> Tuple[Optional[int], Optional[str]]:
+    """Return ``(status_code, response_body)`` for a Horizon error, else ``(None, None)``."""
+    if not isinstance(exc, BaseHorizonError):
+        return None, None
+    status = getattr(exc, "status", None)
+    body = getattr(exc, "message", None)
+    return (
+        status if isinstance(status, int) else None,
+        body if isinstance(body, str) else None,
+    )
+
+
+def _stellar_failure_message(
+    exc: Exception,
+    transaction_code: Optional[str],
+    operation_codes: List[str],
+) -> str:
+    """Build a human-readable message for a Stellar failure.
+
+    Prefers the most specific signal available: a failing operation code first
+    (that is what actually went wrong), then the transaction code, then whatever
+    Horizon or the exception itself described.
+    """
+    operation_code = next(
+        (code for code in operation_codes if code != OPERATION_SUCCESS_CODE), None
+    )
+    if operation_code is not None:
+        description = OPERATION_RESULT_CODE_DESCRIPTIONS.get(operation_code)
+        if description:
+            return f"Stellar transaction failed: {description} ({operation_code})"
+        return f"Stellar transaction failed: {operation_code}"
+
+    if transaction_code is not None:
+        description = TRANSACTION_RESULT_CODE_DESCRIPTIONS.get(transaction_code)
+        return f"Stellar transaction failed: {description or transaction_code}"
+
+    account_id = getattr(exc, "account_id", None)
+    if account_id:
+        return f"Stellar account {account_id} does not exist on the network"
+
+    title = getattr(exc, "title", None)
+    detail = getattr(exc, "detail", None)
+    if title and detail:
+        return f"Stellar request failed: {title} - {detail}"
+    if title or detail:
+        return f"Stellar request failed: {title or detail}"
+
+    text = str(exc).strip()
+    return f"Stellar request failed: {text or type(exc).__name__}"
+
+
 def raise_for_invalid_request(
     status_code: int,
     response_body: Optional[str] = None,