Merge pull request #6 from taskiq-python/feature/parameter-names

Author: s3rius

README.md

@@ -97,3 +97,31 @@ with graph.sync_ctx({DefaultDep: DefaultDep()}) as ctx:
 ```
 
 You can run this code. It will resolve dd dependency into a `DefaultDep` variable you provide.
+
+
+## Getting parameters information
+
+If you want to get the information about how this dependency was specified,
+you can use special class `ParamInfo` for that.
+
+```python
+from taskiq_dependencies import Depends, DependencyGraph, ParamInfo
+
+
+def dependency(info: ParamInfo = Depends()) -> str:
+    assert info.name == "dd"
+    return info.name
+
+def target_func(dd: str = Depends(dependency)):
+    print(dd)
+    return 1
+
+
+graph = DependencyGraph(target_func)
+
+with graph.sync_ctx() as ctx:
+    print(ctx.resolve_kwargs())
+
+```
+
+The ParamInfo has the information about name and parameters signature. It's useful if you want to create a dependency that changes based on parameter name, or signature.

taskiq_dependencies/__init__.py

@@ -8,5 +8,6 @@
 """
 from taskiq_dependencies.dependency import Depends
 from taskiq_dependencies.graph import DependencyGraph
+from taskiq_dependencies.utils import ParamInfo
 
-__all__ = ["DependencyGraph", "Depends"]
+__all__ = ["DependencyGraph", "Depends", "ParamInfo"]

taskiq_dependencies/ctx.py

@@ -3,6 +3,8 @@
 from copy import copy
 from typing import TYPE_CHECKING, Any, Dict, Generator, List, Optional
 
+from taskiq_dependencies.utils import ParamInfo
+
 if TYPE_CHECKING:
     from taskiq_dependencies.graph import DependencyGraph  # pragma: no cover
 
@@ -62,6 +64,11 @@ def traverse_deps(  # noqa: C901, WPS210
                 # we skip it.
                 if subdep.dependency is None:
                     continue
+                # If the user want to get ParamInfo,
+                # we get declaration of the current dependency.
+                if subdep.dependency == ParamInfo:
+                    kwargs[subdep.param_name] = ParamInfo(dep.param_name, dep.signature)
+                    continue
                 if subdep.use_cache:
                     # If this dependency can be calculated, using cache,
                     # we try to get it from cache.
@@ -82,7 +89,12 @@ def traverse_deps(  # noqa: C901, WPS210
 
             # We don't want to calculate least function,
             # Because it's a target function.
-            if index < len(self.graph.ordered_deps) - 1:
+            if (  # noqa: WPS337
+                index < len(self.graph.ordered_deps) - 1
+                # We skip all ParamInfo dependencies,
+                # because we calculate them when needed.
+                and dep.dependency != ParamInfo
+            ):
                 user_kwargs = dep.kwargs
                 user_kwargs.update(kwargs)
                 cache[dep.dependency] = yield dep.dependency(**user_kwargs)

taskiq_dependencies/dependency.py

@@ -1,3 +1,4 @@
+import inspect
 import uuid
 from typing import (  # noqa: WPS235
     Any,
@@ -107,12 +108,14 @@ def __init__(  # noqa: WPS234
         *,
         use_cache: bool = True,
         kwargs: Optional[Dict[str, Any]] = None,
+        signature: Optional[inspect.Parameter] = None,
     ) -> None:
         self._id = uuid.uuid4()
         self.dependency = dependency
         self.use_cache = use_cache
         self.param_name = ""
         self.kwargs = kwargs or {}
+        self.signature = signature
 
     def __hash__(self) -> int:
         return hash(self._id)

taskiq_dependencies/graph.py

@@ -119,6 +119,7 @@ def _build_graph(self) -> None:  # noqa: C901, WPS210
                     default_value = Dependency(
                         dependency=default_value.dependency,
                         use_cache=default_value.use_cache,
+                        signature=param,
                     )
 
                 # We check, that default value is an instance of
@@ -158,6 +159,7 @@ def _build_graph(self) -> None:  # noqa: C901, WPS210
                     dependency_func,
                     use_cache=default_value.use_cache,
                     kwargs=default_value.kwargs,
+                    signature=param,
                 )
                 # Also we set the parameter name,
                 # it will help us in future when

taskiq_dependencies/utils.py

@@ -0,0 +1,25 @@
+import inspect
+from typing import Optional
+
+
+class ParamInfo:
+    """
+    Parameter information.
+
+    This class helps you to get information,
+    about how the current dependency was specified.
+
+    If there's no dependant function, the name will be an empty string
+    and the definition will be None.
+    """
+
+    def __init__(
+        self,
+        name: str,
+        signature: Optional[inspect.Parameter] = None,
+    ) -> None:
+        self.name = name
+        self.definition = signature
+
+    def __repr__(self) -> str:
+        return f"ParamInfo<name={self.name}>"

tests/test_graph.py

@@ -4,7 +4,7 @@
 
 import pytest
 
-from taskiq_dependencies import DependencyGraph, Depends
+from taskiq_dependencies import DependencyGraph, Depends, ParamInfo
 
 
 @pytest.mark.anyio
@@ -258,3 +258,35 @@ def __init__(self, dep=Depends()) -> None:  # type: ignore
 
     with pytest.raises(ValueError):
         DependencyGraph(Target)
+
+
+def test_get_param_info() -> None:
+    """Tests that param info resolved correctly."""
+
+    def dep(info: ParamInfo = Depends()) -> ParamInfo:
+        return info
+
+    def target(my_test_param: ParamInfo = Depends(dep)) -> None:
+        return None
+
+    with DependencyGraph(target=target).sync_ctx() as g:
+        kwargs = g.resolve_kwargs()
+
+    info: ParamInfo = kwargs["my_test_param"]
+    assert info.name == "my_test_param"
+    assert info.definition
+    assert info.definition.annotation == ParamInfo
+
+
+def test_param_info_no_dependant() -> None:
+    """Tests that if ParamInfo is used on the target, no error is raised."""
+
+    def target(info: ParamInfo = Depends()) -> None:
+        return None
+
+    with DependencyGraph(target=target).sync_ctx() as g:
+        kwargs = g.resolve_kwargs()
+
+    info: ParamInfo = kwargs["info"]
+    assert info.name == ""
+    assert info.definition is None