Merge remote-tracking branch 'origin/develop' into macedo/scxml

Author: fgmacedo

docs/authors.md

@@ -10,6 +10,7 @@
 * [Rafael Rêgo](mailto:crafards@gmail.com)
 * [Raphael Schrader](mailto:raphael@schradercloud.de)
 * [João S. O. Bueno](mailto:gwidion@gmail.com)
+* [Rodrigo Nogueira](mailto:rodrigo.b.nogueira@gmail.com)
 
 
 ## Scaffolding

statemachine/signature.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 from functools import partial
 from inspect import BoundArguments
 from inspect import Parameter
@@ -6,6 +8,12 @@
 from itertools import chain
 from types import MethodType
 from typing import Any
+from typing import FrozenSet
+from typing import Optional
+from typing import Tuple
+
+BindCacheKey = Tuple[int, FrozenSet[str]]
+BindTemplate = Tuple[Tuple[str, ...], Optional[str], Optional[str]]  # noqa: UP007
 
 
 def _make_key(method):
@@ -44,6 +52,11 @@ def cached_function(cls, method):
 
 class SignatureAdapter(Signature):
     is_coroutine: bool = False
+    _bind_cache: dict[BindCacheKey, BindTemplate]
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self._bind_cache = {}
 
     @classmethod
     @signature_cache
@@ -60,19 +73,71 @@ def from_callable(cls, method):
         adapter.is_coroutine = iscoroutinefunction(method)
         return adapter
 
-    def bind_expected(self, *args: Any, **kwargs: Any) -> BoundArguments:  # noqa: C901
+    def bind_expected(self, *args: Any, **kwargs: Any) -> BoundArguments:
+        cache_key: BindCacheKey = (len(args), frozenset(kwargs.keys()))
+        template = self._bind_cache.get(cache_key)
+
+        if template is not None:
+            return self._fast_bind(args, kwargs, template)
+
+        result = self._full_bind(cache_key, *args, **kwargs)
+        return result
+
+    def _fast_bind(
+        self,
+        args: tuple[Any, ...],
+        kwargs: dict[str, Any],
+        template: BindTemplate,
+    ) -> BoundArguments:
+        param_names, kwargs_param_name, var_positional_name = template
+        arguments: dict[str, Any] = {}
+        past_var_positional = False
+
+        for i, name in enumerate(param_names):
+            if name == var_positional_name:
+                # Collect all remaining positional args into a tuple
+                arguments[name] = args[i:]
+                past_var_positional = True
+            elif past_var_positional:
+                # After *args, remaining params are keyword-only
+                arguments[name] = kwargs.get(name)
+            elif i < len(args):
+                # Match _full_bind: if param is also in kwargs, kwargs wins
+                # (POSITIONAL_OR_KEYWORD params prefer kwargs over positional args)
+                if name in kwargs:
+                    arguments[name] = kwargs[name]
+                else:
+                    arguments[name] = args[i]
+            else:
+                arguments[name] = kwargs.get(name)
+
+        if kwargs_param_name is not None:
+            matched = set(param_names)
+            arguments[kwargs_param_name] = {k: v for k, v in kwargs.items() if k not in matched}
+
+        return BoundArguments(self, arguments)  # type: ignore[arg-type]
+
+    def _full_bind(  # noqa: C901
+        self,
+        cache_key: BindCacheKey,
+        *args: Any,
+        **kwargs: Any,
+    ) -> BoundArguments:
         """Get a BoundArguments object, that maps the passed `args`
         and `kwargs` to the function's signature.  It avoids to raise `TypeError`
         trying to fill all the required arguments and ignoring the unknown ones.
 
         Adapted from the internal `inspect.Signature._bind`.
         """
-        arguments = {}
+        arguments: dict[str, Any] = {}
+        param_names_used: list[str] = []
 
         parameters = iter(self.parameters.values())
         arg_vals = iter(args)
         parameters_ex: Any = ()
         kwargs_param = None
+        kwargs_param_name: str | None = None
+        var_positional_name: str | None = None
 
         while True:
             # Let's iterate through the positional arguments and corresponding
@@ -140,12 +205,15 @@ def bind_expected(self, *args: Any, **kwargs: Any) -> BoundArguments:  # noqa: C
                         values = [arg_val]
                         values.extend(arg_vals)
                         arguments[param.name] = tuple(values)
+                        param_names_used.append(param.name)
+                        var_positional_name = param.name
                         break
 
                     if param.name in kwargs and param.kind != Parameter.POSITIONAL_ONLY:
                         arguments[param.name] = kwargs.pop(param.name)
                     else:
                         arguments[param.name] = arg_val
+                    param_names_used.append(param.name)
 
         # Now, we iterate through the remaining parameters to process
         # keyword arguments
@@ -171,14 +239,19 @@ def bind_expected(self, *args: Any, **kwargs: Any) -> BoundArguments:  # noqa: C
                 # arguments.
                 pass
             else:
-                arguments[param_name] = arg_val  #
+                arguments[param_name] = arg_val
+                param_names_used.append(param_name)
 
         if kwargs:
             if kwargs_param is not None:
                 # Process our '**kwargs'-like parameter
-                arguments[kwargs_param.name] = kwargs  # type: ignore [assignment]
+                arguments[kwargs_param.name] = kwargs  # type: ignore[assignment]
+                kwargs_param_name = kwargs_param.name
             else:
                 # 'ignoring we got an unexpected keyword argument'
                 pass
 
-        return BoundArguments(self, arguments)  # type: ignore [arg-type]
+        template: BindTemplate = (tuple(param_names_used), kwargs_param_name, var_positional_name)
+        self._bind_cache[cache_key] = template
+
+        return BoundArguments(self, arguments)  # type: ignore[arg-type]

