"""Custom Exception classes.""" from http import HTTPStatus from werkzeug.exceptions import HTTPException class FormattedException(Exception): """Exception handles formatted error.""" status_code = 500 # Internal server error def __init__(self, message: str, status_code: int | None = None): """Initialize the exception.""" message = f'{message[:100]}...' if len(message) > 100 else message super().__init__(message) self.message = message if status_code: self.status_code = status_code def to_dict(self): """Return dict representation of the exception.""" return {'error': self.message} class FormattedProviderException(FormattedException): """Exception handles formatted provider error.""" provider = None def __init__(self, message: str, status_code: int | None = None): """Initialize the exception.""" super().__init__(message, status_code) if self.provider is None: raise NotImplementedError('Provider is not set') def to_dict(self): """Return dict representation of the exception.""" return {'error': self.message, 'provider': self.provider} class PayoneerException(FormattedProviderException): """Generic Payoneer exception with helpers""" ERROR_FIELDS = ('error', 'error_description', 'error_details') provider = 'payoneer' @classmethod def raise_if_error_response( cls, response_data: dict, status_code: int | None = None ): """Helper method to validate response and raise if error.""" if response_data.get('error'): error_message = { field: response_data[field] for field in cls.ERROR_FIELDS if field in response_data } raise cls(error_message, status_code) class AbacusAccountDetailsForReceiptFetchException(Exception): """Exception handles getting abacus account details for tax receipt.""" pass class BrandFetchException(Exception): """Exception handles getting vendor brand error.""" pass class CorpEntityFetchException(FormattedException): """Exception handles getting account corp entity error.""" status_code = 400 class AbacusUpdateStateException(Exception): """Exception handles abacus update state error.""" pass class PayoneerPayeeDetailsException(Exception): """Exception handles Payoneer payee details request error.""" pass class PayoneerMassPayoutsException(PayoneerException): """Exception handles Payoneer mass payouts request error.""" status_code = 400 class HttpMassPayoutsException(FormattedProviderException): """Exception handles http mass payouts request error.""" provider = 'http' class NoPayoneerCredentialsException(PayoneerException): """Exception on empty credentials.""" status_code = 500 def __init__(self, message='Payoneer credentials not found'): super().__init__(message) class RegisterPayeeException(PayoneerException): """Exception handles new payee request error.""" status_code = 400 class ReleasePayeeException(PayoneerException): """Exception handles release payee request error.""" status_code = 400 class UpdatePayeeException(PayoneerException): """Exception handles update payee request error.""" status_code = 400 class CreatePayoneerRegistrationLinkException(PayoneerException): """Exception handles create registration link error.""" status_code = 500 @classmethod def raise_if_error_response( cls, response_data: dict, status_code: int | None = None ): """Helper method to validate response and raise if error.""" status_code = response_data.get('error_details', {}).get('code') or status_code super().raise_if_error_response(response_data, status_code) class GetPayeeDetailsException(PayoneerException): """Exception handles get payee details request error.""" class GetRegisterPayeeFormatException(PayoneerException): """Exception handles get register payee format error.""" status_code = 500 def __init__(self, message='Error getting payee registration format'): super().__init__(message) class KafkaSendEventException(Exception): """Exception handles send event to kafka error.""" pass class OwsRequestHeadersException(Exception): """Exception handles wrong ows request headers.""" pass class ViesApiException(Exception): """View api service Exception.""" pass class HmrcApiException(Exception): """Hmrc api service Exception.""" pass class PayeeCreateException(Exception): """Create payee exception""" def __init__(self, message): super().__init__(f'Payee create error: {message}') class AbacusStateException(Exception): """ABACUS state exception""" def __init__(self, message): super().__init__(f'ABACUS state error: {message}') class NoHmrcCredentialsException(FormattedProviderException): """Exception on empty credentials.""" provider = 'HMRC' status_code = 500 def __init__(self, message='HMRC credentials not found'): super().__init__(message) class RegisterWhitelabelProfileException(HTTPException): """Register whitelabel profile exception""" code = HTTPStatus.UNPROCESSABLE_ENTITY class GetTypedPayeeException(Exception): """Get typed payee exception""" def __init__(self, message): super().__init__(f'Getting payee subtype failed: {message}') class AccountPayeeNotFoundException(FormattedException): """Exception handles account payee not found error.""" def __init__(self, message: str): super().__init__(message, status_code=400) class TaxWithholdingOverrideException(FormattedException): """Exception handles tax withholding override error.""" def __init__(self, message: str): super().__init__(f'Tax withholding override error: {message}')