API Reference¶

universalasync ¶

Modules:

loop_local –
utils –
wrapper –

Classes:

LoopLocal –

Keep one value per event loop, like threading.local does per thread, and close each value on its own loop.

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

LoopLocal ¶

LoopLocal(close: Callable[[T], Coroutine[Any, Any, Any]])


              flowchart TD
              universalasync.LoopLocal[LoopLocal]

              

              click universalasync.LoopLocal href "" "universalasync.LoopLocal"

Keep one value per event loop, like threading.local does per thread, and close each value on its own loop.

Wrapped methods run on the current thread's event loop, so resources bound to an event loop (like aiohttp.ClientSession) need one instance per loop. LoopLocal creates them lazily, and closes every one of them on the loop that owns it when close is called, or when the LoopLocal is garbage collected.

Parameters:

close ¶
(Callable) –

async function closing a value. It must not reference the object holding this LoopLocal, or that object is never garbage collected. When the loop owning a value is already closed, the value is closed on another loop, so close must not need the closed loop

Referenced by:

universalasync Overview

Methods:

close –

Close all values, each on the event loop that owns it
get –

Return the value of the current event loop, creating it first if needed

Source code in universalasync/loop_local.py

def __init__(self, close: Callable[[T], Coroutine[Any, Any, Any]]) -> None:
    self._close = close
    self._values: dict[asyncio.AbstractEventLoop, T] = {}
    weakref.finalize(self, _close_values_or_raise, self._values, close)

close ¶

close() -> Any

Close all values, each on the event loop that owns it

Referenced by:

API Reference universalasync LoopLocal

Source code in universalasync/loop_local.py

def close(self) -> Any:
    """Close all values, each on the event loop that owns it"""
    if asyncio._get_running_loop() is None:
        _close_values_or_raise(self._values, self._close)
        return None
    return self._close_async()

get ¶

get(create: Callable[[], T]) -> T

Return the value of the current event loop, creating it first if needed

When a value is created, values of event loops closed since then are closed and dropped.

Parameters:

create ¶
(Callable) –

function creating the value. Unlike close, it is not stored, so it may reference the object holding this LoopLocal

Returns:

T ( T ) –

value bound to the current event loop

Source code in universalasync/loop_local.py

def get(self, create: Callable[[], T]) -> T:
    """Return the value of the current event loop, creating it first if needed

    When a value is created, values of event loops closed since then are closed and dropped.

    Args:
        create (Callable): function creating the value. Unlike `close`, it is not stored,
            so it may reference the object holding this `LoopLocal`

    Returns:
        T: value bound to the current event loop
    """
    loop = get_event_loop()
    if loop not in self._values:
        closed = {owner: self._values.pop(owner) for owner in list(self._values) if owner.is_closed()}
        for error in _close_values(closed, self._close):
            _report_close_error(loop, error)
        self._values[loop] = create()
    return self._values[loop]

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)

API Reference¶

universalasync ¶

LoopLocal ¶

`close` ¶

close ¶

get ¶

`create` ¶

async_to_sync ¶

`source` ¶

async_to_sync_wraps ¶

`function` ¶

get_event_loop ¶

idle `async` ¶

API Reference¶

universalasync ¶

LoopLocal ¶

close ¶

close ¶

get ¶

create ¶

async_to_sync ¶

source ¶

async_to_sync_wraps ¶

function ¶

get_event_loop ¶

idle async ¶

`close` ¶

`create` ¶

`source` ¶

`function` ¶

idle `async` ¶