from json import JSONDecodeError
from typing import Optional
from .client.response import Response
__all__ = [
"SdkError",
"BadSignatureError",
"Ed25519PublicKeyInvalidError",
"Ed25519SecretSeedInvalidError",
"MissingEd25519SecretSeedError",
"MuxedEd25519AccountInvalidError",
"MemoInvalidException",
"AssetCodeInvalidError",
"AssetIssuerInvalidError",
"NoApproximationError",
"SignatureExistError",
"BaseRequestError",
"BaseHorizonError",
"ConnectionError",
"NotFoundError",
"BadRequestError",
"BadResponseError",
"UnknownRequestError",
"NotPageableError",
"StreamClientError",
"FeatureNotEnabledError",
"SorobanRpcErrorResponse",
"AccountNotFoundException",
"PrepareTransactionException",
]
from .soroban_rpc import SimulateTransactionResponse
[文档]
class SdkError(Exception):
"""Base exception for all stellar sdk related errors"""
[文档]
class BadSignatureError(SdkError, ValueError):
"""Raised when the signature was forged or otherwise corrupt."""
[文档]
class Ed25519PublicKeyInvalidError(SdkError, ValueError):
"""Ed25519 public key is incorrect."""
[文档]
class Ed25519SecretSeedInvalidError(SdkError, ValueError):
"""Ed25519 secret seed is incorrect."""
[文档]
class MissingEd25519SecretSeedError(SdkError, ValueError):
"""Missing Ed25519 secret seed in the keypair"""
class MuxedEd25519AccountInvalidError(SdkError, ValueError):
"""Muxed Ed25519 public key is incorrect."""
[文档]
class MemoInvalidException(SdkError, ValueError):
"""Memo is incorrect."""
[文档]
class AssetCodeInvalidError(SdkError, ValueError):
"""Asset Code is incorrect."""
[文档]
class AssetIssuerInvalidError(SdkError, ValueError):
"""Asset issuer is incorrect."""
[文档]
class NoApproximationError(SdkError):
"""Approximation cannot be found"""
[文档]
class SignatureExistError(SdkError, ValueError):
"""A keypair can only sign a transaction once."""
[文档]
class BaseRequestError(SdkError):
"""Base class for requests errors."""
[文档]
class ConnectionError(BaseRequestError):
"""Base class for client connection errors."""
[文档]
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 = {}
try:
message = response.json()
except JSONDecodeError:
pass # pragma: no cover
self.type: Optional[str] = message.get("type")
self.title: Optional[str] = message.get("title")
self.detail: Optional[str] = message.get("detail")
self.extras: Optional[dict] = message.get("extras")
self.result_xdr: Optional[str] = message.get("extras", {}).get("result_xdr")
def __str__(self):
return self.message
[文档]
class NotFoundError(BaseHorizonError):
"""This exception is thrown when the requested resource does not exist.
status_code == 400
"""
def __init__(self, response):
super().__init__(response)
[文档]
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)
[文档]
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
[文档]
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: Optional[str], data: Optional[str] = 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
def raise_request_exception(response: Response) -> None:
status_code = response.status_code
if status_code == 200:
pass # pragma: no cover
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)