Separate inline provider discovery from scope propagation

modern_di/registries/providers_registry.py

@@ -219,34 +219,44 @@ def _inline_children(provider: AbstractProvider[typing.Any]) -> list[AbstractPro
     return [value for value in values if isinstance(value, AbstractProvider) and value._declared_as is None]  # noqa: SLF001
 
 
-def _stamp_inline_scopes(registered: tuple[AbstractProvider[typing.Any], ...]) -> None:
-    """Give each scope-defaulted inline provider the shallowest scope among the providers using it.
-
-    A scope set by an earlier registration is kept.
-    """
+def _discover_inline_graph(
+    registered: tuple[AbstractProvider[typing.Any], ...],
+) -> tuple[dict[int, list[AbstractProvider[typing.Any]]], dict[int, list[int]]]:
+    """Find inline children and record their parents in discovery order."""
     roots = {id(provider) for provider in registered}
     children_of: dict[int, list[AbstractProvider[typing.Any]]] = {}
-    parents_of: dict[int, list[AbstractProvider[typing.Any]]] = {}
+    parents_of: dict[int, list[int]] = {}
     stack = list(registered)
     while stack:
         parent = stack.pop()
-        if id(parent) in children_of:
+        parent_id = id(parent)
+        if parent_id in children_of:
             continue
         children = [child for child in _inline_children(parent) if id(child) not in roots]
-        children_of[id(parent)] = children
+        children_of[parent_id] = children
         for child in children:
-            parents_of.setdefault(id(child), []).append(parent)
+            parents_of.setdefault(id(child), []).append(parent_id)
             stack.append(child)
+    return children_of, parents_of
+
+
+def _stamp_inline_scopes(registered: tuple[AbstractProvider[typing.Any], ...]) -> None:
+    """Give each scope-defaulted inline provider the shallowest scope among the providers using it.
+
+    A scope set by an earlier registration is kept.
+    """
+    children_of, parents_of = _discover_inline_graph(registered)
     waiting = {child_id: len(parents) for child_id, parents in parents_of.items()}
     passed_down = {id(provider): provider.scope for provider in registered}
     ready = list(registered)
     while ready:
         parent = ready.pop()
         for child in children_of[id(parent)]:
-            waiting[id(child)] -= 1
-            if waiting[id(child)]:
+            child_id = id(child)
+            waiting[child_id] -= 1
+            if waiting[child_id]:
                 continue
-            scope = min(passed_down[id(p)] for p in parents_of[id(child)])
+            scope = min(passed_down[parent_id] for parent_id in parents_of[child_id])
             child._stamp_inline_scope(scope)  # noqa: SLF001
-            passed_down[id(child)] = scope if type(child) is _Derived else child.scope
+            passed_down[child_id] = scope if type(child) is _Derived else child.scope
             ready.append(child)

tests/test_inline_scope.py

@@ -1,4 +1,5 @@
 import dataclasses
+import enum
 
 import pytest
 
@@ -96,6 +97,24 @@ class Dependencies(Group):
         assert child.resolve(_DbConnection).dsn == child.resolve(_Engine).dsn == "postgres://"
 
 
+def test_shared_inline_provider_breaks_scope_ties_in_parent_discovery_order() -> None:
+    class CustomScope(enum.IntEnum):
+        REQUEST = 3
+
+    leaf = providers.Factory(object)
+    nested = providers.Factory(lambda leaf: leaf, scope=CustomScope.REQUEST, kwargs={"leaf": leaf})
+
+    class Dependencies(Group):
+        direct = providers.Factory(
+            lambda leaf, nested: (leaf, nested), scope=Scope.REQUEST, kwargs={"leaf": leaf, "nested": nested}
+        )
+        indirect = providers.Factory(lambda nested: [nested], scope=Scope.REQUEST, kwargs={"nested": nested})
+
+    Container(groups=[Dependencies])
+
+    assert leaf.scope is CustomScope.REQUEST
+
+
 def test_inline_provider_scope_is_fixed_at_first_registration() -> None:
     dsn = providers.Factory(lambda: "postgres://")