Last active
April 23, 2022 15:21
-
-
Save ultrafunkamsterdam/be35a6b9c3e5e35ea99725744b24d142 to your computer and use it in GitHub Desktop.
HIBRYD - mix sync, async, functions and coroutines without crashing
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
''' | |
_ _ ___ ___ ___ __ __ ___ | |
| || | |_ _| | _ ) | _ \ \ \ / / | \ | |
| __ | | | | _ \ | / \ V / | |) | | |
|_||_| |___| |___/ |_|_\ _|_|_ |___/ | |
_|"""""|_|"""""|_|"""""|_|"""""|_| """ |_|"""""| | |
`-0-0-'"`-0-0-'"`-0-0-'"`-0-0-'"`-0-0-'"`-0-0-' | |
yes, that is intentionally wrong | |
''' | |
from asyncio import ( | |
iscoroutine as _iscoro, | |
iscoroutinefunction as _iscorofn, | |
get_running_loop as _getloop, | |
run as _run | |
) | |
from functools import ( | |
partial as _partial, | |
update_wrapper as _update_wrapper | |
) | |
from logging import getLogger as _getLogger | |
class hibryd: | |
""" | |
decorator for functions and coroutine functions | |
this makes your function or coroutine callable using "await" or without | |
""" | |
_log = _getLogger('HIBRYD') | |
_log.setLevel(25) | |
def __new__(cls, *args, **kwargs): | |
return super().__new__(cls) | |
def __init__(self, fn): | |
self._fn_is_coro = _iscoro(fn) ^ _iscorofn(fn) | |
self._fn = fn | |
_update_wrapper(self, fn) | |
def __call__(self, *args, **kwargs): | |
try: | |
loop = _getloop() | |
if self._fn_is_coro: | |
return loop.create_task(self._fn(*args, **kwargs)) | |
self._log.warning( | |
"called regular function {0.__module__}.{0.__name__} in async (await) context. " | |
"this is not not recommended ! " | |
.format(self._fn)) | |
return loop.run_in_executor(None, _partial(self._fn, *args, **kwargs)) | |
except RuntimeError: | |
if self._fn_is_coro: | |
self._log.warning( | |
"called awaitable coroutine {0.__module__}.{0.__name__} in sync (no await) context " | |
"this is not not recommended ! " | |
.format(self._fn)) | |
return _run(self._fn(*args, **kwargs)) | |
return self._fn(*args, **kwargs) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment