import contextlib
from json import JSONDecodeError
from .client.response import Response
__all__ = [
"AccountNotFoundException",
"AssetCodeInvalidError",
"AssetIssuerInvalidError",
"BadRequestError",
"BadResponseError",
"BadSignatureError",
"BaseHorizonError",
"BaseRequestError",
"ConnectionError",
"ContentSizeLimitExceededError",
"ContractCodeNotFoundError",
"ContractInstanceNotFoundError",
"ContractWasmRetrievalError",
"Ed25519PublicKeyInvalidError",
"Ed25519SecretSeedInvalidError",
"FeatureNotEnabledError",
"MemoInvalidException",
"MissingEd25519SecretSeedError",
"NoApproximationError",
"NotFoundError",
"NotPageableError",
"PrepareTransactionException",
"SACHasNoWasmError",
"SdkError",
"SignatureExistError",
"SorobanRpcErrorResponse",
"StreamClientError",
"UnknownRequestError",
]
from .soroban_rpc import SimulateTransactionResponse
[docs]
class SdkError(Exception):
"""Base exception for all stellar sdk related errors"""
[docs]
class BadSignatureError(SdkError, ValueError):
"""Raised when the signature was forged or otherwise corrupt."""
[docs]
class Ed25519PublicKeyInvalidError(SdkError, ValueError):
"""Ed25519 public key is incorrect."""
[docs]
class Ed25519SecretSeedInvalidError(SdkError, ValueError):
"""Ed25519 secret seed is incorrect."""
[docs]
class MissingEd25519SecretSeedError(SdkError, ValueError):
"""Missing Ed25519 secret seed in the keypair"""
[docs]
class MemoInvalidException(SdkError, ValueError):
"""Memo is incorrect."""
[docs]
class AssetCodeInvalidError(SdkError, ValueError):
"""Asset Code is incorrect."""
[docs]
class AssetIssuerInvalidError(SdkError, ValueError):
"""Asset issuer is incorrect."""
[docs]
class NoApproximationError(SdkError):
"""Approximation cannot be found"""
[docs]
class SignatureExistError(SdkError, ValueError):
"""A keypair can only sign a transaction once."""
[docs]
class BaseRequestError(SdkError):
"""Base class for requests errors."""
[docs]
class ConnectionError(BaseRequestError):
"""Base class for client connection errors."""
[docs]
class BaseHorizonError(BaseRequestError):
"""Base class for horizon request errors.
:param response: client response
"""
def __init__(self, response: Response) -> None:
super().__init__(response)
self.message: str = response.text
self.status: int = response.status_code
message = {}
with contextlib.suppress(JSONDecodeError):
message = response.json()
self.type: str | None = message.get("type")
self.title: str | None = message.get("title")
self.detail: str | None = message.get("detail")
self.extras: dict | None = message.get("extras")
self.result_xdr: str | None = message.get("extras", {}).get("result_xdr")
def __repr__(self):
return self.message
[docs]
class NotFoundError(BaseHorizonError):
"""This exception is thrown when the requested resource does not exist.
status_code == 400
"""
def __init__(self, response):
super().__init__(response)
[docs]
class BadRequestError(BaseHorizonError):
"""The request from the client has an error.
400 <= status_code < 500 and status_code != 404
"""
def __init__(self, response):
super().__init__(response)
[docs]
class BadResponseError(BaseHorizonError):
"""The response from the server has an error.
500 <= status_code < 600
"""
def __init__(self, response):
super().__init__(response)
class UnknownRequestError(BaseHorizonError):
"""Unknown request exception, please create an issue feedback for this issue."""
class NotPageableError(BaseRequestError):
"""There is no previous or next page"""
class StreamClientError(BaseRequestError):
"""Failed to fetch stream resource.
:param current_cursor: The cursor of the last message obtained can be used for reconnect.
:param message: error message
"""
def __init__(self, current_cursor: str, message: str) -> None:
super().__init__(message)
self.current_cursor = current_cursor
[docs]
class FeatureNotEnabledError(SdkError):
"""The feature is not enabled."""
class SorobanRpcErrorResponse(BaseRequestError):
"""The exception is thrown when the RPC server returns an error response."""
def __init__(self, code: int, message: str | None, data: str | None = None) -> None:
super().__init__(message)
self.code = code
self.data = data
self.message = message
class AccountNotFoundException(SdkError):
"""The exception is thrown when trying to load an account that doesn't exist on the Stellar network."""
def __init__(self, account_id: str) -> None:
super().__init__(f"Account not found, account_id: {account_id}")
self.account_id = account_id
class PrepareTransactionException(SdkError):
"""The exception is thrown when trying to prepare a transaction."""
def __init__(
self, message: str, simulate_transaction_response: SimulateTransactionResponse
) -> None:
super().__init__(message)
self.message = message
self.simulate_transaction_response = simulate_transaction_response
class ContractWasmRetrievalError(SdkError):
"""Raised when contract Wasm code cannot be retrieved from Stellar RPC."""
class ContractInstanceNotFoundError(ContractWasmRetrievalError):
"""Raised when a contract instance ledger entry cannot be found."""
class SACHasNoWasmError(ContractWasmRetrievalError):
"""Raised when a Stellar Asset Contract has no Wasm code."""
class ContractCodeNotFoundError(ContractWasmRetrievalError):
"""Raised when a contract code ledger entry cannot be found."""
class ContentSizeLimitExceededError(BaseRequestError):
"""The exception is thrown when the response content size exceeds the specified limit.
This is a security measure to prevent denial-of-service attacks via memory exhaustion.
:param limit: The maximum allowed content size in bytes.
:param content_size: The actual content size in bytes (may be approximate if streaming).
"""
def __init__(self, limit: int, content_size: int | None = None) -> None:
if content_size is not None:
message = f"Response content size ({content_size} bytes) exceeds the limit ({limit} bytes)"
else:
message = f"Response content size exceeds the limit ({limit} bytes)"
super().__init__(message)
self.limit = limit
self.content_size = content_size
def raise_request_exception(response: Response) -> None:
status_code = response.status_code
if 200 <= status_code < 300:
pass
elif status_code == 404:
raise NotFoundError(response)
elif 400 <= status_code < 500:
raise BadRequestError(response)
elif 500 <= status_code < 600:
raise BadResponseError(response)
else:
raise UnknownRequestError(response)