Created
July 17, 2017 02:23
-
-
Save RyanKung/8446f68eca82cc27ff9f41148e4bcb92 to your computer and use it in GitHub Desktop.
a decorator of `maybe coroutine`
This file contains hidden or 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 typing import Union, Callable, Generic, GenericMeta, TypeVar | |
| from types import CoroutineType | |
| from asyncio import iscoroutinefunction | |
| from functools import wraps | |
| from asyncio import coroutine | |
| from typing import Callable, Any | |
| MaybeCorouteine = Union[CoroutineType, Callable] | |
| Handler = Callable[..., WsgiResponse] | |
| MaybeType = Generic[TypeVar("Just"), TypeVar("Error")] | |
| Maybe: GenericMeta = type("Maybe", (MaybeType, ), {}) | |
| def maybe_coroutine(fn: Callable) -> Callable[..., MaybeCorouteine]: | |
| if iscoroutinefunction(fn): | |
| return coroutine | |
| else: | |
| return lambda x: x | |
| def maybe_async_cps(fn: Callable) -> MaybeCorouteine: | |
| def parted(context: Callable[[Maybe], Response]): | |
| @wraps(fn) | |
| @maybe_coroutine(fn) | |
| def _(*args, **kwargs): | |
| try: | |
| if iscoroutinefunction(fn): | |
| result = yield from fn(*args, **kwargs) | |
| else: | |
| result = fn(*args, **kwargs) | |
| except Exception as e: | |
| result = e | |
| return context(result) | |
| return _ | |
| return parted | |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment