Source code for stellar_sdk.exceptions

from json import JSONDecodeError
from typing import Optional

from .client.response import Response
from .type_checked import type_checked

__all__ = [

# The following is kept for compatibility
ValueError = ValueError
TypeError = TypeError
AttributeError = AttributeError

[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"""
class MuxedEd25519AccountInvalidError(SdkError, ValueError): """Muxed Ed25519 public key is incorrect."""
[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 = {} 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
[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."""
@type_checked 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)