omnipy.util.callable_types

CLASS	DESCRIPTION
CallableType

FUNCTION	DESCRIPTION
callable_type_from_flags
decorate_callable_by_type
decorate_result_by_type

CallableType

Bases: Enum

ATTRIBUTE	DESCRIPTION
ASYNC
ASYNC_GENERATOR
SYNC
SYNC_GENERATOR

Source code in src/omnipy/util/callable_types.py

class CallableType(Enum):
    SYNC = auto()
    SYNC_GENERATOR = auto()
    ASYNC = auto()
    ASYNC_GENERATOR = auto()

ASYNC class-attribute instance-attribute

ASYNC = auto()

ASYNC_GENERATOR class-attribute instance-attribute

ASYNC_GENERATOR = auto()

SYNC class-attribute instance-attribute

SYNC = auto()

SYNC_GENERATOR class-attribute instance-attribute

SYNC_GENERATOR = auto()

callable_type_from_flags

callable_type_from_flags(*, has_async: bool, has_generator: bool) -> CallableType

Source code in src/omnipy/util/callable_types.py

def callable_type_from_flags(*, has_async: bool, has_generator: bool) -> CallableType:
    if has_async:
        if has_generator:
            return CallableType.ASYNC_GENERATOR
        else:
            return CallableType.ASYNC
    else:
        if has_generator:
            return CallableType.SYNC_GENERATOR
        else:
            return CallableType.SYNC

decorate_callable_by_type

decorate_callable_by_type(
    call_func: Callable[_P, Any],
    call_type: CallableType,
    *,
    context_factory: Callable[[], AbstractContextManager[object]] | None = None,
    resolve_async_result: bool = False,
) -> Callable[_P, Any]

Source code in src/omnipy/util/callable_types.py

def decorate_callable_by_type(  # noqa: C901
    call_func: Callable[_P, Any],
    call_type: CallableType,
    *,
    context_factory: Callable[[], AbstractContextManager[object]] | None = None,
    resolve_async_result: bool = False,
) -> Callable[_P, Any]:
    from omnipy.util.helpers import resolve

    make_context = context_factory or nullcontext

    if call_type is CallableType.SYNC:

        def _sync_wrapper(*args: _P.args, **kwargs: _P.kwargs):
            with make_context():
                return call_func(*args, **kwargs)

        return _sync_wrapper

    elif call_type is CallableType.SYNC_GENERATOR:

        def _sync_generator_wrapper(*args: _P.args, **kwargs: _P.kwargs):
            with make_context():
                yield from call_func(*args, **kwargs)

        return _sync_generator_wrapper

    elif call_type is CallableType.ASYNC:

        async def _async_wrapper(*args: _P.args, **kwargs: _P.kwargs):
            with make_context():
                result = call_func(*args, **kwargs)
                if resolve_async_result:
                    return await resolve(result)
                return await result

        return _async_wrapper

    else:

        async def _async_generator_wrapper(*args: _P.args, **kwargs: _P.kwargs):
            with make_context():
                job_result = call_func(*args, **kwargs)
                sent = None
                try:
                    while True:
                        sent = yield await job_result.asend(sent)
                except StopAsyncIteration:
                    return

        return _async_generator_wrapper

decorate_result_by_type

decorate_result_by_type(
    *, on_finished: Callable[[], None] | None = None
) -> Callable[[Callable[_P, _R]], Callable[_P, _R]]

Source code in src/omnipy/util/callable_types.py

def decorate_result_by_type(
    *,
    on_finished: Callable[[], None] | None = None,
) -> Callable[[Callable[_P, _R]], Callable[_P, _R]]:
    register_finished = on_finished or _noop

    def _decorator(call_func: Callable[_P, _R]) -> Callable[_P, _R]:
        @functools.wraps(call_func)
        def _result_wrapper(*args: _P.args, **kwargs: _P.kwargs) -> _R:
            return _decorate_result_with_finished_callback(
                call_func(*args, **kwargs),
                register_finished,
            )

        return _result_wrapper

    return _decorator