Last active
June 30, 2024 09:22
-
-
Save Integralist/fb1b5dbb6271632298f44d62a2221905 to your computer and use it in GitHub Desktop.
[Python Async Decorator] #python #asyncio #decorator
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 | |
import time | |
from functools import wraps | |
def dec(fn): | |
if asyncio.iscoroutinefunction(fn): | |
@wraps(fn) | |
async def wrapper(*args, **kwargs): | |
print(fn, args, kwargs) # <function foo at 0x10952d598> () {} | |
await asyncio.sleep(5) | |
print("done with wrapper, going to call fn") | |
return await fn() | |
return wrapper | |
else: | |
@wraps(fn) | |
def wrapper(*args, **kwargs): | |
print(fn, args, kwargs) # <function bar at 0x108fb5a60> () {} | |
time.sleep(5) | |
print("done with wrapper, going to call fn") | |
return fn() | |
return wrapper | |
@dec | |
async def foo(): | |
return await asyncio.sleep(5, result="async function done") | |
@dec | |
def bar(): | |
time.sleep(5) | |
return "sync function done" | |
loop = asyncio.get_event_loop() | |
result = loop.run_until_complete(foo()) | |
print(result) # async function done | |
print(bar()) # sync function done | |
loop.close() |
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 | |
from functools import wraps | |
def dec(fn): | |
@wraps(fn) | |
async def wrapper(*args, **kwargs): | |
print(fn, args, kwargs) # <function foo at 0x10952d598> () {} | |
await asyncio.sleep(5) | |
print("done with wrapper, going to call fn") | |
return await fn() | |
return wrapper | |
@dec | |
async def foo(): | |
return await asyncio.sleep(5, result="i'm done") | |
loop = asyncio.get_event_loop() | |
result = loop.run_until_complete(foo()) | |
print(result) # i'm done | |
loop.close() |
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 | |
from functools import wraps | |
async def foo(): | |
return await asyncio.sleep(5, result="i'm done") | |
loop = asyncio.get_event_loop() | |
result = loop.run_until_complete(foo()) | |
print(result) # i'm done | |
loop.close() |
probably your forget to pass args to wrapped fn
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
The proper wrapper args are (*args, **kwargs) and the wrapped function should be called as fn(*args, **kwargs).
Obviously all this print and sleep code wouldn't be there in the general case either. But I like the idea of one decorator for both async and sync code.