Source code for ipfshttpclient.exceptions

"""
The class hierachy for exceptions is::

	Error
	 ├── VersionMismatch
	 ├── AddressError
	 ├── EncoderError
	 │    ├── EncoderMissingError
	 │    ├── EncodingError
	 │    └── DecodingError
	 └── CommunicationError
	      ├── ProtocolError
	      ├── StatusError
	      ├── ErrorResponse
	      │    └── PartialErrorResponse
	      ├── ConnectionError
	      └── TimeoutError

"""
import typing as ty

import multiaddr.exceptions  # type: ignore[import]


class Error(Exception):
	"""Base class for all exceptions in this module."""
	__slots__ = ()


class AddressError(Error, multiaddr.exceptions.Error):  # type: ignore[no-any-unimported, misc]
	"""Raised when the provided daemon location Multiaddr does not match any
	of the supported patterns."""
	__slots__ = ("addr",)
	#addr: ty.Union[str, bytes]
	
	def __init__(self, addr: ty.Union[str, bytes]) -> None:
		self.addr = addr  # type: ty.Union[str, bytes]
		Error.__init__(self, "Unsupported Multiaddr pattern: {0!r}".format(addr))


class VersionMismatch(Error):
	"""Raised when daemon version is not supported by this client version."""
	__slots__ = ("current", "minimum", "maximum")
	#current: ty.Sequence[int]
	#minimum: ty.Sequence[int]
	#maximum: ty.Sequence[int]
	
	def __init__(self, current: ty.Sequence[int], minimum: ty.Sequence[int],
	             maximum: ty.Sequence[int]) -> None:
		self.current = current  # type: ty.Sequence[int]
		self.minimum = minimum  # type: ty.Sequence[int]
		self.maximum = maximum  # type: ty.Sequence[int]
		
		msg = "Unsupported daemon version '{}' (not in range: {} ≤ … < {})".format(
			".".join(map(str, current)), ".".join(map(str, minimum)), ".".join(map(str, maximum))
		)
		super().__init__(msg)


###############
# encoding.py #
###############
class EncoderError(Error):
	"""Base class for all encoding and decoding related errors."""
	__slots__ = ("encoder_name",)
	#encoder_name: str
	
	def __init__(self, message: str, encoder_name: str) -> None:
		self.encoder_name = encoder_name  # type: str
		
		super().__init__(message)


class EncoderMissingError(EncoderError):
	"""Raised when a requested encoder class does not actually exist."""
	__slots__ = ()
	
	def __init__(self, encoder_name: str) -> None:
		super().__init__("Unknown encoder: '{}'".format(encoder_name), encoder_name)


class EncodingError(EncoderError):
	"""Raised when encoding a Python object into a byte string has failed
	due to some problem with the input data."""
	__slots__ = ("original",)
	#original: Exception
	
	def __init__(self, encoder_name: str, original: Exception) -> None:
		self.original = original  # type: Exception
		
		super().__init__("Object encoding error: {}".format(original), encoder_name)


class DecodingError(EncoderError):
	"""Raised when decoding a byte string to a Python object has failed due to
	some problem with the input data."""
	__slots__ = ("original",)
	#original: Exception
	
	def __init__(self, encoder_name: str, original: Exception) -> None:
		self.original = original  # type: Exception
		
		super().__init__("Object decoding error: {}".format(original), encoder_name)


###########
# http.py #
###########
class CommunicationError(Error):
	"""Base class for all network communication related errors."""
	__slots__ = ("original",)
	#original: ty.Optional[Exception]
	
	def __init__(self, original: ty.Optional[Exception],
	             _message: ty.Optional[str] = None) -> None:
		self.original = original  # type: ty.Optional[Exception]
		
		msg = ""  # type: str
		if _message:
			msg = _message
		else:
			msg = "{}: {}".format(type(original).__name__, str(original))
		super().__init__(msg)


class ProtocolError(CommunicationError):
	"""Raised when parsing the response from the daemon has failed.
	
	This can most likely occur if the service on the remote end isn't in fact
	an IPFS daemon."""
	__slots__ = ()


class StatusError(CommunicationError):
	"""Raised when the daemon responds with an error to our request."""
	__slots__ = ()


class ErrorResponse(StatusError):
	"""Raised when the daemon has responded with an error message because the
	requested operation could not be carried out."""
	__slots__ = ()
	
	def __init__(self, message: str, original: ty.Optional[Exception]) -> None:
		super().__init__(original, message)


class PartialErrorResponse(ErrorResponse):
	"""Raised when the daemon has responded with an error message after having
	already returned some data."""
	__slots__ = ()
	
	def __init__(self, message: str, original: ty.Optional[Exception] = None) -> None:
		super().__init__(message, original)


class ConnectionError(CommunicationError):
	"""Raised when connecting to the service has failed on the socket layer."""
	__slots__ = ()


class TimeoutError(CommunicationError):
	"""Raised when the daemon didn't respond in time."""
	__slots__ = ()