tests/test_signature.py

@@ -2,7 +2,9 @@
 from functools import partial
 
 import pytest
+
 from statemachine.dispatcher import callable_method
+from statemachine.signature import SignatureAdapter
 
 
 def single_positional_param(a):
@@ -161,3 +163,131 @@ def test_support_for_partial(self):
 
         assert wrapped_func("A", "B") == ("A", "B", "activated")
         assert wrapped_func.__name__ == positional_and_kw_arguments.__name__
+
+
+def named_and_kwargs(source, **kwargs):
+    return source, kwargs
+
+
+class TestCachedBindExpected:
+    """Tests that exercise the cache fast-path by calling the same
+    wrapped function twice with the same argument shape."""
+
+    def setup_method(self):
+        SignatureAdapter.from_callable.clear_cache()
+
+    def test_named_param_not_leaked_into_kwargs(self):
+        """Named params should not appear in the **kwargs dict on cache hit."""
+        wrapped = callable_method(named_and_kwargs)
+
+        # 1st call: cache miss -> _full_bind
+        result1 = wrapped(source="A", target="B", event="go")
+        assert result1 == ("A", {"target": "B", "event": "go"})
+
+        # 2nd call: cache hit -> _fast_bind
+        result2 = wrapped(source="X", target="Y", event="stop")
+        assert result2 == ("X", {"target": "Y", "event": "stop"})
+
+    def test_kwargs_only_receives_unmatched_keys_with_positional(self):
+        """When mixing positional and keyword args with **kwargs."""
+        wrapped = callable_method(named_and_kwargs)
+
+        result1 = wrapped("A", target="B")
+        assert result1 == ("A", {"target": "B"})
+
+        result2 = wrapped("X", target="Y")
+        assert result2 == ("X", {"target": "Y"})
+
+    def test_var_positional_collected_as_tuple(self):
+        """VAR_POSITIONAL (*args) must be collected into a tuple on cache hit."""
+
+        def fn(*args, **kwargs):
+            return args, kwargs
+
+        wrapped = callable_method(fn)
+
+        result1 = wrapped(1, 2, 3, key="val")
+        assert result1 == ((1, 2, 3), {"key": "val"})
+
+        result2 = wrapped(4, 5, key="other")
+        assert result2 == ((4, 5), {"key": "other"})
+
+    def test_keyword_only_after_var_positional(self):
+        """KEYWORD_ONLY params after *args must be extracted from kwargs on cache hit."""
+
+        def fn(*args, event, **kwargs):
+            return args, event, kwargs
+
+        wrapped = callable_method(fn)
+
+        result1 = wrapped(100, event="ev1", source="s0")
+        assert result1 == ((100,), "ev1", {"source": "s0"})
+
+        result2 = wrapped(200, event="ev2", source="s1")
+        assert result2 == ((200,), "ev2", {"source": "s1"})
+
+    def test_positional_or_keyword_prefers_kwargs_over_positional(self):
+        """When a POSITIONAL_OR_KEYWORD param is in both args and kwargs, kwargs wins."""
+
+        def fn(event, source, target):
+            return event, source, target
+
+        wrapped = callable_method(fn)
+
+        # 1st call: positional arg provided but 'event' also in kwargs -> kwargs wins
+        result1 = wrapped("discarded_content", event="ev1", source="s0", target="t0")
+        assert result1 == ("ev1", "s0", "t0")
+
+        # 2nd call: cache hit, same behavior expected
+        result2 = wrapped("other_content", event="ev2", source="s1", target="t1")
+        assert result2 == ("ev2", "s1", "t1")
+
+    def test_empty_var_positional(self):
+        """Empty *args is handled correctly on cache hit."""
+
+        def fn(*args, **kwargs):
+            return args, kwargs
+
+        wrapped = callable_method(fn)
+
+        # 1st call with args
+        result1 = wrapped(1, key="val")
+        assert result1 == ((1,), {"key": "val"})
+
+        # 2nd call: only kwargs, no positional args — different cache key (len=0)
+        result2 = wrapped(key="val2")
+        assert result2 == ((), {"key": "val2"})
+
+        # 3rd call: hits cache for len=0
+        result3 = wrapped(key="val3")
+        assert result3 == ((), {"key": "val3"})
+
+    def test_named_params_before_var_positional(self):
+        """Named params before *args are filled correctly on cache hit."""
+
+        def fn(a, b, *args, **kwargs):
+            return a, b, args, kwargs
+
+        wrapped = callable_method(fn)
+
+        result1 = wrapped(1, 2, 3, 4, key="val")
+        assert result1 == (1, 2, (3, 4), {"key": "val"})
+
+        result2 = wrapped(10, 20, 30, key="val2")
+        assert result2 == (10, 20, (30,), {"key": "val2"})
+
+    def test_kwargs_wins_with_var_positional_present(self):
+        """POSITIONAL_OR_KEYWORD before *args prefers kwargs when name matches."""
+
+        def fn(event, *args, **kwargs):
+            return event, args, kwargs
+
+        wrapped = callable_method(fn)
+
+        # 1st call: 'event' in both positional and kwargs — kwargs wins
+        result1 = wrapped("discarded", "extra", event="ev1", key="a")
+        assert result1 == ("ev1", ("extra",), {"key": "a"})
+
+        # 2nd call: cache hit, same behavior
+        result2 = wrapped("other", "more", event="ev2", key="b")
+        assert result2 == ("ev2", ("more",), {"key": "b"})