Skip to content

marvin.utilities.asyncio

Utilities for working with asyncio.

ExposeSyncMethodsMixin

A mixin that can take functions decorated with expose_sync_method and automatically create synchronous versions.

create_task

Creates async background tasks in a way that is safe from garbage collection.

See https://textual.textualize.io/blog/2023/02/11/the-heisenbug-lurking-in-your-async-code/

Example:

async def my_coro(x: int) -> int: return x + 1

safely submits my_coro for background execution

create_task(my_coro(1))

expose_sync_method

Decorator that automatically exposes synchronous versions of async methods. Note it doesn't work with classmethods.

Parameters:

Name Type Description Default
name str

The name of the synchronous method.

required

Returns:

Type Description
Callable[..., Any]

The decorated function.

Example

Basic usage:

class MyClass(ExposeSyncMethodsMixin):

    @expose_sync_method("my_method")
    async def my_method_async(self):
        return 42

my_instance = MyClass()
await my_instance.my_method_async() # returns 42
my_instance.my_method()  # returns 42

make_sync

Creates a synchronous function from an asynchronous function.

run_async async

Runs a synchronous function in an asynchronous manner.

Parameters:

Name Type Description Default
fn Callable[..., T]

The function to run.

required
*args Any

Positional arguments to pass to the function.

()
**kwargs Any

Keyword arguments to pass to the function.

{}

Returns:

Type Description
T

The return value of the function.

Example

Basic usage:

def my_sync_function(x: int) -> int:
    return x + 1

await run_async(my_sync_function, 1)

run_sync

Runs a coroutine from a synchronous context. A thread will be spawned to run the event loop if necessary, which allows coroutines to run in environments like Jupyter notebooks where the event loop runs on the main thread.

Parameters:

Name Type Description Default
coroutine Coroutine[Any, Any, T]

The coroutine to run.

required

Returns:

Type Description
T

The return value of the coroutine.

Example

Basic usage:

async def my_async_function(x: int) -> int:
    return x + 1

run_sync(my_async_function(1))

run_sync_if_awaitable

If the object is awaitable, run it synchronously. Otherwise, return the object.

Parameters:

Name Type Description Default
obj Any

The object to run.

required

Returns:

Type Description
Any

The return value of the object if it is awaitable, otherwise the object

Any

itself.

Example

Basic usage:

async def my_async_function(x: int) -> int:
    return x + 1

run_sync_if_awaitable(my_async_function(1))