Skip to content

omnipy.util.callable_types

Helpers and literal enums for classifying and wrapping callable shapes.

CLASS DESCRIPTION
AsyncCallableType

Literal enum values for asynchronous callable variants.
CallableType

Combined literal enum values for every supported callable variant.
GeneratorCallableType

Literal enum values for generator-based callable variants.
PlainCallableType

Literal enum values for non-generator callable variants.
SyncCallableType

Literal enum values for synchronous callable variants.
FUNCTION DESCRIPTION
callable_type_from_flags

Return a callable-type literal from boolean async/generator flags.
decorate_callable_by_type

Wrap a callable while preserving its declared callable shape.
decorate_result_by_type

Create a decorator that runs a callback when a result has fully finished.
get_callable_type

Inspect a callable and classify its sync/async and generator shape.

AsyncCallableType

Bases: _AsyncCoroutineType, _AsyncGeneratorType


              flowchart BT
              omnipy.util.callable_types.AsyncCallableType[AsyncCallableType]
              omnipy.util.callable_types._AsyncCoroutineType[_AsyncCoroutineType]
              omnipy.util.callable_types._AsyncGeneratorType[_AsyncGeneratorType]
              omnipy.util.literal_enum.LiteralEnum[LiteralEnum]

                              omnipy.util.callable_types._AsyncCoroutineType --> omnipy.util.callable_types.AsyncCallableType
                                omnipy.util.literal_enum.LiteralEnum --> omnipy.util.callable_types._AsyncCoroutineType
                

                omnipy.util.callable_types._AsyncGeneratorType --> omnipy.util.callable_types.AsyncCallableType
                                omnipy.util.literal_enum.LiteralEnum --> omnipy.util.callable_types._AsyncGeneratorType
                



              click omnipy.util.callable_types.AsyncCallableType href "" "omnipy.util.callable_types.AsyncCallableType"
              click omnipy.util.callable_types._AsyncCoroutineType href "" "omnipy.util.callable_types._AsyncCoroutineType"
              click omnipy.util.callable_types._AsyncGeneratorType href "" "omnipy.util.callable_types._AsyncGeneratorType"
              click omnipy.util.literal_enum.LiteralEnum href "" "omnipy.util.literal_enum.LiteralEnum"

Literal enum values for asynchronous callable variants.

Subclassed by: Used by:
ATTRIBUTE DESCRIPTION
Literals

Source code in src/omnipy/util/callable_types.py 
47
48
49
50
class AsyncCallableType(_AsyncCoroutineType, _AsyncGeneratorType):
    """Literal enum values for asynchronous callable variants."""

    Literals = Literal[_AsyncCoroutineType.Literals, _AsyncGeneratorType.Literals]

Literals class-attribute instance-attribute 

Literals = Literal[_AsyncCoroutineType.Literals, _AsyncGeneratorType.Literals]
Used by:

CallableType

Bases: SyncCallableType, AsyncCallableType


              flowchart BT
              omnipy.util.callable_types.CallableType[CallableType]
              omnipy.util.callable_types.SyncCallableType[SyncCallableType]
              omnipy.util.callable_types._SyncFunctionType[_SyncFunctionType]
              omnipy.util.callable_types._SyncGeneratorType[_SyncGeneratorType]
              omnipy.util.callable_types.AsyncCallableType[AsyncCallableType]
              omnipy.util.callable_types._AsyncCoroutineType[_AsyncCoroutineType]
              omnipy.util.callable_types._AsyncGeneratorType[_AsyncGeneratorType]
              omnipy.util.literal_enum.LiteralEnum[LiteralEnum]

                              omnipy.util.callable_types.SyncCallableType --> omnipy.util.callable_types.CallableType
                                omnipy.util.callable_types._SyncFunctionType --> omnipy.util.callable_types.SyncCallableType
                                omnipy.util.literal_enum.LiteralEnum --> omnipy.util.callable_types._SyncFunctionType
                

                omnipy.util.callable_types._SyncGeneratorType --> omnipy.util.callable_types.SyncCallableType
                                omnipy.util.literal_enum.LiteralEnum --> omnipy.util.callable_types._SyncGeneratorType
                


                omnipy.util.callable_types.AsyncCallableType --> omnipy.util.callable_types.CallableType
                                omnipy.util.callable_types._AsyncCoroutineType --> omnipy.util.callable_types.AsyncCallableType
                                omnipy.util.literal_enum.LiteralEnum --> omnipy.util.callable_types._AsyncCoroutineType
                

                omnipy.util.callable_types._AsyncGeneratorType --> omnipy.util.callable_types.AsyncCallableType
                                omnipy.util.literal_enum.LiteralEnum --> omnipy.util.callable_types._AsyncGeneratorType
                




              click omnipy.util.callable_types.CallableType href "" "omnipy.util.callable_types.CallableType"
              click omnipy.util.callable_types.SyncCallableType href "" "omnipy.util.callable_types.SyncCallableType"
              click omnipy.util.callable_types._SyncFunctionType href "" "omnipy.util.callable_types._SyncFunctionType"
              click omnipy.util.callable_types._SyncGeneratorType href "" "omnipy.util.callable_types._SyncGeneratorType"
              click omnipy.util.callable_types.AsyncCallableType href "" "omnipy.util.callable_types.AsyncCallableType"
              click omnipy.util.callable_types._AsyncCoroutineType href "" "omnipy.util.callable_types._AsyncCoroutineType"
              click omnipy.util.callable_types._AsyncGeneratorType href "" "omnipy.util.callable_types._AsyncGeneratorType"
              click omnipy.util.literal_enum.LiteralEnum href "" "omnipy.util.literal_enum.LiteralEnum"

