client.py

from enum import auto, Enum
import typing
import random


class TimeoutError(Exception):
    def __str__(self):
        return "TimeoutError"


def is_error() -> bool:
    return random.choice([True, False])


def callback(**params) -> dict:
    if is_error():
        return {"response": "Hello world"}
    else:
        raise TimeoutError()


def _noop_error_handler(err, ctx):
    pass


class RequestState(Enum):
    NOT_SENT = auto()
    SENT = auto()


class ResultState(Enum):
    NOTHING = auto()
    SUCCESS = auto()
    HANDLED_EXCEPTION = auto()
    UNHANDLED_EXCEPTION = auto()


class RequestContext:
    def __init__(self, **params):
        self.state: RequestState = RequestState.NOT_SENT
        self._params = params
        self.result = ResultContext()

        self._exception_handlers: typing.Dict[
            typing.Type[Exception], typing.Callable[[Exception, ResultContext], None],
        ] = {TimeoutError: _noop_error_handler}

    def _handle_exception(self, exception: Exception) -> bool:
        if self._exception_handlers[type(exception)] is not None:
            self._exception_handlers[type(exception)](exception, self.result)
            return True
        return False

    def set_exception_handler(
        self,
        exception: typing.Type[Exception],
        handler: typing.Callable[[Exception, "ResultContext"], None],
    ) -> None:
        if not self._exception_handlers.get(exception):
            raise ValueError("Unallowed exception")
        self._exception_handlers[exception] = handler

    def send(self) -> None:
        try:
            result = callback(**self._params)
            self.result.state = ResultState.SUCCESS
            self.result.data = result
        except Exception as exc:
            self.result.exception = exc
            if self._handle_exception(exc):
                self.result.state = ResultState.HANDLED_EXCEPTION
            else:
                self.result.state = ResultState.UNHANDLED_EXCEPTION
        self.state = RequestState.SENT


class ResultContext:
    def __init__(self):
        self.state: ResultState = ResultState.NOTHING
        self._exception: typing.Optional[Exception] = None
        self._exception_data: typing.Optional[dict] = None
        self._data: typing.Optional[dict] = None

    @property
    def exception(self) -> typing.Optional[Exception]:
        return self._exception

    @exception.setter
    def exception(self, exc: Exception) -> None:
        self._exception = exc

    @property
    def exception_data(self) -> typing.Optional[dict]:
        return self._exception_data

    @exception_data.setter
    def exception_data(self, data: dict) -> None:
        self._exception_data = data

    @property
    def data(self) -> typing.Optional[dict]:
        return self._data

    @data.setter
    def data(self, data: dict) -> None:
        self._data = data


class Client:
    def request(self, **params) -> RequestContext:
        return RequestContext(**params)


def main():
    def handle_timeout(err: Exception, ctx: ResultContext) -> None:
        ctx.exception_data = {"handled": "yes"}

    client = Client()
    ctx = client.request(some="hello")
    ctx.set_exception_handler(TimeoutError, handle_timeout)
    ctx.send()

    result = ctx.result
    print(result.data, result.exception_data, result.exception)

main()