asyncio

marvin.utilities.asyncio

Constants

T

T = TypeVar('T')

Functions

run_sync

def run_sync(coro: Coroutine[Any, Any, T]) -> T

Run a coroutine synchronously.

This function uses asyncio to run a coroutine in a synchronous context. It attempts the following strategies in order:

1. If no event loop is running, creates a new one and runs the coroutine
2. If a loop is running, attempts to run the coroutine on that loop
3. As a last resort, creates a new thread with its own event loop to run the coroutine

Context variables are properly propagated between threads in all cases.

Example:

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

result = run_sync(f(1))

Args: coro: The coroutine to run synchronously

Returns: The result of the coroutine

run_sync_in_thread

def run_sync_in_thread(coro: Coroutine[Any, Any, T]) -> T

Run a coroutine synchronously in a new thread.

This function creates a new thread with its own event loop to run the coroutine. Context variables are properly propagated between threads. This is useful when you need to run async code in a context where you can’t use the current event loop (e.g., inside an async frame).

Example:

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

result = run_sync_in_thread(f(1))

Args: coro: The coroutine to run synchronously

Returns: The result of the coroutine

---

Parent Module: utilities