Speed up cached factory resolution

Author: danielstarman

src/miniject/_container.py

@@ -303,33 +303,38 @@ def _invoke_factory(
         **overrides: Any,
     ) -> Any:
         """Call a factory, resolving its parameters from the container."""
-        sig_and_hints = introspect_factory(factory, resolution_error=ResolutionError)
-        if sig_and_hints is None:
-            if inspect.iscoroutinefunction(factory):
-                _raise_async_resolution_error(factory, stack=_stack)
-            instance = factory()
-            if inspect.isawaitable(instance):
-                if inspect.iscoroutine(instance):
-                    instance.close()
-                _raise_async_resolution_error(factory, stack=_stack, returned_awaitable=True)
-            return instance
-        sig, hints = sig_and_hints
+        plan = introspect_factory(factory, resolution_error=ResolutionError)
+        if plan is None:
+            return _call_sync_factory_checked(
+                factory,
+                stack=_stack,
+                is_async=inspect.iscoroutinefunction(factory),
+            )
+        fast_args = self._build_fast_positional_args(
+            plan.fast_positional_deps,
+            _stack=_stack,
+            overrides=overrides,
+        )
+        if fast_args is not None:
+            return _call_sync_factory_checked(
+                factory,
+                args=fast_args,
+                stack=_stack,
+                is_async=plan.is_async,
+            )
         kwargs = self._build_factory_kwargs(
             factory,
-            sig,
-            hints,
+            plan.signature,
+            plan.hints,
             _stack=_stack,
             overrides=overrides,
         )
