Skip to content

Instantly share code, notes, and snippets.

@ultrafunkamsterdam
Last active April 23, 2022 15:21
Show Gist options
  • Save ultrafunkamsterdam/be35a6b9c3e5e35ea99725744b24d142 to your computer and use it in GitHub Desktop.
Save ultrafunkamsterdam/be35a6b9c3e5e35ea99725744b24d142 to your computer and use it in GitHub Desktop.
HIBRYD - mix sync, async, functions and coroutines without crashing
'''
_ _ ___ ___ ___ __ __ ___
| || | |_ _| | _ ) | _ \ \ \ / / | \
| __ | | | | _ \ | / \ 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