support args and kwargs

Author: Lancetnik

fast_depends/__about__.py

@@ -1,3 +1,3 @@
 """FastDepends - extracted and cleared from HTTP domain FastAPI Dependency Injection System"""
 
-__version__ = "2.0.5"
+__version__ = "2.1.0"

fast_depends/core/build.py

@@ -1,5 +1,15 @@
 import inspect
-from typing import Any, Awaitable, Callable, Dict, Optional, Sequence, Tuple, Union
+from typing import (
+    Any,
+    Awaitable,
+    Callable,
+    Dict,
+    List,
+    Optional,
+    Sequence,
+    Tuple,
+    Union,
+)
 
 from typing_extensions import (
     Annotated,
@@ -49,7 +59,14 @@ def build_call_model(
     class_fields: Dict[str, Tuple[Any, Any]] = {}
     dependencies: Dict[str, CallModel[..., Any]] = {}
     custom_fields: Dict[str, CustomField] = {}
+    positional_args: List[str] = []
+    keyword_args: List[str] = []
     for param in typed_params:
+        if param.kind is param.KEYWORD_ONLY:
+            keyword_args.append(param.name)
+        elif param.name not in ("args", "kwargs"):
+            positional_args.append(param.name)
+
         dep: Optional[Depends] = None
         custom: Optional[CustomField] = None
 
@@ -82,7 +99,13 @@ def build_call_model(
         else:
             annotation = param.annotation
 
-        default = param.default
+        if param.name == "args":
+            default = ()
+        elif param.name == "kwargs":
+            default = {}
+        else:
+            default = param.default
+
         if isinstance(default, Depends):
             assert (
                 not dep
@@ -146,6 +169,8 @@ def build_call_model(
         is_async=is_call_async,
         dependencies=dependencies,
         custom_fields=custom_fields,
+        positional_args=positional_args,
+        keyword_args=keyword_args,
         extra_dependencies=[
             build_call_model(
                 d.dependency,

fast_depends/core/model.py

@@ -9,6 +9,7 @@
     Iterable,
     List,
     Optional,
+    Tuple,
     Type,
     Union,
 )
@@ -47,11 +48,30 @@ class CallModel(Generic[P, T]):
     dependencies: Dict[str, "CallModel[..., Any]"]
     extra_dependencies: Iterable["CallModel[..., Any]"]
     custom_fields: Dict[str, CustomField]
+    keyword_args: Tuple[str]
+    positional_args: Tuple[str]
 
     # Dependencies and custom fields
     use_cache: bool
     cast: bool
 
+    __slots__ = (
+        "call",
+        "is_async",
+        "is_generator",
+        "model",
+        "response_model",
+        "params",
+        "alias_arguments",
+        "keyword_args",
+        "positional_args",
+        "dependencies",
+        "extra_dependencies",
+        "custom_fields",
+        "use_cache",
+        "cast",
+    )
+
     @property
     def call_name(self) -> str:
         return getattr(self.call, "__name__", type(self.call).__name__)
@@ -83,6 +103,8 @@ def __init__(
         is_async: bool = False,
         dependencies: Optional[Dict[str, "CallModel[..., Any]"]] = None,
         extra_dependencies: Optional[Iterable["CallModel[..., Any]"]] = None,
+        keyword_args: Optional[List[str]] = None,
+        positional_args: Optional[List[str]] = None,
         custom_fields: Optional[Dict[str, CustomField]] = None,
     ):
         self.call = call
@@ -100,6 +122,8 @@ def __init__(
         self.custom_fields = custom_fields or {}
 
         self.alias_arguments = [f.alias or name for name, f in fields.items()]
+        self.keyword_args = tuple(keyword_args or [])
+        self.positional_args = tuple(positional_args or [])
 
         self.params = fields.copy()
         for name in self.dependencies.keys():
@@ -152,13 +176,20 @@ def _solve(
 
         casted_model = self.model(**solved_kw)
 
-        casted_kw = {
+        args_ = [
+            getattr(casted_model, arg, solved_kw.get(arg))
+            for arg in self.positional_args
+        ]
+        args_.extend(getattr(casted_model, "args", ()))
+
+        kwargs_ = {
             arg: getattr(casted_model, arg, solved_kw.get(arg))
-            for arg in (*self.params.keys(), *self.dependencies.keys())
+            for arg in self.keyword_args
         }
+        kwargs_.update(getattr(casted_model, "kwargs", {}))
 
         response: T
-        response = yield casted_kw
+        response = yield args_, kwargs_
 
         if self.cast is True and self.response_model is not None:
             casted_resp = self.response_model(response=response)
@@ -225,16 +256,17 @@ def solve(
         for custom in self.custom_fields.values():
             kwargs = custom.use(**kwargs)
 
-        final_kw = cast_gen.send(kwargs)
+        final_args, final_kwargs = cast_gen.send(kwargs)
 
         if self.is_generator:
             response = solve_generator_sync(
+                *final_args,
                 call=self.call,
                 stack=stack,
-                **final_kw,
+                **final_kwargs,
             )
         else:
-            response = self.call(**final_kw)
+            response = self.call(*final_args, **final_kwargs)
 
         try:
             cast_gen.send(response)
@@ -300,16 +332,17 @@ async def asolve(
         for custom in self.custom_fields.values():
             kwargs = await run_async(custom.use, **kwargs)
 
-        final_kw = cast_gen.send(kwargs)
+        final_args, final_kwargs = cast_gen.send(kwargs)
 
         if self.is_generator:
             response = await solve_generator_async(
+                *final_args,
                 call=self.call,
                 stack=stack,
-                **final_kw,
+                **final_kwargs,
             )
         else:
-            response = await run_async(self.call, **final_kw)
+            response = await run_async(self.call, *final_args, **final_kwargs)
 
         try:
             cast_gen.send(response)

fast_depends/utils.py

@@ -48,38 +48,53 @@ async def run_in_threadpool(
 
 
 async def solve_generator_async(
-    *, call: Callable[..., Any], stack: AsyncExitStack, **sub_values: Any
+    *sub_args: Any, call: Callable[..., Any], stack: AsyncExitStack, **sub_values: Any
 ) -> Any:
     if is_gen_callable(call):
         cm = contextmanager_in_threadpool(contextmanager(call)(**sub_values))
     elif is_async_gen_callable(call):  # pragma: no branch
-        cm = asynccontextmanager(call)(**sub_values)
+        cm = asynccontextmanager(call)(*sub_args, **sub_values)
     return await stack.enter_async_context(cm)
 
 
 def solve_generator_sync(
-    *, call: Callable[..., Any], stack: ExitStack, **sub_values: Any
+    *sub_args: Any, call: Callable[..., Any], stack: ExitStack, **sub_values: Any
 ) -> Any:
-    cm = contextmanager(call)(**sub_values)
+    cm = contextmanager(call)(*sub_args, **sub_values)
     return stack.enter_context(cm)
 
 
 def args_to_kwargs(
     arguments: Iterable[str], *args: Any, **kwargs: Any
 ) -> Dict[str, Any]:
+    arguments = tuple(filter(lambda i: i not in ("args", "kwargs"), arguments))
+
     if not args:
         return kwargs
 
-    unused = filter(lambda x: x not in kwargs, arguments)
+    merged = {"kwargs": kwargs.get("kwargs", {})}
+
+    for arg, v in kwargs.items():
+        if arg not in arguments:
+            merged["kwargs"][arg] = v
+        else:
+            merged[arg] = v
+
+    for arg in filter(lambda x: x not in merged, arguments):
+        if args:
+            merged[arg], args = args[0], args[1:]
 
-    return dict((*zip(unused, args), *kwargs.items()))
+    merged["args"] = args
+
+    return merged
 
 
 def get_typed_signature(
     call: Callable[..., Any]
 ) -> Tuple[List[inspect.Parameter], Any]:
     signature = inspect.signature(call)
     globalns = getattr(call, "__globals__", {})
+
     return [
         inspect.Parameter(
             name=param.name,

tests/async/test_cast.py

@@ -1,3 +1,5 @@
+from typing import Tuple, Dict
+
 import pytest
 from pydantic import BaseModel, Field, ValidationError
 from typing_extensions import Annotated
@@ -104,3 +106,50 @@ async def some_func(a: A) -> float:
         return a
 
     assert isinstance(await some_func(b="2"), float)
+
+
+@pytest.mark.asyncio
+async def test_args_kwargs_1():
+    @inject
+    async def simple_func(
+        a: int,
+        *args: Tuple[float, ...],
+        b: int,
+        **kwargs: Dict[str, int],
+    ):
+        return a, args, b, kwargs
+
+    assert (1, (2.0, 3.0), 3, {"key": 1}) == await simple_func(
+        1.0, 2.0, 3, b=3.0, key=1.0
+    )
+
+
+@pytest.mark.asyncio
+async def test_args_kwargs_2():
+    @inject
+    async def simple_func(
+        a: int,
+        *args: Tuple[float, ...],
+        b: int,
+    ):
+        return a, args, b
+
+    assert (1, (2.0, 3.0), 3) == await simple_func(
+        1.0,
+        2.0,
+        3,
+        b=3.0,
+    )
+
+
+@pytest.mark.asyncio
+async def test_args_kwargs_3():
+    @inject
+    async def simple_func(a: int, *, b: int):
+        return a, b
+
+    assert (1, 3) == await simple_func(
+        1.0,
+        4,
+        b=3.0,
+    )

tests/sync/test_cast.py

@@ -1,3 +1,5 @@
+from typing import Tuple, Dict
+
 import pytest
 from pydantic import BaseModel, Field, ValidationError
 from typing_extensions import Annotated
@@ -107,3 +109,45 @@ def some_func(a: A) -> float:
         return a
 
     assert isinstance(some_func(b="2"), float)
+
+
+def test_args_kwargs_1():
+    @inject
+    def simple_func(
+        a: int,
+        *args: Tuple[float, ...],
+        b: int,
+        **kwargs: Dict[str, int],
+    ):
+        return a, args, b, kwargs
+
+    assert (1, (2.0, 3.0), 3, {"key": 1}) == simple_func(1.0, 2.0, 3, b=3.0, key=1.0)
+
+
+def test_args_kwargs_2():
+    @inject
+    def simple_func(
+        a: int,
+        *args: Tuple[float, ...],
+        b: int,
+    ):
+        return a, args, b
+
+    assert (1, (2.0, 3.0), 3) == simple_func(
+        1.0,
+        2.0,
+        3,
+        b=3.0,
+    )
+
+
+def test_args_kwargs_3():
+    @inject
+    def simple_func(a: int, *, b: int):
+        return a, b
+
+    assert (1, 3) == simple_func(
+        1.0,
+        4,
+        b=3.0,
+    )