Skip to content

async_to_sync

from pulsefire.functools import async_to_sync

Convert a coroutine function to run synchronously. Use as decorator @async_to_sync().

Example: 

@async_to_sync()
async def sample_func(number: int):
    ...

sample_func(0)

Parameters:

Name Type Description Default
runner Callable[[Awaitable[Any]], Any]

A callable that runs the awaitable synchronously.

 run

Raises:

Type Description
TypeError

When func is not a coroutine function.
Source code in pulsefire/functools.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
def async_to_sync(runner: Callable[[Awaitable[Any]], Any] = asyncio.run):
    """Convert a coroutine function to run synchronously. Use as decorator `@async_to_sync()`.

    Example:
    ```python
    @async_to_sync()
    async def sample_func(number: int):
        ...

    sample_func(0)
    ```

    Parameters:
        runner: A callable that runs the awaitable synchronously.

    Raises:
        TypeError: When `func` is not a coroutine function.
    """

    def decorator[**P, R](func: Callable[P, Awaitable[R]]) -> Callable[P, R]:
        if not inspect.iscoroutinefunction(func):
            raise TypeError(f"{func} is not a coroutine function")

        @functools.wraps(func)
        def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
            return runner(func(*args, **kwargs))

        return wrapper

    return decorator