Last active
February 11, 2020 18:54
-
-
Save jsbueno/6bacdf3e817d4fc45463b9db6ce9ac7f to your computer and use it in GitHub Desktop.
A decorator to be applied on other decorators that make then automatically work for both sync and async functions (coroutine functions)
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
In [97]: @log_run | |
...: def ble(x): | |
...: time.sleep(0.5) | |
...: print(x) | |
...: | |
...: | |
In [98]: @log_run | |
...: async def bli(x): | |
...: await asyncio.sleep(1) | |
...: print(x) | |
...: | |
...: | |
<ipython-input-90-d2b3d3b48e7b>:12: DeprecationWarning: "@coroutine" decorator is deprecated since Python 3.8, use "async def" instead | |
wrapped = asyncio.coroutine(wrapped) | |
In [99]: loop.run_until_complete(bli(10)) | |
start | |
stop | |
10 | |
In [100]: ble(5) | |
start | |
5 | |
stop |
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
import asyncio, inspect | |
from functools import wraps | |
def hybrid_async(deco): | |
@wraps(deco) | |
def wrapper(func): | |
coro = inspect.iscoroutinefunction(func) | |
if coro: | |
async def inner(*args, **kw): | |
return await func(*args, **kw) | |
else: | |
inner = func | |
wrapped = wraps(func)(deco(inner)) | |
if coro: | |
wrapped = asyncio.coroutine(wrapped) | |
return wrapped | |
return wrapper | |
# Example wrapped decorator: | |
@hybrid_async | |
def log_run(func): | |
@wraps(func) | |
def wrapper(*args, **kw): | |
print("start", flush=True) | |
v = func(*args, **kw) | |
print("stop", flush=True) | |
return v | |
return wrapper | |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment