feat: composable type injection

Author: tandemdude

linkd/__init__.py

@@ -21,6 +21,7 @@
 """A powerful async-only dependency injection framework for Python."""
 
 from linkd import ext
+from linkd.compose import *
 from linkd.conditions import *
 from linkd.container import *
 from linkd.context import *
@@ -35,6 +36,7 @@
     "INJECTED",
     "AutoInjecting",
     "CircularDependencyException",
+    "Compose",
     "Container",
     "ContainerClosedException",
     "Context",

linkd/compose.py

@@ -0,0 +1,106 @@
+# -*- coding: utf-8 -*-
+# Copyright (c) 2025-present tandemdude
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in all
+# copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+# SOFTWARE.
+from __future__ import annotations
+
+__all__ = ["Compose"]
+
+import inspect
+import textwrap
+import typing as t
+
+if t.TYPE_CHECKING:
+    import typing_extensions as t_ex
+    from typing_extensions import dataclass_transform
+else:
+    dataclass_transform = lambda: lambda x: x  # noqa: E731
+
+_ACTUAL_ATTR: t.Final[str] = "__linkd_actual__"
+_DEPS_ATTR: t.Final[str] = "__linkd_deps__"
+
+
+def _is_compose_class(item: t.Any) -> t_ex.TypeIs[type[Compose]]:  # type: ignore[reportUnusedFunction]
+    return hasattr(item, _ACTUAL_ATTR) and inspect.isclass(item)
+
+
+class ComposeMeta(type):
+    """Metaclass handling code generation for user-defined :obj:`~Compose` subclasses."""
+
+    @staticmethod
+    def _codegen_compose_cls(name: str, attrs: dict[str, t.Any]) -> type[t.Any]:
+        joined_names = ",".join(attrs)
+        joined_quoted_names = ",".join(f'"{n}"' for n in attrs) + ("," if len(attrs) == 1 else "")
+
+        lines = [
+            f"class {name}:",
+            f"    __slots__ = ({joined_quoted_names})",
+            f"    def __init__(self,{joined_names}):",
+            *(textwrap.indent(f"self.{n} = {n}", " " * 8) for n in attrs),
+        ]
+
+        exec("\n".join(lines), {}, (generated_locals := {}))
+        return t.cast("type[t.Any]", generated_locals[name])
+
+    def __new__(cls, name: str, bases: tuple[type[t.Any]], attrs: dict[str, t.Any], **kwargs: t.Any) -> type[t.Any]:
+        if attrs["__module__"] == "linkd.compose" and attrs["__qualname__"] == "Compose":
+            return super().__new__(cls, name, bases, attrs)
+
+        generated = cls._codegen_compose_cls(name, attrs["__annotations__"])
+        setattr(generated, _ACTUAL_ATTR, super().__new__(cls, name, bases, attrs))
+
+        return generated
+
+
+@dataclass_transform()
+class Compose(metaclass=ComposeMeta):
+    """
+    Class allowing for "composing" of multiple dependencies into a single object, to help declutter
+    the signatures of functions that require a large number of dependencies.
+
+    If you have ever used msgspec or pydantic this will feel familiar. To use this feature, simply create a
+    subclass of this class and define fields for the dependencies you wish to use.
+
+    .. code-block:: python
+
+        class ComposedDependencies(linkd.Compose):
+            foo: FooDep
+            bar: BarDep
+            baz: BazDep
+
+    Then, in place of specifying ``foo``, ``bar`` and ``baz`` within the function signature, you can request
+    a single object of type ``ComposedDependencies``.
+
+    .. code-block:: python
+
+        async def function(deps: ComposedDependencies):
+            ...
+
+    Linkd will automatically try to create an instance of your composed class with all the dependencies that
+    it requires. As with defining dependencies within the function signature, you can use fallback and `If` or `Try`
+    syntax within the composed class field annotations.
+
+    .. warning::
+        None of the fields may contain a dependency on a composed class.
+
+    .. warning::
+        Composed classes cannot have any defined methods, they will be erased at runtime.
+    """
+
+    __slots__ = ()

linkd/conditions.py

@@ -26,6 +26,7 @@
 import types
 import typing as t
 
+from linkd import compose
 from linkd import exceptions
 from linkd import utils as di_utils
 
@@ -47,7 +48,10 @@ class BaseCondition(abc.ABC):
 
     def __init__(self, inner: type[t.Any] | types.UnionType | tuple[t.Any, ...] | None) -> None:
         if isinstance(inner, tuple) or inner is None or t.get_origin(inner) in (types.UnionType, t.Union):
-            raise SyntaxError("{self.__class__.__name__!r} can only be parameterized by concrete types")
+            raise SyntaxError(f"{self.__class__.__name__!r} can only be parameterized by concrete types")
+
+        if compose._is_compose_class(inner):
+            raise ValueError(f"{self.__class__.__name__!r} cannot be parameterized by composed types")
 
         self.inner: type[t.Any] = inner  # type: ignore[reportAttributeAccessIssue]
         self.inner_id: str = di_utils.get_dependency_id(inner)
@@ -206,6 +210,9 @@ def create(cls, expr: t.Any, /) -> DependencyExpression[t.Any]:
         Returns:
             The created dependency expression.
         """
+        if compose._is_compose_class(expr):
+            raise ValueError("cannot create a dependency expression from a composed type")
+
         requested_dependencies: list[BaseCondition] = []
         required: bool = True
 
@@ -216,6 +223,9 @@ def create(cls, expr: t.Any, /) -> DependencyExpression[t.Any]:
             args = expr.order
 
         for arg in args:
+            if compose._is_compose_class(arg):
+                raise ValueError("composed types cannot be used within dependency expressions")
+
             if arg is types.NoneType or arg is None:
                 required = False
                 continue

linkd/solver.py

@@ -38,6 +38,7 @@
 from collections.abc import Mapping
 from collections.abc import Sequence
 
+from linkd import compose
 from linkd import conditions
 from linkd import container
 from linkd import context as context_
@@ -58,12 +59,14 @@
 R = t.TypeVar("R")
 T = t.TypeVar("T")
 AsyncFnT = t.TypeVar("AsyncFnT", bound=t.Callable[..., Coroutine[t.Any, t.Any, t.Any]])
+DependencyExprOrComposed: t.TypeAlias = t.Union[conditions.DependencyExpression[t.Any], type[compose.Compose]]
+
+LOGGER = logging.getLogger(__name__)
 
 DI_ENABLED: t.Final[bool] = os.environ.get("LINKD_DI_DISABLED", "false").lower() != "true"
 DI_CONTAINER: contextvars.ContextVar[container.Container | None] = contextvars.ContextVar(
     "linkd_container", default=None
 )
-LOGGER = logging.getLogger(__name__)
 
 INJECTED: t.Final[t.Any] = utils.Marker("INJECTED")
 """
@@ -288,17 +291,39 @@ async def close(self) -> None:
 CANNOT_INJECT: t.Final[t.Any] = utils.Marker("CANNOT_INJECT")
 
 
-def _parse_injectable_params(func: Callable[..., t.Any]) -> tuple[list[tuple[str, t.Any]], dict[str, t.Any]]:
-    positional_or_keyword_params: list[tuple[str, t.Any]] = []
-    keyword_only_params: dict[str, t.Any] = {}
+def _parse_composed_dependencies(cls: type[compose.Compose]) -> dict[str, conditions.DependencyExpression[t.Any]]:
+    if (existing := getattr(cls, compose._DEPS_ATTR, None)) is not None:
+        return existing
+
+    actual_class = getattr(cls, compose._ACTUAL_ATTR, None)
+    if actual_class is None:
+        raise TypeError(f"class {cls} is not a composed dependency")
+
+    actual_class = t.cast("type[t.Any]", actual_class)
+    hints = t.get_type_hints(
+        actual_class, localns={m: sys.modules[m] for m in utils.ANNOTATION_PARSE_LOCAL_INCLUDE_MODULES}
+    )
+    return {
+        name: conditions.DependencyExpression.create(annotation)
+        for name, annotation in hints.items()
+        if name in getattr(cls, "__slots__")
+    }
+
+
+def _parse_injectable_params(
+    func: Callable[..., t.Any],
+) -> tuple[list[tuple[str, DependencyExprOrComposed]], dict[str, DependencyExprOrComposed]]:
+    positional_or_keyword_params: list[tuple[str, DependencyExprOrComposed]] = []
+    keyword_only_params: dict[str, DependencyExprOrComposed] = {}
 
     parameters = inspect.signature(
         func, locals={m: sys.modules[m] for m in utils.ANNOTATION_PARSE_LOCAL_INCLUDE_MODULES}, eval_str=True
     ).parameters
     for parameter in parameters.values():
+        annotation = parameter.annotation
         if (
             # If the parameter has no annotation
-            parameter.annotation is inspect.Parameter.empty
+            annotation is inspect.Parameter.empty
             # If the parameter is not positional-or-keyword or keyword-only
             or parameter.kind
             in (inspect.Parameter.POSITIONAL_ONLY, inspect.Parameter.VAR_POSITIONAL, inspect.Parameter.VAR_KEYWORD)
@@ -309,12 +334,17 @@ def _parse_injectable_params(func: Callable[..., t.Any]) -> tuple[list[tuple[str
                 positional_or_keyword_params.append((parameter.name, CANNOT_INJECT))
             continue
 
-        expr = conditions.DependencyExpression.create(parameter.annotation)
+        if compose._is_compose_class(annotation):
+            setattr(annotation, compose._DEPS_ATTR, _parse_composed_dependencies(annotation))
+
+        item = (
+            annotation if compose._is_compose_class(annotation) else conditions.DependencyExpression.create(annotation)
+        )
         if parameter.kind is inspect.Parameter.POSITIONAL_OR_KEYWORD:
-            positional_or_keyword_params.append((parameter.name, expr))
+            positional_or_keyword_params.append((parameter.name, item))
         else:
             # It has to be a keyword-only parameter
-            keyword_only_params[parameter.name] = expr
+            keyword_only_params[parameter.name] = item
 
     return positional_or_keyword_params, keyword_only_params
 
@@ -378,7 +408,7 @@ def _codegen_dependency_func(
     ) -> DependencyResolverFunctionT:
         pos_or_kw, kw_only = _parse_injectable_params(self._func)
 
-        exec_globals: dict[str, conditions.DependencyExpression[t.Any]] = {}
+        exec_globals: dict[str, DependencyExprOrComposed] = {}
 
         def gen_random_name() -> str:
             while True:
@@ -389,30 +419,46 @@ def gen_random_name() -> str:
             # this can never happen but pycharm is being stupid
             return ""
 
+        def resolver(dependency: DependencyExprOrComposed, refname: str) -> t.Any:
+            if not compose._is_compose_class(dependency):
+                return f"await {refname}.resolve(container)"
+
+            init_params: list[str] = []
+            subdeps = t.cast(
+                "dict[str, conditions.DependencyExpression[t.Any]]", getattr(dependency, compose._DEPS_ATTR)
+            )
+            for subdep_name, subdep in subdeps.items():
+                exec_globals[ident := gen_random_name()] = subdep
+                init_params.append(f"{subdep_name}=await {ident}.resolve(container)")
+
+            return f"{refname}({','.join(init_params)})"
+
         fn_lines = ["arglen = len(args)", "new_kwargs = {}; new_kwargs.update(kwargs)"]
 
         for i, tup in enumerate(pos_or_kw):
-            name, type_expr = tup
-            if type_expr is CANNOT_INJECT:
+            name, dep = tup
+            if dep is CANNOT_INJECT:
                 continue
 
-            exec_globals[n := gen_random_name()] = type_expr
+            exec_globals[n := gen_random_name()] = dep
+
             fn_lines.append(
-                f"if '{name}' not in new_kwargs and arglen < ({i + 1} - offset): new_kwargs['{name}'] = await {n}.resolve(container)"  # noqa: E501
+                f"if '{name}' not in new_kwargs and arglen < ({i + 1} - offset): new_kwargs['{name}'] = {resolver(dep, n)}"  # noqa: E501
             )
 
-        for name, type_expr in kw_only.items():
-            if type_expr is CANNOT_INJECT:
+        for name, dep in kw_only.items():
+            if dep is CANNOT_INJECT:
                 continue
 
-            exec_globals[n := gen_random_name()] = type_expr
-            fn_lines.append(f"if '{name}' not in new_kwargs: new_kwargs['{name}'] = await {n}.resolve(container)")
+            exec_globals[n := gen_random_name()] = dep
+            fn_lines.append(f"if '{name}' not in new_kwargs: new_kwargs['{name}'] = {resolver(dep, n)}")
 
         fn_lines.append("return new_kwargs")
 
-        fn = "async def resolve_dependencies(container, offset, args, kwargs):\n" + "\n".join(
+        fn = "async def resolve_dependencies(container,offset,args,kwargs):\n" + "\n".join(
             textwrap.indent(line, "    ") for line in fn_lines
         )
+
         exec(fn, exec_globals, (generated_locals := {}))
         return generated_locals["resolve_dependencies"]  # type: ignore[reportReturnType]
 

pyproject.toml

@@ -154,6 +154,11 @@ convention = "google"
 strict-imports = true
 require-superclass = true
 require-subclass = true
+exclude-classes = """
+(
+    ^linkd\\.compose:ComposeMeta$
+)
+"""
 
 [tool.pyright]
 include = ["linkd", "examples", "tests"]

tests/test_solver.py

@@ -69,11 +69,15 @@ def m(foo: str, bar: int = linkd.INJECTED, *, baz: float, bork: bool = linkd.INJ
         pos, kw = _parse_injectable_params(m)
 
         assert len(pos) == 2
+        assert isinstance(pos[0][1], linkd.DependencyExpression)
         assert pos[0][1]._order[0].inner is str and pos[0][1]._required
+        assert isinstance(pos[1][1], linkd.DependencyExpression)
         assert pos[1][1]._order[0].inner is int and pos[1][1]._required
 
         assert len(kw) == 2
+        assert isinstance(kw["baz"], linkd.DependencyExpression)
         assert kw["baz"]._order[0].inner is float and kw["baz"]._required
+        assert isinstance(kw["bork"], linkd.DependencyExpression)
         assert kw["bork"]._order[0].inner is bool and kw["bork"]._required