"""Handle errors and raise exceptions.""" import abc import typing from config import app_logger as logger class BaseOwsServiceException(abc.ABC, Exception): """Custom exception when making ows service requests.""" def __init__(self, message: str, **kwargs: typing.Any): error_message = f'{self.SERVICE} failure: {message} {kwargs}' logger.error(error_message) super().__init__(error_message) @property @abc.abstractmethod def SERVICE(self) -> str: raise NotImplementedError() class OwsPaymentException(BaseOwsServiceException): """ows-payment exception.""" SERVICE = 'ows-payment' class OwsAccountException(BaseOwsServiceException): """ows-abacus-account exception.""" SERVICE = 'ows-abacus-account' class OwsStateException(BaseOwsServiceException): """ows-abacus-state exception.""" SERVICE = 'ows-abacus-state' class OwsEventException(BaseOwsServiceException): """ows-abacus-event exception.""" SERVICE = 'ows-abacus-event' class OwsRoyaltiesException(BaseOwsServiceException): """ows-royalties exception.""" SERVICE = 'ows-royalties' class DataConsistencyException(Exception): """Custom exception for data consistency errors.""" def __init__(self, message: str, **kwargs: typing.Any): error_message = f'Data consistency failure: {message} {kwargs}' logger.error(error_message) super().__init__(error_message) class PaymentsCreationException(Exception): """Custom exception for data payments creation errors.""" def __init__(self, message: str, **kwargs: typing.Any): error_message = f'Payments creation failure: {message} {kwargs}' logger.error(error_message) super().__init__(error_message) class FlowthroughUpdateException(Exception): """Custom exception for flowthrough update errors.""" def __init__(self, message: str, **kwargs: typing.Any): error_message = f'Flowthrough update failure: {message} {kwargs}' logger.error(error_message) super().__init__(error_message)