# -*- coding: utf-8 -*- """ Utilities for reusable and conditional error processing. This utilities allow converting this: .. code-block:: python try: ... except ValueError as exception: result = fix_value_error(exception) except KeyError as exception: result = fix_key_error(exception) except requests.RequestsException as exception: if exception.response is not None: if (exception.response.status_code == 409) and (exception.response.json()['error']['code'] = 'some_exc'): result = fix_duplicate_environment(exception) elif (exception.response.status_code == 422): result = fix_unprocessable_entity(exception) else: result = fix_default(exception) else: result = fix_default(exception) except BaseException as exception: result = fix_default(exception) to this: .. code-block:: python handlers = HttpErrorHandlers() handlers.register_handler(BaseException, fix_default) handlers.register_handler(ValueError, fix_value_error) handlers.register_handler(KeyError, fix_key_error) handlers.register_http_handler(fix_duplicate_environment, 409, 'some_exc') handlers.register_http_handler(fix_unprocessable_entity, 422) try: ... except BaseException as exception: result = handlers.handle(exception) .. rubric:: Handlers Handlers are used to process exceptions. Each handler is associated with particular exception and exception context. If you have multiple handlers for inherited exceptions - the closest one to the exception will be selected. E.g. you have handlers for :code:`KeyError`, :code:`Exception` and :code:`BaseException`. For handling :code:`LookupError` will be used handler for :code:`Exception`. For :code:`KeyboardInterrupt` - handler for :code:`BaseException`. And for any "direct hit" - corresponding handler. Exception context - is additional sequence of details, which can be used to select a more specific handlers. E.g. for :exc:`~requests.RequestException` this context have next form: :code:`[response.status_code, response.json()['error']['code']]`. If the :code:`response.json()['error']['code']` is unavailable - this part of context is omitted. Contexts are matched in a "longest prefix" manner. E.g. you have handlers for the same error, but with contexts :code:`['a', 'b', 'c']`, :code:`['a', 'b', 'd']`, :code:`['a', 'b']`, :code:`['a', 'e']`. If you have exception with a context :code:`['a', 'b', 'c', 'x']` - handler for context :code:`['a', 'b', 'c']` will be used. For exception with a context :code:`['a', 'b']` - handler for context :code:`['a', 'b']` will be used. For exception with an empty context - none of the above will be used. This exception contexts are detected for each exception separately with exception parsers (see :meth:`ErrorHandlers.register_parser`). Functions to handle exceptions must have next format: .. code-block:: python def handler(exception: BaseException) -> typing.Any: ... Result value of the handler will be returned through the :meth:`ErrorHandlers.handle`. """ __all__ = ['ErrorHandlers', 'HttpErrorHandlers'] import typing import requests TConException = typing.TypeVar('TConException', bound=BaseException, contravariant=True) if typing.TYPE_CHECKING: import typing_extensions class Handler(typing_extensions.Protocol[TConException]): # pylint: disable=too-few-public-methods """Common protocol for error handling functions.""" def __call__(self, exception: TConException) -> typing.Any: """ Execute handler. :param exception: Exception to handle. :return: Result of the exception handling. """ class Parser(typing_extensions.Protocol[TConException]): # pylint: disable=too-few-public-methods """Common protocol for context parsers for exceptions.""" def __call__(self, exception: TConException) -> typing.Tuple[typing.Any, ...]: """ Prepare exception context. :param exception: Exception, for which context must be prepared. :return: Context for the exception. """ class ErrorHandler(typing.NamedTuple): """ Internal description of the error handler. .. py:attribute:: exception_type Type of the exception, that can be handled bu this handler. .. py:attribute:: context Context prefix required to execute this handler. .. py:attribute:: handler Function to handle an exception. """ exception_type: typing.Type[BaseException] context: typing.Tuple[typing.Any, ...] handler: 'Handler' class ErrorParser(typing.NamedTuple): """ Internal description of the error parser. .. py:property:: exception_type Type of the exception, that can be handled bu this parser. .. py:property:: parser Function to prepare context for exception. """ exception_type: typing.Type[BaseException] parser: 'Parser' class ErrorHandlers: """Collection of error handlers.""" __slots__ = ('__handlers', '__parsers') def __init__(self) -> None: """Initialize new :class:`~ErrorHandlers` instance.""" self.__handlers: 'typing_extensions.Final[typing.List[ErrorHandler]]' = [] self.__parsers: 'typing_extensions.Final[typing.List[ErrorParser]]' = [] def handle(self, exception: BaseException) -> typing.Any: """ Find handler for the `exception` and execute it. :param exception: Exception to handle. :return: Whatever handler returns. """ context: typing.Tuple[typing.Any, ...] = () parser: ErrorParser for parser in self.__parsers: if isinstance(exception, parser.exception_type): context = parser.parser(exception) break handler: ErrorHandler for handler in self.__handlers: if isinstance(exception, handler.exception_type) and (handler.context == context[:len(handler.context)]): return handler.handler(exception) return None def register_handler( self, exception_type: typing.Type[TConException], handler: 'Handler[TConException]', *context: typing.Any, ) -> None: """ Add new exception handler. :param exception_type: Exception, which may be handled. :param handler: Handler function. :param context: Required context prefix. """ index: int = 0 while index < len(self.__handlers): current: ErrorHandler = self.__handlers[index] if issubclass(exception_type, current.exception_type): if current.exception_type != exception_type: break if len(current.context) <= len(context): break index += 1 self.__handlers.insert(index, ErrorHandler(exception_type=exception_type, context=context, handler=handler)) def register_parser(self, exception_type: typing.Type[TConException], parser: 'Parser[TConException]') -> None: """ Add new exception parser. :param exception_type: Exceptions, which may be parsed. :param parser: Context parsing function. """ index: int = 0 while index < len(self.__parsers): if issubclass(exception_type, self.__parsers[index].exception_type): break index += 1 self.__parsers.insert(index, ErrorParser(exception_type=exception_type, parser=parser)) class HttpErrorHandlers(ErrorHandlers): """Collection of error handlers, which is prepared to parse HTTP requests from NucleusAPI.""" def __init__(self) -> None: """Initialize new :class:`~HttpErrorHandlers` instance.""" super().__init__() self.register_parser(requests.RequestException, self.parse_http) @staticmethod def parse_http(exception: requests.RequestException) -> typing.Tuple[typing.Any, ...]: """Retrieve context for :class:`~requests.RequestException`.""" result: typing.List[typing.Any] = [] if exception.response is not None: result.append(exception.response.status_code) try: result.append(exception.response.json()['error']['code']) except (ValueError, TypeError, KeyError): pass return tuple(result) def register_http_handler(self, handler: 'Handler[requests.RequestException]', *context: typing.Any) -> None: """ Add new exception handler for :class:`~requests.RequestException`. Interface is similar to :meth:`~ErrorHandlers.register_handler`, except missing `exception_type` argument. """ self.register_handler(requests.RequestException, handler, *context)