Last active
August 24, 2024 12:08
-
-
Save anatoly-kussul/f2d7444443399e51e2f83a76f112364d to your computer and use it in GitHub Desktop.
Python sync-async decorator factory
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
class SyncAsyncDecoratorFactory: | |
""" | |
Factory creates decorator which can wrap either a coroutine or function. | |
To return something from wrapper use self._return | |
If you need to modify args or kwargs, you can yield them from wrapper | |
""" | |
def __new__(cls, *args, **kwargs): | |
instance = super().__new__(cls) | |
# This is for using decorator without parameters | |
if len(args) == 1 and not kwargs and (inspect.iscoroutinefunction(args[0]) or inspect.isfunction(args[0])): | |
instance.__init__() | |
return instance(args[0]) | |
return instance | |
class ReturnValue(Exception): | |
def __init__(self, return_value): | |
self.return_value = return_value | |
@contextmanager | |
def wrapper(self, *args, **kwargs): | |
raise NotImplementedError | |
@classmethod | |
def _return(cls, value): | |
raise cls.ReturnValue(value) | |
def __call__(self, func): | |
@wraps(func) | |
def call_sync(*args, **kwargs): | |
try: | |
with self.wrapper(*args, **kwargs) as new_args: | |
if new_args: | |
args, kwargs = new_args | |
return self.func(*args, **kwargs) | |
except self.ReturnValue as r: | |
return r.return_value | |
@wraps(func) | |
async def call_async(*args, **kwargs): | |
try: | |
with self.wrapper(*args, **kwargs) as new_args: | |
if new_args: | |
args, kwargs = new_args | |
return await self.func(*args, **kwargs) | |
except self.ReturnValue as r: | |
return r.return_value | |
self.func = func | |
return call_async if inspect.iscoroutinefunction(func) else call_sync |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
This class is cool! Thanks!