You’re welcome to write in your own language, then use a traditional translation tool to translate to English. That will get way better results and still convey your personal voice and thoughts, rather than generating something very generic and verbose that doesn’t sound like a human.
1 Like
I’ve prepared a pull request in an example project that could be used to simplify classes and would also simplify a migration from WSGI to ASGI.
1 Like
Yes, I’m going to start doing that. Many times what I also tried to do was try to address all the points I wanted to express or try to make sure I expressed myself well.
Thanks, that helped a bunch. I was focusing on the call sites, but there is more to it than that, and the common client helps isolate migration changes to just the call sites which I hadn’t considered. I think the async and sync clients could be combined in one even without your proposal, but from what I’ve seen that doesn’t seem to be common, most differentiate by type of client, and then use different names for the sync and async methods even though they could be named the same thing.
I’m still skeptical the non-obviousness of whether a call returns the value or a coroutine that must be awaited is worth it, and also don’t have a clear understanding of what the deterministic rules used by coroutinedispatch are.
In my proof-of-concept code, the context is detected using a couple of simple deterministic rules:
- Inspecting the calling function’s frame (sys._getframe)
When the decorated method is called, _called_from_async() looks one step up the call stack (sys._getframe(2)) to inspect the frame of the function that initiated the call.
- Checking the bytecode flag (CO_COROUTINE)
CPython sets specific flags on a function’s compiled code object (f_code.co_flags). I check if the CO_COROUTINE flag (0x0080) is active using a bitwise AND operation:
bool(frame.f_code.co_flags & self.CO_COROUTINE)
The result is true if the calling function was defined with async def.
The result is false if the calling function is a function with standard def.
- Fallback Rules
Interpreter Safety: If sys._getframe fails or is incompatible with the environment, the error is caught and safely set to False (synchronous context).
Missing Overloads: If an asynchronous context is detected, but there is no registered asynchronous method for those argument types, TypeError is caught and a compatible synchronous version is checked (and vice versa).
That’s assuming that, in all cases, when you’re being called asynchronously, the next level up IS an async function. You’re assuming there’s no wrapper layer of any kind. Keep in mind, there’s nothing wrong with calling an async function from a sync function; you get back a coroutine object that presumably you’ll need to await later, but for example, you could map() an async function over a list, and then await them as a group.
I would not consider frame and bytecode inspection a reliable way to detect which variant of a function to call reliable, even though it is deterministic. For example, perhaps a sync function calls async functions to create a coroutine that another function schedules and awaits for, but IIUC your process will incorrectly infer a coroutinedispatch should be sync when the user actually needs it to be async. To adapt to this assumption they would have to change their sync function to async and await it. This will introduce the possibility of a context switch and may not be feasible or require redesigning the code to work around this constraint imposed by a library that uses coroutinedispatch. I would be very reluctant to force my users to color their functions to get dispatch to work “properly”…I’d just give them two methods and they can use the one they need without worrying about inferences being wrong and the implications of working around them.
I do understand (I think) the motivation for this library, and agree that if it were simple it would be a great idea. I remain very skeptical that a solution can be found that “just works” and doesn’t move the complexity around in ways that don’t create problems that are worse than the one you are trying to solve.
Has its usability been explored by using it in any real-world projects? Code that demonstrates functionality often overlooks practical considerations that only become apparent once applied to actual use-cases, a variety of patterns, or non-trivial workflows. As I understand it, this is practically a requirement for adoption into the standard library to ensure it is broadly usable, won’t be a source of endless bugs, or abandon ware that needs to be supported for many years as it works through the deprecation and end-of-life process. Are any projects actually using this decorator?
I see a person using a lib mostly in two ways.
- Everything is sync
- Everything is async but the “jump” in a main() or something like that
At least is what I see on code.
When everything I code is sync, nothing changes. When everything I code is async, I am not rewriting any code. Am just writing within the expectation that User.objects.all() is awaitable, for example.
In a most-async codebase, I expect most stuff to be awaitable anyway. In that sense, there is no “switch”.
Now if you are considering an existing lib with an existing code using it, of course will be a breaking change!
I do see for both the lib and the usercode maintainers.
Lib maintainers:
- Can let both codes very together, easier to catch a drift during maintenance.
- Can have the good names on both sync and async versions
- Need not to have a SyncClass (or Class) and AsyncClass around, with BaseClass or ABC to deal with.
- Given the contracts are kept for both versions (and I am assuming that a sane maintainer keeping
fetchandfetch_asyncdoing the same thing will also letfetch.__acall__equal), it does not add more documentation and semantic load. It lowers the load.
The acallable thing emerged from a private lib that I had to maintain that become an architecture mess with more structure effort than “meat” feature effort.
Usercode maintainers:
- Need not to search for the “async version” of the stuff to be used. (Does
AsyncClassexist? or onlyClass? Are they feature-par?). If a function/method is acallable, this is the async version! If not, no one exists. - Need not to deal with the ugly renaming of async stuff. Well, readability counts more for some people but I really do not like to do
fetch_async(table.afilter(...).aexclude(...).aorder_by(...).avalues_list()) - My sync code does not even know that async is possible, but I learn the API
- I already know the lib API for my next async code.
In the end, this is most about discoverability and ergonomics for usercode devs; and less boilerplate & drift maintenance and easier ergonomics for lib devs.
…at least from my POV, who wrote it from frustration with a lib boilerplate maintenance 
The same could be said about fetch and fetch_async (or afetch). It is not about the dispatching mechanics, do you agree? If there are vastly different semantics then the maintainer is doing something weird.
I do not see an existing lib or usercode to demand a rewrite. Unless it is a breaking change on the lib. But that will need a rewrite anyway, no?
1 Like
Would you prefer that a just-decorated def fetch() to raise a NotImplementedError instead? Well, this is easily changed if this is the problem. I just tried to let something “useful” as default. At least a naked coro can be upgraded and not break the usercode expectations.
That said, or your point (1) will never occur because your will never let an acallable without the nondefault async implementation, or we will agree to change the default to something better.
The dispatching decision on acallable is inspecting CO_COROUTINE. In fact, evolved from the code you shared here on this thread.
I am not following something here. acallable is not doing anything different than @coroutinedispatch. Is only doing exactly this but in an easier-to-use way imho
Again I am not following something. My IDE detected acallable-decorated functions exactly as you described how @coroutinedispatch would be: int | Awaitable[int] for example.
I can see the benefit of “being sure” of what version is being called. But you can force this by calling a property that forces this.
fetch -> int | Awaitable[int]fetch.sync -> intfetch.__acall__ -> Awaitable[int]
Yet for your lib this would be not enough, and ok
Bottomline, what we are discussing really seems to be:
is beneficial to be able to dispatch based on call site being sync or async?
I think that it does. And having a proper way, in stdlib, besides frame introspection, would be welcome.
This thread started with:
My comments concerning migrating code are in this context. Specifically, the application has a mix of sync and async and the end user of the library is actively working to move sync to async.
That is exactly the point I made in the comment you then quoted without the relevant context. This same comment was clearly in the context of migrating as it said “to actually switch from sync to async”. To reiterate, a common name for sync and async variants of a function does little to help with migration because the code needs to change regardless of using an identical function name.
As for the benefits:
I have acknowledge the benefits to the library maintainers, but also that ease of use for the library users should be prioritized over maintainer ease of use since the reason the library exists at all is for the convenience of the users not the maintainers.
The “good names” aspect is closely tied to having or not having a shared class that contains the methods. If you have a separate class (SyncClass and AsyncClass) they can both share the same method names, just don’t put those different methods on a base class since they aren’t the same (some return one thing while the other returns a coroutine). If you want a common base class then they should have different names since they are semantically different. Pick your poison. The concern I have raised is that naming them the same thing when they are semantically different creates problems for the library users that are worse than the problem (for the library maintainers) that is trying to be solved. This tradeoff does not make sense to me. To repeat myself, If the problems on the user side did not exist then it would be a clear win because it would make the maintenance easier without a downside.
I am not surprised that maintaining a library that supports both sync and async APIs had more code to implement the two divergent APIs than to implement the functionality they exposed. It is not uncommon for APIs to have as much code as the functionality, and you had two APIs. I’m not sure how relevant this is to this discussion though since the proposal in this thread doesn’t offer a common implementation of the API variants, just a way to wrap it so the API functions have the same name across the sync and async variants. If your solution to the problems related to this proposal did help with this, that’s great, but not really relevant to this thread. If yours was discussed somewhere can you share the link (apparently I have an interest in this topic and would like to see how you tackled these issues).
I’m not following this benefit…function/methods are always callable, even async functions. Being callable is what makes them functions. Can you clarify the benefit you are claiming in this bullet point?
Yes, when switching from sync to async you don’t have to rename the function, but you do have to switch to awaiting it, which negates much of the benefit of having a common name IMO since you still have to await the value the call returns. This is not about readability (I tried to clarify that in a previous comment to another commenter).
Prefixing async variants with ‘a’ doesn’t make it so I have to learn another API for the async. It is trivial to know that the async variants is prefixed with ‘a’. This is a common paradigm within python itself (_enter_ → _aenter_, etc).
This is the crux of my position as well. The rules for which variant is called are not discoverable or ergonomic IMO since they are hidden behind a decorator that (in this proposal at least) inspects frames and bytecodes. I do not disagree that there are clear benefits to library maintainers, but I think their convenience is less of a priority than the users of the libraries.
To be clear, the semantics I was referring to are the ones inherent with one implementation being sync and the other async. One returns the value, the other returns a coroutine that needs to be awaited to get the value. So, yes, fetch and afetch have vastly different semantics IMO. But those semantics are clear in the function definition and the naming convention, whereas @coroutinedispatch obfuscates those semantic differences.
This was justification in the proposal I was raising concerns with.
I see. Thanks for the clarification. Will prepare a proper response later for the other topics, but just quick specifically on this one:
I am naming “acallable” something with __call__ AND __acall__.
Having sync and async versions, dispatched automatically.
Sorry for the confusion.
Thanks for clarifying. I also realized I misread ‘acallable’ as just callable. I think it would have made sense if I’d read it properly 
1 Like
To this day, I don’t know of any projects that use this decorator. This idea came to me when I was working with Django and was forced to migrate from WSGI to ASGI. Django’s ORM has many asynchronous versions of its methods, from get to aget, from first to afirst, and so on. If Django had used this feature, the change would have been trivial, but since they didn’t, I had to dedicate more than a day to the migration.
1 Like
Ultimately, this decorator isn’t meant to be used in 100% of the code; there are cases where it makes sense and cases where it doesn’t. In the case you’ve suggested, it would make its equivalent synchronous call and execute as the library creators intended. Or, if it’s in your own code, you’ll know that if you want to receive a coroutine in a synchronous environment, you shouldn’t use this decorator.
You’re right, it makes more sense to return a NotImplementedError. Regarding the typing issue, it’s something I’d have to address more carefully and with more time. I’d probably have to implement a plugin, and I’ll do that if this proposal gains some traction. For now, after fixing some bugs and adding a few improvements, it would look like this.
"""Overload a callable by argument types, distinguishing sync from async
implementations by the calling context rather than by name."""
import functools
import inspect
import sys
import warnings
from collections.abc import Callable
from typing import Any, NamedTuple, Union, get_args, get_origin, get_type_hints
# Fixed separation between tiers so a concrete type always outranks any
# Union (regardless of how deep its members are), which always outranks Any.
_ANY_SPECIFICITY = 0
_UNION_PENALTY = 5000
_CONCRETE_BASE = 10000
class _Overload(NamedTuple):
"""One registered overload: its recorded parameter types (self/cls
stripped), the matching bound-away signature, and the function to call."""
arg_types: tuple
signature: inspect.Signature
func: Callable
def _describe_types(arg_types: tuple) -> tuple:
"""Render recorded parameter types for error messages, e.g. (int, str)
instead of (<class 'int'>, <class 'str'>)."""
return tuple("Any" if t is Any else getattr(t, "__name__", str(t)) for t in arg_types)
def _get_signature(func: Callable) -> inspect.Signature:
sig = inspect.signature(func)
params = list(sig.parameters.values())
if params and params[0].name in ("self", "cls"):
params = params[1:]
return sig.replace(parameters=params)
def _get_arg_types(func: Callable) -> tuple:
try:
hints = get_type_hints(func)
params = _get_signature(func).parameters.values()
return tuple(hints.get(p.name, Any) for p in params)
except (NameError, TypeError, ValueError):
return ()
def _has_leading_self(func: Callable) -> bool:
params = list(inspect.signature(func).parameters.values())
return bool(params) and params[0].name in ("self", "cls")
def _type_matches(arg: Any, expected: Any) -> bool:
if expected is Any:
return True
origin = get_origin(expected)
if origin is Union:
return any(_type_matches(arg, member) for member in get_args(expected))
if origin is not None:
expected = origin
return isinstance(arg, expected)
def _match_types(provided_args: tuple, expected_types: tuple) -> bool:
if len(provided_args) != len(expected_types):
return False
return all(
_type_matches(arg, expected) for arg, expected in zip(provided_args, expected_types)
)
def _param_specificity(expected: Any) -> int:
"""Higher = more specific. concrete type > Union[...] > Any."""
if expected is Any:
return _ANY_SPECIFICITY
origin = get_origin(expected)
if origin is Union:
# A Union is only ever as specific as its least specific member,
# further discounted, so it never outranks pinning to one exact type.
members = get_args(expected)
return min(_param_specificity(member) for member in members) - _UNION_PENALTY
base = origin if origin is not None else expected
mro_len = len(getattr(base, "__mro__", (base,)))
has_type_args = 1 if origin is not None else 0
return _CONCRETE_BASE + mro_len * 10 + has_type_args
class coroutinedispatch:
"""Dispatch an overloaded callable by argument types and by whether the
call site is inside an ``async def`` or not.
Register the first implementation with ``@coroutinedispatch`` and any
number of additional sync/async overloads with ``.register``. At call
time:
1. The calling frame is inspected to decide whether this is a sync or
an async call (see ``_called_from_async``).
2. Every overload registered for that context is bound against the
actual call arguments (applying defaults), and the most specific
matching overload is selected: a concrete type outranks
``Union[...]``, which outranks ``Any``; a deeper subclass outranks a
shallower one. Ties keep the first-registered overload.
3. If nothing matches in that context, the *other* context is tried
before giving up, so a purely-sync overload can still be called from
async code and vice versa.
"""
CO_COROUTINE = getattr(inspect, "CO_COROUTINE", 0x0080)
def __init__(self, func: Callable):
self._sync_methods: list[_Overload] = []
self._async_methods: list[_Overload] = []
self._method_cache = {}
self._name = func.__name__
self._is_method = _has_leading_self(func)
self._register(func)
def _register(self, func: Callable) -> None:
is_async = inspect.iscoroutinefunction(func)
overload = _Overload(_get_arg_types(func), _get_signature(func), func)
target = self._async_methods if is_async else self._sync_methods
if any(existing.arg_types == overload.arg_types for existing in target):
context = "async" if is_async else "sync"
warnings.warn(
f"'{self._name}' already has a registered {context} overload for "
f"argument types {overload.arg_types!r}; keeping the earlier one.",
stacklevel=3,
)
return
target.append(overload)
def _called_from_async(self) -> bool:
"""Detect if the caller outside this module is inside an async def."""
try:
frame = sys._getframe(1)
except ValueError:
return False
while frame is not None and frame.f_globals.get("__name__") == __name__:
frame = frame.f_back
if frame is None:
return False
return bool(frame.f_code.co_flags & self.CO_COROUTINE)
def _find_matching_method(self, is_async: bool, args: tuple, kwargs: dict) -> Callable:
"""Find the most specific method matching the argument types.
Binds kwargs/defaults per candidate first, then among every overload
that matches, picks the one with the highest total per-parameter
specificity. Ties keep the first-registered candidate.
"""
cache_key = (
is_async,
tuple(type(arg) for arg in args),
tuple(sorted((key, type(value)) for key, value in kwargs.items())),
)
cached = self._method_cache.get(cache_key)
if cached is not None:
return cached
overloads = self._async_methods if is_async else self._sync_methods
best_func = None
best_score = None
for overload in overloads:
try:
bound = overload.signature.bind(*args, **kwargs)
except TypeError:
continue
bound.apply_defaults()
values = tuple(bound.arguments[name] for name in overload.signature.parameters)
if not _match_types(values, overload.arg_types):
continue
score = sum(_param_specificity(t) for t in overload.arg_types)
if best_score is None or score > best_score:
best_score = score
best_func = overload.func
if best_func is not None:
self._method_cache[cache_key] = best_func
return best_func
arg_type_names = tuple(type(arg).__name__ for arg in args)
context = "async" if is_async else "sync"
available = [_describe_types(overload.arg_types) for overload in overloads]
raise NotImplementedError(
f"No matching {context} method '{self._name}' found for "
f"arguments: {arg_type_names}, keyword arguments: {sorted(kwargs)}. "
f"Available: {available}"
)
def register(self, func: Callable) -> Callable:
"""Register a new method overload."""
self._register(func)
return self
def __call__(self, *args: Any, **kwargs: Any) -> Any:
is_async_context = self._called_from_async()
# Exclude 'self'/'cls' only when this dispatcher wraps a method
check_args = args[1:] if self._is_method and args else args
method = self._resolve(is_async_context, check_args, kwargs)
return method(*args, **kwargs)
def _resolve(self, is_async_context: bool, args: tuple, kwargs: dict) -> Callable:
"""Try the calling context first, then the other, before giving up.
This lets a purely-sync overload still be called from async code
(and vice versa) instead of failing just because nothing was
registered for the context that happened to make the call.
"""
try:
return self._find_matching_method(is_async_context, args, kwargs)
except NotImplementedError:
pass
try:
return self._find_matching_method(not is_async_context, args, kwargs)
except NotImplementedError:
pass
context = "async" if is_async_context else "sync"
sync_available = [_describe_types(overload.arg_types) for overload in self._sync_methods]
async_available = [
_describe_types(overload.arg_types) for overload in self._async_methods
]
raise NotImplementedError(
f"No matching method '{self._name}' found for the given arguments "
f"(called from {context} context). "
f"Available sync: {sync_available}. Available async: {async_available}."
)
def __get__(self, obj, objtype=None):
"""Support for bound methods"""
if obj is None:
return self
return functools.partial(self.__call__, obj)
The trouble is, I’ve described a place where the caller knows that it doesn’t make sense to use this decorator, but the caller isn’t the one to make that decision. So does that mean no library should ever use it?
1 Like
This doesn’t mean that no library should use it. In the case you’ve suggested, you’d have to adapt your code to how the library intends it to be used, and if that means you need to be in an asynchronous context to retrieve a coroutine, then you’d have to do it.
It also wouldn’t change the logic you’ve proposed by using an asynchronous context. You can also retrieve coroutines in an asynchronous context and await them when needed.
Forcing a caller to use an async method with an await to create a coroutine may be more than a mere inconvenience. It forces them to introduce a context switch which may make managing concurrency more difficult (ie introduce a lock rather than simply not yielding to other tasks). @alanjds has an implementation that provides the sync and async versions so code can select which one to use rather than relying on the dispatch, but that negates a lot of the benefits by bypassing the dispatching. I think a library that forces users to use async functions to create coroutines is too limited. I think you should consider adding a way for users to explicitly decide which variant should be used when the dispatch rules don’t align with their needs.
I understand your concern. I’ve just implemented the run_sync and run_async methods, which are optional and force the method that will be returned.
You can use:
# Context-dependent
client.get_awesome_data()
# Context-independent
client.get_awesome_data.run_async()
client.get_awesome_data.run_sync()
This would be the implementation
"""Overload a callable by argument types, distinguishing sync from async
implementations by the calling context rather than by name."""
from __future__ import annotations
import inspect
import sys
import warnings
from collections.abc import Callable
from typing import Any, NamedTuple, Union, get_args, get_origin, get_type_hints
# Fixed separation between tiers so a concrete type always outranks any
# Union (regardless of how deep its members are), which always outranks Any.
_ANY_SPECIFICITY = 0
_UNION_PENALTY = 5000
_CONCRETE_BASE = 10000
class _Overload(NamedTuple):
"""One registered overload: its recorded parameter types (self/cls
stripped), the matching bound-away signature, and the function to call."""
arg_types: tuple
signature: inspect.Signature
func: Callable
def _describe_types(arg_types: tuple) -> tuple:
"""Render recorded parameter types for error messages, e.g. (int, str)
instead of (<class 'int'>, <class 'str'>)."""
return tuple("Any" if t is Any else getattr(t, "__name__", str(t)) for t in arg_types)
def _get_signature(func: Callable) -> inspect.Signature:
sig = inspect.signature(func)
params = list(sig.parameters.values())
if params and params[0].name in ("self", "cls"):
params = params[1:]
return sig.replace(parameters=params)
def _get_arg_types(func: Callable) -> tuple:
try:
hints = get_type_hints(func)
params = _get_signature(func).parameters.values()
return tuple(hints.get(p.name, Any) for p in params)
except (NameError, TypeError, ValueError):
return ()
def _has_leading_self(func: Callable) -> bool:
params = list(inspect.signature(func).parameters.values())
return bool(params) and params[0].name in ("self", "cls")
def _type_matches(arg: Any, expected: Any) -> bool:
if expected is Any:
return True
origin = get_origin(expected)
if origin is Union:
return any(_type_matches(arg, member) for member in get_args(expected))
if origin is not None:
expected = origin
return isinstance(arg, expected)
def _match_types(provided_args: tuple, expected_types: tuple) -> bool:
if len(provided_args) != len(expected_types):
return False
return all(
_type_matches(arg, expected) for arg, expected in zip(provided_args, expected_types)
)
def _param_specificity(expected: Any) -> int:
"""Higher = more specific. concrete type > Union[...] > Any."""
if expected is Any:
return _ANY_SPECIFICITY
origin = get_origin(expected)
if origin is Union:
# A Union is only ever as specific as its least specific member,
# further discounted, so it never outranks pinning to one exact type.
members = get_args(expected)
return min(_param_specificity(member) for member in members) - _UNION_PENALTY
base = origin if origin is not None else expected
mro_len = len(getattr(base, "__mro__", (base,)))
has_type_args = 1 if origin is not None else 0
return _CONCRETE_BASE + mro_len * 10 + has_type_args
class coroutinedispatch:
"""Dispatch an overloaded callable by argument types and by whether the
call site is inside an ``async def`` or not.
Register the first implementation with ``@coroutinedispatch`` and any
number of additional sync/async overloads with ``.register``. At call
time:
1. The calling frame is inspected to decide whether this is a sync or
an async call (see ``_called_from_async``).
2. Every overload registered for that context is bound against the
actual call arguments (applying defaults), and the most specific
matching overload is selected: a concrete type outranks
``Union[...]``, which outranks ``Any``; a deeper subclass outranks a
shallower one. Ties keep the first-registered overload.
3. If nothing matches in that context, the *other* context is tried
before giving up, so a purely-sync overload can still be called from
async code and vice versa.
Automatic detection is a default, not a mandate: call ``.run_sync(...)``
or ``.run_async(...)`` to bypass it and force a specific variant when
the detected context doesn't match what the caller actually needs (for
instance, to avoid an unwanted context switch, or to get back a
coroutine to schedule explicitly instead of awaiting it immediately).
"""
CO_COROUTINE = getattr(inspect, "CO_COROUTINE", 0x0080)
def __init__(self, func: Callable):
self._sync_methods: list[_Overload] = []
self._async_methods: list[_Overload] = []
self._method_cache = {}
self._name = func.__name__
self._is_method = _has_leading_self(func)
self._register(func)
def _register(self, func: Callable) -> None:
is_async = inspect.iscoroutinefunction(func)
overload = _Overload(_get_arg_types(func), _get_signature(func), func)
target = self._async_methods if is_async else self._sync_methods
if any(existing.arg_types == overload.arg_types for existing in target):
context = "async" if is_async else "sync"
warnings.warn(
f"'{self._name}' already has a registered {context} overload for "
f"argument types {overload.arg_types!r}; keeping the earlier one.",
stacklevel=3,
)
return
target.append(overload)
def _called_from_async(self) -> bool:
"""Detect if the caller outside this module is inside an async def."""
try:
frame = sys._getframe(1)
except ValueError:
return False
while frame is not None and frame.f_globals.get("__name__") == __name__:
frame = frame.f_back
if frame is None:
return False
return bool(frame.f_code.co_flags & self.CO_COROUTINE)
def _find_matching_method(self, is_async: bool, args: tuple, kwargs: dict) -> Callable:
"""Find the most specific method matching the argument types.
Binds kwargs/defaults per candidate first, then among every overload
that matches, picks the one with the highest total per-parameter
specificity. Ties keep the first-registered candidate.
"""
cache_key = (
is_async,
tuple(type(arg) for arg in args),
tuple(sorted((key, type(value)) for key, value in kwargs.items())),
)
cached = self._method_cache.get(cache_key)
if cached is not None:
return cached
overloads = self._async_methods if is_async else self._sync_methods
best_func = None
best_score = None
for overload in overloads:
try:
bound = overload.signature.bind(*args, **kwargs)
except TypeError:
continue
bound.apply_defaults()
values = tuple(bound.arguments[name] for name in overload.signature.parameters)
if not _match_types(values, overload.arg_types):
continue
score = sum(_param_specificity(t) for t in overload.arg_types)
if best_score is None or score > best_score:
best_score = score
best_func = overload.func
if best_func is not None:
self._method_cache[cache_key] = best_func
return best_func
arg_type_names = tuple(type(arg).__name__ for arg in args)
context = "async" if is_async else "sync"
available = [_describe_types(overload.arg_types) for overload in overloads]
raise NotImplementedError(
f"No matching {context} method '{self._name}' found for "
f"arguments: {arg_type_names}, keyword arguments: {sorted(kwargs)}. "
f"Available: {available}"
)
def register(self, func: Callable) -> Callable:
"""Register a new method overload."""
self._register(func)
return self
def __call__(self, *args: Any, **kwargs: Any) -> Any:
is_async_context = self._called_from_async()
# Exclude 'self'/'cls' only when this dispatcher wraps a method
check_args = args[1:] if self._is_method and args else args
method = self._resolve(is_async_context, check_args, kwargs)
return method(*args, **kwargs)
def run_sync(self, *args: Any, **kwargs: Any) -> Any:
"""Force the sync overload, bypassing automatic context detection."""
return self._call_forced(is_async=False, args=args, kwargs=kwargs)
def run_async(self, *args: Any, **kwargs: Any) -> Any:
"""Force the async overload, bypassing automatic context detection.
Returns the coroutine as-is; the caller is responsible for awaiting
(or otherwise scheduling) it.
"""
return self._call_forced(is_async=True, args=args, kwargs=kwargs)
def _call_forced(self, is_async: bool, args: tuple, kwargs: dict) -> Any:
check_args = args[1:] if self._is_method and args else args
method = self._find_matching_method(is_async, check_args, kwargs)
return method(*args, **kwargs)
def _resolve(self, is_async_context: bool, args: tuple, kwargs: dict) -> Callable:
"""Try the calling context first, then the other, before giving up.
This lets a purely-sync overload still be called from async code
(and vice versa) instead of failing just because nothing was
registered for the context that happened to make the call.
"""
try:
return self._find_matching_method(is_async_context, args, kwargs)
except NotImplementedError:
pass
try:
return self._find_matching_method(not is_async_context, args, kwargs)
except NotImplementedError:
pass
context = "async" if is_async_context else "sync"
sync_available = [_describe_types(overload.arg_types) for overload in self._sync_methods]
async_available = [
_describe_types(overload.arg_types) for overload in self._async_methods
]
raise NotImplementedError(
f"No matching method '{self._name}' found for the given arguments "
f"(called from {context} context). "
f"Available sync: {sync_available}. Available async: {async_available}."
)
def __get__(self, obj: Any, objtype: type | None = None) -> coroutinedispatch | _BoundDispatch:
"""Support for bound methods"""
if obj is None:
return self
return _BoundDispatch(self, obj)
class _BoundDispatch:
"""`instance.method` result: callable with automatic sync/async
dispatch, plus `.run_sync`/`.run_async` to force one explicitly."""
__slots__ = ("_dispatcher", "_obj")
def __init__(self, dispatcher: coroutinedispatch, obj: Any):
self._dispatcher = dispatcher
self._obj = obj
def __call__(self, *args: Any, **kwargs: Any) -> Any:
return self._dispatcher(self._obj, *args, **kwargs)
def run_sync(self, *args: Any, **kwargs: Any) -> Any:
return self._dispatcher.run_sync(self._obj, *args, **kwargs)
def run_async(self, *args: Any, **kwargs: Any) -> Any:
return self._dispatcher.run_async(self._obj, *args, **kwargs)
1 Like