Skip to content

Instantly share code, notes, and snippets.

@crizCraig
Created January 10, 2024 21:00
Show Gist options
  • Save crizCraig/ca4c532966ad2b2506b8935ed84e2c55 to your computer and use it in GitHub Desktop.
Save crizCraig/ca4c532966ad2b2506b8935ed84e2c55 to your computer and use it in GitHub Desktop.
def generic_unsync(func: Callable[..., Any]) -> Callable[..., Any]:
"""
A function that dynamically decides whether to run asynchronously
or synchronously based on whether it's in an async context,
but always returns the result synchronously so that
you can use it from an async or sync context.
"""
from unsync import unsync # type: ignore # pylint: disable=import-outside-toplevel
@unsync # type: ignore
async def async_wrapper(*args: Any, **kwargs: Any) -> Any:
loop = asyncio.get_running_loop()
result = await loop.run_in_executor(None, func, *args, **kwargs)
return result
def sync_wrapper(*args: Any, **kwargs: Any) -> Any:
if in_async_context():
return async_wrapper(*args, **kwargs).result()
else:
return func(*args, **kwargs)
return sync_wrapper
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment