API Reference

universalasync

Modules:

• utils –
• wrapper –

Functions:

• async_to_sync –

  Convert all public async methods/properties of an object to universal methods.

• async_to_sync_wraps –

  Wrap an async method/property to universal method.

• get_event_loop –

  Useful utility for getting event loop. Acts like get_event_loop(), but also creates new event loop if needed

• idle –

  Useful for making event loop idle in the main thread for other threads to work

async_to_sync

async_to_sync(source: T) -> T

Convert all public async methods/properties of an object to universal methods.

See async_to_sync_wraps for more info

Parameters:

• source

  (object) –

  object to convert

Returns:

• object ( T ) –

  converted object. Note that parameter passed is being modified anyway

Source code in universalasync/wrapper.py

def async_to_sync(source: T) -> T:
    """Convert all public async methods/properties of an object to universal methods.

    See [`async_to_sync_wraps`][universalasync.wrapper.async_to_sync_wraps] for more info

    Args:
        source (object): object to convert

    Returns:
        object: converted object. Note that parameter passed is being modified anyway
    """
    for name in dir(source):
        method = getattr(source, name)

        if not name.startswith("_"):
            if inspect.iscoroutinefunction(method) or inspect.isasyncgenfunction(method) or inspect.isdatadescriptor(method):
                function = getattr(source, name)
                setattr(source, name, async_to_sync_wraps(function))

        elif name == "__aenter__" and not hasattr(source, "__enter__"):
            source.__enter__ = async_to_sync_wraps(method)  # type: ignore

        elif name == "__aexit__" and not hasattr(source, "__exit__"):
            source.__exit__ = async_to_sync_wraps(method)  # type: ignore

    return source

async_to_sync_wraps

async_to_sync_wraps(function: Callable) -> Callable

Wrap an async method/property to universal method.

This allows to run wrapped methods in both async and sync contexts transparently without any additional code

When run from another thread, it runs coroutines in new thread's event loop

See Example for full example

Parameters:

• function

  (Callable) –

  function/property to wrap

Returns:

• Callable ( Callable ) –

  modified function

Referenced by:

• API Reference  universalasync  async_to_sync

Source code in universalasync/wrapper.py

def async_to_sync_wraps(function: Callable) -> Callable:
    """Wrap an async method/property to universal method.

    This allows to run wrapped methods in both async and sync contexts transparently without any additional code

    When run from another thread, it runs coroutines in new thread's event loop

    See [Example](index.md#example-of-usage) for full example

    Args:
        function (Callable): function/property to wrap

    Returns:
        Callable: modified function
    """
    is_property = inspect.isdatadescriptor(function)
    if is_property:
        function = cast(types.MethodDescriptorType, function).__get__

    @functools.wraps(function)
    def async_to_sync_wrap(*args: Any, **kwargs: Any) -> Any:
        loop = get_event_loop()
        coroutine = function(*args, **kwargs)

        if loop.is_running():
            return coroutine
        return run_sync_ctx(coroutine, loop)

    result = async_to_sync_wrap
    if is_property:
        result = cast(Callable, property(cast(Callable, result)))  # type: ignore
    return result

get_event_loop

get_event_loop() -> AbstractEventLoop

Useful utility for getting event loop. Acts like get_event_loop(), but also creates new event loop if needed

This will return a working event loop in 100% of cases.

Returns:

• AbstractEventLoop –

  asyncio.AbstractEventLoop: event loop

Source code in universalasync/utils.py

def get_event_loop() -> asyncio.AbstractEventLoop:
    """Useful utility for getting event loop. Acts like get_event_loop(), but also creates new event loop if needed

    This will return a working event loop in 100% of cases.

    Returns:
        asyncio.AbstractEventLoop: event loop
    """
    loop = _get_event_loop()
    if loop.is_closed():
        return _create_new_event_loop()
    return loop

idle async

idle() -> None

Useful for making event loop idle in the main thread for other threads to work

Source code in universalasync/__init__.py

@async_to_sync_wraps
async def idle() -> None:
    """Useful for making event loop idle in the main thread for other threads to work"""
    is_idling = True

    def signal_handler(*args: Any, **kwargs: Any) -> None:
        nonlocal is_idling

        is_idling = False

    for s in (signal.SIGINT, signal.SIGTERM, signal.SIGABRT):
        signal.signal(s, signal_handler)

    while is_idling:
        await asyncio.sleep(1)