Skip to content

Instantly share code, notes, and snippets.

@RyanKung
Created July 17, 2017 02:23
Show Gist options
  • Select an option

  • Save RyanKung/8446f68eca82cc27ff9f41148e4bcb92 to your computer and use it in GitHub Desktop.

Select an option

Save RyanKung/8446f68eca82cc27ff9f41148e4bcb92 to your computer and use it in GitHub Desktop.
a decorator of `maybe coroutine`
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