Keyword only args work, incl defaults

punq/__init__.py

@@ -156,20 +156,24 @@ class _Empty:
 empty = _Empty()
 
 
-def _match_defaults(args, defaults):
+def _match_defaults(spec):
     """Matches args with their defaults in the result of getfullargspec.
 
     inspect.getfullargspec returns a complex object that includes the defaults
     on args and kwonly args. This function takes a list of args, and a tuple of
     the last N defaults and returns a dict of args to defaults.
     """
-    if defaults is None:
-        return {}
+    ns = {}
+    if spec.defaults is not None:
+        offset = len(spec.args) - len(spec.defaults)
+        defaults = ([None] * offset) + list(spec.defaults)
 
-    offset = len(args) - len(defaults)
-    defaults = ([None] * offset) + list(defaults)
+        ns = {key: value for key, value in zip(spec.args, defaults) if value is not None}
 
-    return {key: value for key, value in zip(args, defaults) if value is not None}
+    if spec.kwonlydefaults is not None:
+        ns.update(spec.kwonlydefaults)
+
+    return ns
 
 
 class _Registry:
@@ -239,7 +243,7 @@ def register_service_and_instance(self, service, instance):
         """
         self.__registrations[service].append(_Registration(service, Scope.singleton, lambda: instance, {}, {}))
 
-    def register_concrete_service(self, service, scope):
+    def register_concrete_service(self, service, scope, resolve_args=None):
         """Register a service as its own implementation.
 
         Examples:
@@ -259,7 +263,7 @@ def register_concrete_service(self, service, scope):
         if not inspect.isclass(service):
             raise InvalidSelfRegistrationError(service)
         self.__registrations[service].append(
-            _Registration(service, scope, service, self._get_needs_for_ctor(service), {})
+            _Registration(service, scope, service, self._get_needs_for_ctor(service), resolve_args or {})
         )
 
     def build_context(self, key, existing=None):
@@ -283,7 +287,7 @@ def register(self, service, factory=empty, instance=empty, scope=Scope.transient
         if instance is not empty:
             self.register_service_and_instance(service, instance)
         elif factory is empty:
-            self.register_concrete_service(service, scope)
+            self.register_concrete_service(service, scope, resolve_args)
         elif callable(factory):
             self.register_service_and_impl(service, scope, factory, resolve_args)
         else:
@@ -450,9 +454,9 @@ def resolve_all(self, service, **kwargs):
     def _build_impl(self, registration, resolution_args, context):
         """Instantiate the registered service."""
         spec = inspect.getfullargspec(registration.builder)
-        target_args = spec.args
+        target_args = spec.args + spec.kwonlyargs
 
-        args = _match_defaults(spec.args, spec.defaults)
+        args = _match_defaults(spec)
         args.update({
             k: self._resolve_impl(v, resolution_args, context, args.get(k))
             for k, v in registration.needs.items()

tests/test_kwonly_deps.py

@@ -0,0 +1,71 @@
+from typing import Protocol
+
+import punq
+
+
+class Parser(Protocol):
+    def parse(self, val: str) -> str: ...
+
+
+class Writer(Protocol):
+    def write(self, val: str) -> None: ...
+
+
+class Doer:
+    def __init__(self, *, p: Parser, w: Writer, name: str = "Barry"):
+        self.parser = p
+        self.writer = w
+        self.name = name
+
+    def do(self, val: str):
+        self.writer.write(f"{self.name}: {self.parser.parse(val)}")
+
+
+class ListWriter:
+    def __init__(self, target: list):
+        self.output = target
+
+    def write(self, val) -> None:
+        self.output.append(val)
+
+
+class ReverseParser:
+    def parse(self, val) -> str:
+        return val[::-1]
+
+
+class IdentityParser:
+    def parse(self, val) -> str:
+        return val
+
+
+def test_can_resolve_with_kwonlyargs():
+    container = punq.Container()
+    result = []
+    container.register(Parser, ReverseParser)
+    container.register(Writer, instance=ListWriter(result))
+    container.register(Doer)
+
+    doer = container.resolve(Doer)
+    doer.do("hello world")
+
+    otherdoer = container.resolve(Doer, name="Tom")
+    otherdoer.do("dlrow olleh")
+
+    assert result == ["Barry: dlrow olleh", "Tom: hello world"]
+
+
+def test_can_register_with_kwonlyargs():
+    container = punq.Container()
+    result = []
+    container.register(Parser, ReverseParser)
+    container.register(Writer, instance=ListWriter(result))
+    container.register(Doer, name="Frodo")
+
+    doer = container.resolve(Doer)
+    doer.do("hello world")
+
+    otherdoer = container.resolve(Doer, name="Bilbo", p=IdentityParser())
+    otherdoer.do("hello world")
+
+    assert result == ["Frodo: dlrow olleh", "Bilbo: hello world"]

tox.ini

@@ -16,4 +16,3 @@ allowlist_externals = uv
 commands =
     uv sync --python {envpython}
     uv run python -m pytest --doctest-modules tests --cov --cov-config=pyproject.toml --cov-report=xml
-    mypy