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))