Callable utils

Utilities for calling sync or async functions in PyAgenity.

This module provides helpers to detect async callables and to invoke functions that may be synchronous or asynchronous, handling thread pool execution and awaitables.

Functions:

Name	Description
call_sync_or_async	Call a function that may be sync or async, returning its result.
run_coroutine	Run an async coroutine from a sync context safely.

Functions

call_sync_or_async async

call_sync_or_async(func, *args, **kwargs)

Call a function that may be sync or async, returning its result.

If the function is synchronous, it runs in a thread pool to avoid blocking the event loop. If the result is awaitable, it is awaited before returning.

Parameters:

Name	Type	Description	Default
func	Callable[..., Any]	The function to call.	required
*args		Positional arguments for the function.	()
**kwargs		Keyword arguments for the function.	{}

Returns:

Name	Type	Description
Any	Any	The result of the function call, awaited if necessary.

Source code in pyagenity/utils/callable_utils.py

async def call_sync_or_async(func: Callable[..., Any], *args, **kwargs) -> Any:
    """
    Call a function that may be sync or async, returning its result.

    If the function is synchronous, it runs in a thread pool to avoid blocking
    the event loop. If the result is awaitable, it is awaited before returning.

    Args:
        func (Callable[..., Any]): The function to call.
        *args: Positional arguments for the function.
        **kwargs: Keyword arguments for the function.

    Returns:
        Any: The result of the function call, awaited if necessary.
    """
    if _is_async_callable(func):
        return await func(*args, **kwargs)

    # Call sync function in a thread pool
    result = await asyncio.to_thread(func, *args, **kwargs)
    # If the result is awaitable, await it
    if inspect.isawaitable(result):
        return await result
    return result

run_coroutine

run_coroutine(func)

Run an async coroutine from a sync context safely.

Source code in pyagenity/utils/callable_utils.py

def run_coroutine(func: Coroutine) -> Any:
    """Run an async coroutine from a sync context safely."""
    # Always try to get/create an event loop and use thread-safe execution
    try:
        loop = asyncio.get_running_loop()
    except RuntimeError:
        # No loop running, create one
        return asyncio.run(func)

    # Loop is running, use thread-safe execution
    fut = asyncio.run_coroutine_threadsafe(func, loop)
    return fut.result()