Merge pull request #29 from taskiq-python/feature/graph-dep

Author: s3rius

README.md

@@ -127,6 +127,8 @@ with graph.sync_ctx() as ctx:
 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.
 
 
+Also ParamInfo contains the initial graph that was used.
+
 ## Exception propagation
 
 By default if error happens within the context, we send this error to the dependency,

taskiq_dependencies/ctx.py

@@ -20,10 +20,13 @@ class BaseResolveContext:
     def __init__(
         self,
         graph: "DependencyGraph",
+        main_graph: "DependencyGraph",
         initial_cache: Optional[Dict[Any, Any]] = None,
         exception_propagation: bool = True,
     ) -> None:
         self.graph = graph
+        # Main graph that contains all the subgraphs.
+        self.main_graph = main_graph
         self.opened_dependencies: List[Any] = []
         self.sub_contexts: "List[Any]" = []
         self.initial_cache = initial_cache or {}
@@ -89,7 +92,11 @@ def traverse_deps(  # noqa: C901
                 # 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)
+                    kwargs[subdep.param_name] = ParamInfo(
+                        dep.param_name,
+                        self.main_graph,
+                        dep.signature,
+                    )
                     continue
                 if subdep.use_cache:
                     # If this dependency can be calculated, using cache,
@@ -197,7 +204,7 @@ def resolver(self, executed_func: Any, initial_cache: Dict[Any, Any]) -> Any:
         :return: dict with resolved kwargs.
         """
         if getattr(executed_func, "dep_graph", False):
-            ctx = SyncResolveContext(executed_func, initial_cache)
+            ctx = SyncResolveContext(executed_func, self.main_graph, initial_cache)
             self.sub_contexts.append(ctx)
             sub_result = ctx.resolve_kwargs()
         elif inspect.isgenerator(executed_func):
@@ -325,7 +332,7 @@ async def resolver(
         :return: dict with resolved kwargs.
         """
         if getattr(executed_func, "dep_graph", False):
-            ctx = AsyncResolveContext(executed_func, initial_cache)  # type: ignore
+            ctx = AsyncResolveContext(executed_func, self.main_graph, initial_cache)  # type: ignore
             self.sub_contexts.append(ctx)
             sub_result = await ctx.resolve_kwargs()
         elif inspect.isgenerator(executed_func):

taskiq_dependencies/graph.py

@@ -7,6 +7,7 @@
 
 from taskiq_dependencies.ctx import AsyncResolveContext, SyncResolveContext
 from taskiq_dependencies.dependency import Dependency
+from taskiq_dependencies.utils import ParamInfo
 
 try:
     from fastapi.params import Depends as FastapiDepends
@@ -63,6 +64,7 @@ def async_ctx(
         if replaced_deps:
             graph = DependencyGraph(self.target, replaced_deps)
         return AsyncResolveContext(
+            graph,
             graph,
             initial_cache,
             exception_propagation,
@@ -89,6 +91,7 @@ def sync_ctx(
         if replaced_deps:
             graph = DependencyGraph(self.target, replaced_deps)
         return SyncResolveContext(
+            graph,
             graph,
             initial_cache,
             exception_propagation,
@@ -122,8 +125,14 @@ def _build_graph(self) -> None:  # noqa: C901
                 continue
             if dep.dependency is None:
                 continue
+            # If we have replaced dependencies, we need to replace
+            # them in the current dependency.
             if self.replaced_deps and dep.dependency in self.replaced_deps:
                 dep.dependency = self.replaced_deps[dep.dependency]
+            # We can say for sure that ParamInfo doesn't have any dependencies,
+            # so we skip it.
+            if dep.dependency == ParamInfo:
+                continue
             # Get signature and type hints.
             origin = getattr(dep.dependency, "__origin__", None)
             if origin is None:

taskiq_dependencies/utils.py

@@ -1,13 +1,16 @@
 import inspect
 import sys
 from contextlib import _AsyncGeneratorContextManager, _GeneratorContextManager
-from typing import Any, AsyncContextManager, ContextManager, Optional
+from typing import TYPE_CHECKING, Any, AsyncContextManager, ContextManager, Optional
 
 if sys.version_info >= (3, 10):
     from typing import TypeGuard
 else:
     from typing_extensions import TypeGuard
 
+if TYPE_CHECKING:
+    from taskiq_dependencies.graph import DependencyGraph
+
 
 class ParamInfo:
     """
@@ -23,9 +26,11 @@ class ParamInfo:
     def __init__(
         self,
         name: str,
+        graph: "DependencyGraph",
         signature: Optional[inspect.Parameter] = None,
     ) -> None:
         self.name = name
+        self.graph = graph
         self.definition = signature
 
     def __repr__(self) -> str:

tests/test_graph.py

@@ -334,13 +334,15 @@ def dep(info: ParamInfo = Depends()) -> ParamInfo:
     def target(my_test_param: ParamInfo = Depends(dep)) -> None:
         return None
 
-    with DependencyGraph(target=target).sync_ctx() as g:
+    graph = DependencyGraph(target=target)
+    with graph.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
+    assert info.graph == graph
 
 
 def test_param_info_no_dependant() -> None:
@@ -349,12 +351,14 @@ def test_param_info_no_dependant() -> None:
     def target(info: ParamInfo = Depends()) -> None:
         return None
 
-    with DependencyGraph(target=target).sync_ctx() as g:
+    graph = DependencyGraph(target=target)
+    with graph.sync_ctx() as g:
         kwargs = g.resolve_kwargs()
 
     info: ParamInfo = kwargs["info"]
     assert info.name == ""
     assert info.definition is None
+    assert info.graph == graph
 
 
 def test_class_based_dependencies() -> None:
@@ -863,3 +867,27 @@ def target(acm: TestACM = Depends(get_test_acm)) -> None:
         kwargs = await ctx.resolve_kwargs()
         assert kwargs["acm"] == test_acm
         assert not test_acm.opened
+
+
+def test_param_info_subgraph() -> None:
+    """
+    Test subgraphs for ParamInfo.
+
+    Test that correct graph is stored in ParamInfo
+    even if evaluated from subgraphs.
+    """
+
+    def inner_dep(info: ParamInfo = Depends()) -> ParamInfo:
+        return info
+
+    def target(info: ParamInfo = Depends(inner_dep, use_cache=False)) -> None:
+        return None
+
+    graph = DependencyGraph(target=target)
+    with graph.sync_ctx() as g:
+        kwargs = g.resolve_kwargs()
+
+    info: ParamInfo = kwargs["info"]
+    assert info.name == ""
+    assert info.definition is None
+    assert info.graph == graph