Combined literal enum values for every supported callable variant.

Returned by: Used by:
ATTRIBUTE DESCRIPTION
Literals

Source code in src/omnipy/util/callable_types.py 
65
66
67
68
class CallableType(SyncCallableType, AsyncCallableType):
    """Combined literal enum values for every supported callable variant."""

    Literals = Literal[SyncCallableType.Literals, AsyncCallableType.Literals]

Literals class-attribute instance-attribute 

Literals = Literal[SyncCallableType.Literals, AsyncCallableType.Literals]
Returned by: Used by:

GeneratorCallableType

Bases: _SyncGeneratorType, _AsyncGeneratorType


              flowchart BT
              omnipy.util.callable_types.GeneratorCallableType[GeneratorCallableType]
              omnipy.util.callable_types._SyncGeneratorType[_SyncGeneratorType]
              omnipy.util.callable_types._AsyncGeneratorType[_AsyncGeneratorType]
              omnipy.util.literal_enum.LiteralEnum[LiteralEnum]

                              omnipy.util.callable_types._SyncGeneratorType --> omnipy.util.callable_types.GeneratorCallableType
                                omnipy.util.literal_enum.LiteralEnum --> omnipy.util.callable_types._SyncGeneratorType
                

                omnipy.util.callable_types._AsyncGeneratorType --> omnipy.util.callable_types.GeneratorCallableType
                                omnipy.util.literal_enum.LiteralEnum --> omnipy.util.callable_types._AsyncGeneratorType
                



              click omnipy.util.callable_types.GeneratorCallableType href "" "omnipy.util.callable_types.GeneratorCallableType"
              click omnipy.util.callable_types._SyncGeneratorType href "" "omnipy.util.callable_types._SyncGeneratorType"
              click omnipy.util.callable_types._AsyncGeneratorType href "" "omnipy.util.callable_types._AsyncGeneratorType"
              click omnipy.util.literal_enum.LiteralEnum href "" "omnipy.util.literal_enum.LiteralEnum"

Literal enum values for generator-based callable variants.

ATTRIBUTE DESCRIPTION
Literals

Source code in src/omnipy/util/callable_types.py 
59
60
61
62
class GeneratorCallableType(_SyncGeneratorType, _AsyncGeneratorType):
    """Literal enum values for generator-based callable variants."""

    Literals = Literal[_SyncGeneratorType.Literals, _AsyncGeneratorType.Literals]

Literals class-attribute instance-attribute 

Literals = Literal[_SyncGeneratorType.Literals, _AsyncGeneratorType.Literals]

PlainCallableType

Bases: _SyncFunctionType, _AsyncCoroutineType


              flowchart BT
              omnipy.util.callable_types.PlainCallableType[PlainCallableType]
              omnipy.util.callable_types._SyncFunctionType[_SyncFunctionType]
              omnipy.util.callable_types._AsyncCoroutineType[_AsyncCoroutineType]
              omnipy.util.literal_enum.LiteralEnum[LiteralEnum]

                              omnipy.util.callable_types._SyncFunctionType --> omnipy.util.callable_types.PlainCallableType
                                omnipy.util.literal_enum.LiteralEnum --> omnipy.util.callable_types._SyncFunctionType
                

                omnipy.util.callable_types._AsyncCoroutineType --> omnipy.util.callable_types.PlainCallableType
                                omnipy.util.literal_enum.LiteralEnum --> omnipy.util.callable_types._AsyncCoroutineType
                



              click omnipy.util.callable_types.PlainCallableType href "" "omnipy.util.callable_types.PlainCallableType"
              click omnipy.util.callable_types._SyncFunctionType href "" "omnipy.util.callable_types._SyncFunctionType"
              click omnipy.util.callable_types._AsyncCoroutineType href "" "omnipy.util.callable_types._AsyncCoroutineType"
              click omnipy.util.literal_enum.LiteralEnum href "" "omnipy.util.literal_enum.LiteralEnum"

