Good morning, thank you so much for taking the time to respond to my proposal.
There are different ways to determine whether we are in an asynchronous context or not. For example, we can also do this to determine if we are in a synchronous or asynchronous context:
CO_COROUTINE = getattr(inspect, 'CO_COROUTINE', 0x0080)
def _called_from_async(depth: int = 2) -> bool:
try:
frame = sys._getframe(depth)
return bool(frame.f_code.co_flags & CO_COROUTINE)
except (ValueError, AttributeError):
return False
Although I’ve tested the code you suggested and it works correctly, I’m seeing that the problem might occur when we’re in an asynchronous function calling another synchronous one. In that case, it can produce a false positive result indicating that it’s being executed within an asynchronous function.
Through testing, sys._getframe works correctly and is slightly faster (from 2µs with asyncio.get_event_loop to 1.27µs with sys._getframe).
Example of coroutinedispatch with sys._getframe:
import inspect
import asyncio
import sys
from typing import Any, Callable, get_type_hints, get_origin
from functools import lru_cache
class coroutinedispatch:
CO_COROUTINE = getattr(inspect, "CO_COROUTINE", 0x0080)
def __init__(self, func: Callable):
self._sync_methods = {}
self._async_methods = {}
self._name = func.__name__
arg_types = self.get_arg_types(func)
if inspect.iscoroutinefunction(func):
self._async_methods[arg_types] = func
else:
self._sync_methods[arg_types] = func
def _called_from_async(self, depth: int = 2) -> bool:
"""Detect if caller is inside an async def."""
try:
frame = sys._getframe(depth)
return bool(frame.f_code.co_flags & self.CO_COROUTINE)
except (ValueError, AttributeError):
return False
def get_arg_types(self, func: Callable) -> tuple:
try:
hints = get_type_hints(func)
sig = inspect.signature(func)
params = list(sig.parameters.values())
if params and params[0].name in ("self", "cls"):
params = params[1:]
return tuple(hints.get(p.name, Any) for p in params)
except Exception:
return ()
@lru_cache(maxsize=128)
def match_types(self, provided_args: tuple, expected_types: tuple) -> bool:
if len(provided_args) != len(expected_types):
return False
for arg, expected in zip(provided_args, expected_types):
if expected is Any:
continue
origin = get_origin(expected)
if origin is not None:
expected = origin
if not isinstance(arg, expected):
return False
return True
@lru_cache(maxsize=128)
def _find_matching_method(self, is_async: bool, *args: Any) -> Callable:
"""Find the method that matches the argument types."""
methods = self._async_methods if is_async else self._sync_methods
# Search for exact match
for arg_types, method in methods.items():
if self.match_types(args, arg_types):
return method
# If no match, raise descriptive error
arg_type_names = tuple(type(arg).__name__ for arg in args)
context = "async" if is_async else "sync"
available = list(methods.keys())
raise TypeError(
f"No matching {context} method '{self._name}' found for "
f"arguments: {arg_type_names}. Available: {available}"
)
def register(self, func: Callable) -> Callable:
"""Register a new method overload."""
arg_types = self.get_arg_types(func)
if inspect.iscoroutinefunction(func):
self._async_methods[arg_types] = func
else:
self._sync_methods[arg_types] = func
return self
def __call__(self, *args: Any, **kwargs: Any) -> Any:
is_async_context = self._called_from_async()
# Exclude 'self' from arguments if present
check_args = args[1:] if args and hasattr(args[0], self._name) else args
try:
method = self._find_matching_method(is_async_context, *check_args)
result = method(*args, **kwargs)
return result
except TypeError:
# If no async/sync method, try the other
try:
is_async_context = not is_async_context
method = self._find_matching_method(is_async_context, *check_args)
result = method(*args, **kwargs)
return result
except TypeError:
raise
def __get__(self, obj, objtype=None):
"""Support for bound methods"""
if obj is None:
return self
import functools
return functools.partial(self.__call__, obj)
I look forward to any further suggestions or problems you might encounter.