Skip to content

Instantly share code, notes, and snippets.

@jsbueno
Last active February 11, 2020 18:54
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save jsbueno/6bacdf3e817d4fc45463b9db6ce9ac7f to your computer and use it in GitHub Desktop.
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)
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
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