Literal enum values for non-generator callable variants.

ATTRIBUTE DESCRIPTION
Literals

Source code in src/omnipy/util/callable_types.py 
53
54
55
56
class PlainCallableType(_SyncFunctionType, _AsyncCoroutineType):
    """Literal enum values for non-generator callable variants."""

    Literals = Literal[_SyncFunctionType.Literals, _AsyncCoroutineType.Literals]

Literals class-attribute instance-attribute 

Literals = Literal[_SyncFunctionType.Literals, _AsyncCoroutineType.Literals]

SyncCallableType

Bases: _SyncFunctionType, _SyncGeneratorType


              flowchart BT
              omnipy.util.callable_types.SyncCallableType[SyncCallableType]
              omnipy.util.callable_types._SyncFunctionType[_SyncFunctionType]
              omnipy.util.callable_types._SyncGeneratorType[_SyncGeneratorType]
              omnipy.util.literal_enum.LiteralEnum[LiteralEnum]

                              omnipy.util.callable_types._SyncFunctionType --> omnipy.util.callable_types.SyncCallableType
                                omnipy.util.literal_enum.LiteralEnum --> omnipy.util.callable_types._SyncFunctionType
                

                omnipy.util.callable_types._SyncGeneratorType --> omnipy.util.callable_types.SyncCallableType
                                omnipy.util.literal_enum.LiteralEnum --> omnipy.util.callable_types._SyncGeneratorType
                



              click omnipy.util.callable_types.SyncCallableType href "" "omnipy.util.callable_types.SyncCallableType"
              click omnipy.util.callable_types._SyncFunctionType href "" "omnipy.util.callable_types._SyncFunctionType"
              click omnipy.util.callable_types._SyncGeneratorType href "" "omnipy.util.callable_types._SyncGeneratorType"
              click omnipy.util.literal_enum.LiteralEnum href "" "omnipy.util.literal_enum.LiteralEnum"

Literal enum values for synchronous callable variants.

Subclassed by: Used by:
ATTRIBUTE DESCRIPTION
Literals

Source code in src/omnipy/util/callable_types.py 
41
42
43
44
class SyncCallableType(_SyncFunctionType, _SyncGeneratorType):
    """Literal enum values for synchronous callable variants."""

    Literals = Literal[_SyncFunctionType.Literals, _SyncGeneratorType.Literals]

Literals class-attribute instance-attribute 

Literals = Literal[_SyncFunctionType.Literals, _SyncGeneratorType.Literals]
Used by:

callable_type_from_flags 

callable_type_from_flags(*, is_async: bool, is_generator: bool) -> CallableType.Literals

Return a callable-type literal from boolean async/generator flags.

PARAMETER DESCRIPTION
is_async

Whether the callable executes asynchronously.

TYPE: bool

is_generator

Whether the callable yields values instead of returning once.

TYPE: bool

RETURNS DESCRIPTION
CallableType.Literals

CallableType.Literals: Callable variant matching the supplied flags.
Source code in src/omnipy/util/callable_types.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
def callable_type_from_flags(*, is_async: bool, is_generator: bool) -> CallableType.Literals:
    """Return a callable-type literal from boolean async/generator flags.

    Args:
        is_async: Whether the callable executes asynchronously.
        is_generator: Whether the callable yields values instead of returning once.

    Returns:
        CallableType.Literals: Callable variant matching the supplied flags.
    """
    if is_async:
        if is_generator:
            return CallableType.ASYNC_GENERATOR
        else:
            return CallableType.ASYNC_COROUTINE
    else:
        if is_generator:
            return CallableType.SYNC_GENERATOR
        else:
            return CallableType.SYNC_FUNCTION

decorate_callable_by_type 

decorate_callable_by_type(
    call_func: Callable[_P, Any],
    param_signatures: MappingProxyType[str, inspect.Parameter],
    return_type: type,
    call_type: CallableType.Literals,
    *,
    context_factory: Callable[[], AbstractContextManager[object]] | None = None,
    resolve_async_result: bool = False,
) -> Callable[_P, Any]

Wrap a callable while preserving its declared callable shape.

PARAMETER DESCRIPTION
call_func

Callable to wrap.

TYPE: Callable[_P, Any]

param_signatures

Parameter mapping used to expose the desired signature.

TYPE: MappingProxyType[str, inspect.Parameter]

return_type

Return annotation to expose on the wrapper.

TYPE: type

call_type

Effective callable type to emulate.

TYPE: CallableType.Literals

context_factory

Optional factory producing a context manager entered for each invocation.

TYPE: Callable[[], AbstractContextManager[object]] | None DEFAULT: None

resolve_async_result

Whether async-coroutine wrappers should resolve nested awaitables before returning.

TYPE: bool DEFAULT: False

RETURNS DESCRIPTION
Callable[_P, Any]

Callable[_P, Any]: Wrapper matching call_type while preserving the supplied signature metadata.
Source code in src/omnipy/util/callable_types.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
def decorate_callable_by_type(  # noqa: C901
    call_func: Callable[_P, Any],
    param_signatures: MappingProxyType[str, inspect.Parameter],
    return_type: type,
    call_type: CallableType.Literals,
    *,
    context_factory: Callable[[], AbstractContextManager[object]] | None = None,
    resolve_async_result: bool = False,
) -> Callable[_P, Any]:
    """Wrap a callable while preserving its declared callable shape.

    Args:
        call_func: Callable to wrap.
        param_signatures: Parameter mapping used to expose the desired signature.
        return_type: Return annotation to expose on the wrapper.
        call_type: Effective callable type to emulate.
        context_factory: Optional factory producing a context manager entered for each
            invocation.
        resolve_async_result: Whether async-coroutine wrappers should resolve nested
            awaitables before returning.

    Returns:
        Callable[_P, Any]: Wrapper matching ``call_type`` while preserving the supplied
            signature metadata.
    """
    from omnipy.util.helpers import resolve

    make_context = context_factory or nullcontext
    signature = inspect.Signature(
        parameters=tuple(param_signatures.values()), return_annotation=return_type)

    if call_type is CallableType.SYNC_FUNCTION:

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

        _sync_wrapper.__signature__ = signature
        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)

        _sync_generator_wrapper.__signature__ = signature
        return _sync_generator_wrapper

    elif call_type is CallableType.ASYNC_COROUTINE:

        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

        _async_wrapper.__signature__ = signature
        return _async_wrapper

    else:

        async def _async_generator_wrapper(*args: _P.args, **kwargs: _P.kwargs):
            with make_context():
                job_result = await resolve(call_func(*args, **kwargs))

                if isinstance(job_result, GeneratorType):
                    for value in job_result:
                        yield value
                    return

                if not isinstance(job_result, AsyncGeneratorType):
                    raise TypeError('Async generator callable resolved to unsupported result type: '
                                    f'{type(job_result).__name__}')

                sent = None
                try:
                    while True:
                        sent = yield await job_result.asend(sent)
                except StopAsyncIteration:
                    return

        _async_generator_wrapper.__signature__ = signature
        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]]

Create a decorator that runs a callback when a result has fully finished.

The callback is delayed until generators are exhausted and awaitables resolve, so completion is tracked at the end of observable consumption rather than at call time.

PARAMETER DESCRIPTION
on_finished

Callback invoked when the wrapped result is fully consumed.

TYPE: Callable[[], None] | None DEFAULT: None

RETURNS DESCRIPTION
Callable[[Callable[_P, _R]], Callable[_P, _R]]

Callable[[Callable[_P, _R]], Callable[_P, _R]]: Decorator that preserves the wrapped callable while adding completion tracking.
Source code in src/omnipy/util/callable_types.py 
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
def decorate_result_by_type(
    *,
    on_finished: Callable[[], None] | None = None,
) -> Callable[[Callable[_P, _R]], Callable[_P, _R]]:
    """Create a decorator that runs a callback when a result has fully finished.

    The callback is delayed until generators are exhausted and awaitables resolve, so
    completion is tracked at the end of observable consumption rather than at call time.

    Args:
        on_finished: Callback invoked when the wrapped result is fully consumed.

    Returns:
        Callable[[Callable[_P, _R]], Callable[_P, _R]]: Decorator that preserves the
            wrapped callable while adding completion tracking.
    """
    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

get_callable_type 

get_callable_type(callable: Callable[..., Any]) -> CallableType.Literals

Inspect a callable and classify its sync/async and generator shape.

PARAMETER DESCRIPTION
callable

Callable object to inspect.

TYPE: Callable[..., Any]

RETURNS DESCRIPTION
CallableType.Literals

CallableType.Literals: Detected callable variant.
Source code in src/omnipy/util/callable_types.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
def get_callable_type(callable: Callable[..., Any]) -> CallableType.Literals:
    """Inspect a callable and classify its sync/async and generator shape.

    Args:
        callable: Callable object to inspect.

    Returns:
        CallableType.Literals: Detected callable variant.
    """
    if inspect.isasyncgenfunction(callable):
        return CallableType.ASYNC_GENERATOR
    elif inspect.iscoroutinefunction(callable):
        return CallableType.ASYNC_COROUTINE
    elif inspect.isgeneratorfunction(callable):
        return CallableType.SYNC_GENERATOR
    else:
        return CallableType.SYNC_FUNCTION