-        if inspect.iscoroutinefunction(factory):
-            _raise_async_resolution_error(factory, stack=_stack)
-
-        instance = factory(**kwargs)
-        if inspect.isawaitable(instance):
-            if inspect.iscoroutine(instance):
-                instance.close()
-            _raise_async_resolution_error(factory, stack=_stack, returned_awaitable=True)
-        return instance
+        return _call_sync_factory_checked(
+            factory,
+            kwargs=kwargs,
+            stack=_stack,
+            is_async=plan.is_async,
+        )
 
     async def _invoke_factory_async(
         self,
@@ -339,17 +344,26 @@ async def _invoke_factory_async(
         **overrides: Any,
     ) -> Any:
         """Call a factory, awaiting it if needed and resolving deps async."""
-        sig_and_hints = introspect_factory(factory, resolution_error=ResolutionError)
-        if sig_and_hints is None:
+        plan = introspect_factory(factory, resolution_error=ResolutionError)
+        if plan is None:
             instance = factory()
             if inspect.isawaitable(instance):
                 return await instance
             return instance
-        sig, hints = sig_and_hints
+        fast_args = await self._build_fast_positional_args_async(
+            plan.fast_positional_deps,
+            _stack=_stack,
+            overrides=overrides,
+        )
+        if fast_args is not None:
+            instance = factory(*fast_args)
+            if inspect.isawaitable(instance):
+                return await instance
+            return instance
         kwargs = await self._build_factory_kwargs_async(
             factory,
-            sig,
-            hints,
+            plan.signature,
+            plan.hints,
             _stack=_stack,
             overrides=overrides,
         )
@@ -397,6 +411,24 @@ def _build_factory_kwargs(
             )
         return kwargs
 
+    def _build_fast_positional_args(
+        self,
+        fast_positional_deps: tuple[type, ...] | None,
+        *,
+        _stack: tuple[type, ...],
+        overrides: dict[str, Any],
+    ) -> tuple[object, ...] | None:
+        if fast_positional_deps is None or overrides:
+            return None
+
+        resolved: list[object] = []
+        for dep in fast_positional_deps:
+            if self._find_binding(dep) is None:
+                return None
+            resolved_instance = cast("object", self._resolve(dep, _stack=_stack))
+            resolved.append(resolved_instance)
+        return tuple(resolved)
+
     async def _build_factory_kwargs_async(
         self,
         factory: Callable[..., Any],
@@ -438,6 +470,24 @@ async def _build_factory_kwargs_async(
             )
         return kwargs
 
+    async def _build_fast_positional_args_async(
+        self,
+        fast_positional_deps: tuple[type, ...] | None,
+        *,
+        _stack: tuple[type, ...],
+        overrides: dict[str, Any],
+    ) -> tuple[object, ...] | None:
+        if fast_positional_deps is None or overrides:
+            return None
+
+        resolved: list[object] = []
+        for dep in fast_positional_deps:
+            if self._find_binding(dep) is None:
+                return None
+            resolved_instance = cast("object", await self._resolve_async(dep, _stack=_stack))
+            resolved.append(resolved_instance)
+        return tuple(resolved)
+
 
 def _require_factory(binding: _Binding) -> Callable[..., Any]:
     """Return a binding factory or raise if the binding is malformed."""
@@ -463,6 +513,25 @@ def _raise_async_resolution_error(
     )
 
 
+def _call_sync_factory_checked(
+    factory: Callable[..., Any],
+    *,
+    stack: tuple[type, ...],
+    is_async: bool,
+    args: tuple[object, ...] = (),
+    kwargs: dict[str, Any] | None = None,
+) -> Any:
+    if is_async:
+        _raise_async_resolution_error(factory, stack=stack)
+
+    instance = factory(*args, **({} if kwargs is None else kwargs))
+    if inspect.isawaitable(instance):
+        if inspect.iscoroutine(instance):
+            instance.close()
+        _raise_async_resolution_error(factory, stack=stack, returned_awaitable=True)
+    return instance
+
+
 def _validate_or_skip_missing_param(
     factory: Callable[..., Any],
     param_name: str,

src/miniject/_introspection.py

@@ -13,7 +13,7 @@
 _DISALLOWED_AUTO_INJECT_TYPES: frozenset[type] = frozenset({bool, bytes, float, int, str})
 _NONE_TYPE: type[None] = type(None)
 _UNION_TYPES: tuple[object, ...] = (typing.Union, types.UnionType)
-_INTROSPECTION_CACHE: dict[object, tuple[inspect.Signature, dict[str, Any]] | None] = {}
+_INTROSPECTION_CACHE: dict[object, FactoryPlan | None] = {}
 
 
 @dataclass(frozen=True, slots=True)
@@ -24,18 +24,28 @@ class ResolvedParamType:
     display_name: str
 
 
+@dataclass(frozen=True, slots=True)
+class FactoryPlan:
+    """Cached callable metadata used during dependency resolution."""
+
+    signature: inspect.Signature
+    hints: dict[str, Any]
+    is_async: bool
+    fast_positional_deps: tuple[type, ...] | None
+
+
 def introspect_factory(
     factory: Callable[..., Any],
     *,
     resolution_error: type[Exception],
-) -> tuple[inspect.Signature, dict[str, Any]] | None:
+) -> FactoryPlan | None:
     """Extract signature and type hints for a factory, or None if not introspectable."""
     try:
         cached = _INTROSPECTION_CACHE.get(factory, _EMPTY)
     except TypeError:
         return _compute_factory_introspection(factory, resolution_error=resolution_error)
     if cached is not _EMPTY:
-        return cast("tuple[inspect.Signature, dict[str, Any]] | None", cached)
+        return cast("FactoryPlan | None", cached)
 
     result = _compute_factory_introspection(factory, resolution_error=resolution_error)
     _INTROSPECTION_CACHE[factory] = result
@@ -46,26 +56,66 @@ def _compute_factory_introspection(
     factory: Callable[..., Any],
     *,
     resolution_error: type[Exception],
-) -> tuple[inspect.Signature, dict[str, Any]] | None:
+) -> FactoryPlan | None:
     """Compute signature and type hints for a factory without caching."""
     hint_target = factory.__init__ if isinstance(factory, type) else factory
+    factory_name = callable_name(factory)
     try:
         sig = inspect.signature(factory)
     except (ValueError, TypeError):
         return None
     except NameError as exc:
         _raise_type_hint_resolution_error(
             hint_target,
-            factory_name=callable_name(factory),
+            factory_name=factory_name,
             resolution_error=resolution_error,
             exc=exc,
         )
     hints = _get_type_hints_or_raise(
         hint_target,
-        factory_name=callable_name(factory),
+        factory_name=factory_name,
         resolution_error=resolution_error,
     )
-    return sig, hints
+    return FactoryPlan(
+        signature=sig,
+        hints=hints,
+        is_async=inspect.iscoroutinefunction(factory),
+        fast_positional_deps=_build_fast_positional_deps(
+            sig,
+            hints,
+            factory_name=factory_name,
+            resolution_error=resolution_error,
+        ),
+    )
+
+
+def _build_fast_positional_deps(
+    sig: inspect.Signature,
+    hints: dict[str, Any],
+    *,
+    factory_name: str,
+    resolution_error: type[Exception],
+) -> tuple[type, ...] | None:
+    deps: list[type] = []
+    for param_name, param in sig.parameters.items():
+        if param_name == "self":
+            continue
+        if param.kind not in (
+            inspect.Parameter.POSITIONAL_ONLY,
+            inspect.Parameter.POSITIONAL_OR_KEYWORD,
+        ):
+            return None
+
+        resolved_type = resolve_param_type(
+            hints.get(param_name),
+            factory_name=factory_name,
+            param_name=param_name,
+            resolution_error=resolution_error,
+        )
+        if resolved_type.binding_key is None or param.default is not inspect.Parameter.empty:
+            return None
+        deps.append(resolved_type.binding_key)
+    return tuple(deps)
 
 
 def _get_type_hints_or_raise(