Created
February 20, 2020 11:53
-
-
Save prostomarkeloff/8d750ff692d162f107dc0c1b3a42034c to your computer and use it in GitHub Desktop.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment