awaitable

privex.helpers.asyncx.awaitable(func: Callable) → Callable

Decorator which helps with creation of async wrapper functions.

Usage

Define your async function as normal, then create a standard python function using this decorator - the function should just call your async function and return it.

>>> async def some_func_async(a: str, b: str):
...     c = a + b
...     return c
...
>>> @awaitable
>>> def some_func(a, b) -> Union[str, Coroutine[Any, Any, str]]:
...     return some_func_async(a, b)
...

Now, inside of async functions, we just await the wrapper function as if it were the original async function.

>>> async def my_async_func():
...     res = await some_func("hello", "world")
...

While inside of synchronous functions, we call the wrapper function as if it were a normal synchronous function. The decorator will create an asyncio event loop, run the function, then return the result - transparent to the calling function.

>>> def my_sync_func():
...     res = some_func("hello world")
...

Blacklists

If you mix a lot of synchronous and asynchronous code, sniffio may return coroutines to synchronous functions that were called from asynchronous functions, which can of course cause problems.

To avoid this issue, you can blacklist function names, module names (and their sub-modules), and/or fully qualified module paths to functions/methods.

Three blacklists are available in this module, which allow you to specify caller functions/methods, modules, or fully qualified module paths to functions/methods for which awaitable() wrapped functions/methods should always execute in an event loop and return synchronously.

Example:

>>> from privex.helpers import asyncx
>>> # All code within the module 'some.module' and it's sub-modules will always have awaitable's run their wrapped
>>> # functions synchronously.
>>> asyncx.AWAITABLE_BLACKLIST_MODS += ['some.module']
>>> # Whenever a function with the name 'example_func' (in any module) calls an awaitable, it will always run synchronously
>>> asyncx.AWAITABLE_BLACKLIST_FUNCS += ['example_func']
>>> # Whenever the specific class method 'other.module.SomeClass.some_sync' calls an awaitable, it will always run synchronously.
>>> asyncx.AWAITABLE_BLACKLIST += ['other.module.SomeClass.some_sync']

Original source: https://github.com/encode/httpx/issues/572#issuecomment-562179966