Repository: reagento/dishka
Pull request #195 (2024-08-02 20:39:49)

#98 Add 'WithParents'


Patch:
diff --git a/.ruff.toml b/.ruff.toml
index 1ce5965..dab03dc 100644
--- a/.ruff.toml
+++ b/.ruff.toml
@@ -24,6 +24,7 @@ lint.ignore = [
    "PLR0913",
    "UP038",
    "TCH001",
+   "SIM103",
 ]
 
 [lint.per-file-ignores]
diff --git a/docs/provider/provide.rst b/docs/provider/provide.rst
index 827a820..81b8df4 100644
--- a/docs/provider/provide.rst
+++ b/docs/provider/provide.rst
@@ -113,4 +113,28 @@ Also, if an error occurs during process handling (inside the ``with`` block), it
         def p(self) -> AnyOf[A, AProtocol]:
             return A()
 
-It works similar to :ref:`alias`.
\ No newline at end of file
+It works similar to :ref:`alias`.
+
+* Do you want to get dependencies by parents? Use ``WithParents`` as a result hint:
+
+.. code-block:: python
+
+    from dishka import WithParents, provide, Provider, Scope
+
+    class A(Protocol): ...
+    class AImpl(A): ...
+
+    class MyProvider(Provider):
+        scope=Scope.APP
+
+        @provide
+        def a(self) -> WithParents[AImpl]:
+            return A()
+
+    container = make_async_container(MyProvider())
+    a = await container.get(A)
+    a = await container.get(AImpl)
+    a is a # True
+
+
+WithParents generates only one factory and many aliases and is equivalent to ``AnyOf[AImpl, A]``. The following parents are ignored: ``type``, ``object``, ``Enum``, ``ABC``, ``ABCMeta``, ``Generic``, ``Protocol``, ``Exception``, ``BaseException``
diff --git a/pyproject.toml b/pyproject.toml
index 9a06ec2..aac556b 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -37,5 +37,3 @@ dependencies = [
 "Homepage" = "https://github.com/reagento/dishka"
 "Documentation" = "https://dishka.readthedocs.io/en/stable/"
 "Bug Tracker" = "https://github.com/reagento/dishka/issues"
-
-
diff --git a/src/dishka/__init__.py b/src/dishka/__init__.py
index 6d5f887..3dc2052 100644
--- a/src/dishka/__init__.py
+++ b/src/dishka/__init__.py
@@ -7,7 +7,7 @@ __all__ = [
     "alias", "decorate", "from_context", "provide", "provide_all",
     "DependencyKey",
     "FromDishka",
-    "BaseScope", "Scope", "new_scope",
+    "BaseScope", "Scope", "new_scope", "WithParents",
 ]
 
 from .async_container import AsyncContainer, make_async_container
@@ -24,4 +24,5 @@ from .entities.depends_marker import FromDishka
 from .entities.key import DependencyKey, FromComponent
 from .entities.provides_marker import AnyOf
 from .entities.scope import BaseScope, Scope, new_scope
+from .entities.with_parents import WithParents
 from .provider import Provider
diff --git a/src/dishka/entities/with_parents.py b/src/dishka/entities/with_parents.py
new file mode 100644
index 0000000..b4c5a7d
--- /dev/null
+++ b/src/dishka/entities/with_parents.py
@@ -0,0 +1,185 @@
+from abc import ABC, ABCMeta
+from enum import Enum
+from types import GenericAlias
+from typing import (
+    TYPE_CHECKING,
+    Final,
+    Generic,
+    Protocol,
+    TypeAlias,
+    TypeVar,
+)
+
+from dishka._adaptix.common import TypeHint
+from dishka._adaptix.feature_requirement import HAS_PY_311
+from dishka._adaptix.type_tools import (
+    get_generic_args,
+    get_type_vars,
+    is_generic,
+    strip_alias,
+)
+from dishka.entities.provides_marker import ProvideMultiple
+
+IGNORE_TYPES: Final = (
+    type,
+    object,
+    Enum,
+    ABC,
+    ABCMeta,
+    Generic,
+    Protocol,
+    Exception,
+    BaseException,
+)
+TypeVarsMap: TypeAlias = dict[TypeHint, TypeHint]
+
+
+def has_orig_bases(obj: TypeHint) -> bool:
+    return hasattr(obj, "__orig_bases__")
+
+if HAS_PY_311:
+    def is_type_var_tuple(obj: TypeHint) -> bool:
+        return getattr(obj, "__typing_is_unpacked_typevartuple__", False)
+else:
+    def is_type_var_tuple(obj: TypeHint) -> bool:
+        return False
+
+def is_ignored_type(origin_obj: TypeHint) -> bool:
+    return origin_obj in IGNORE_TYPES
+
+
+def get_filled_arguments(obj: TypeHint) -> list[TypeHint]:
+    filled_arguments = []
+    for arg in get_generic_args(obj):
+        if isinstance(arg, TypeVar):
+            continue
+        if is_type_var_tuple(arg):
+           continue
+        filled_arguments.append(arg)
+    return filled_arguments
+
+
+def create_type_vars_map(obj: TypeHint) -> TypeVarsMap:
+    origin_obj = strip_alias(obj)
+    if not get_type_vars(origin_obj):
+        return {}
+
+    type_vars = list(get_type_vars(origin_obj))
+    filled_arguments = get_filled_arguments(obj)
+
+    if not filled_arguments or not type_vars:
+        return {}
+
+    type_vars_map = {}
+    reversed_arguments = False
+    while True:
+        if len(type_vars) == 0:
+            break
+        type_var = type_vars[0]
+        if isinstance(type_var, TypeVar):
+            del type_vars[0]
+            type_vars_map[type_var] = filled_arguments.pop(0)
+        else:
+            if len(type_vars) == 1:
+                if reversed_arguments:
+                    filled_arguments.reverse()
+                type_vars_map[type_var] = filled_arguments
+                break
+            type_vars.reverse()
+            filled_arguments.reverse()
+            reversed_arguments = not reversed_arguments
+
+    return type_vars_map
+
+
+def create_generic_class(
+    origin_obj: TypeHint,
+    type_vars_map: TypeVarsMap,
+) -> TypeHint | None:
+    if not is_generic(origin_obj):
+        return None
+
+    generic_args = []
+    for type_var in get_type_vars(origin_obj):
+        arg = type_vars_map[type_var]
+        if isinstance(arg, list):
+            generic_args.extend(arg)
+        else:
+            generic_args.append(arg)
+    return origin_obj[tuple(generic_args)]
+
+
+def recursion_get_parents_for_generic_class(
+    obj: TypeHint,
+    parents: list[TypeHint],
+    type_vars_map: TypeVarsMap,
+) -> None:
+    origin_obj = strip_alias(obj)
+    if is_ignored_type(origin_obj):
+        return
+
+    if not has_orig_bases(origin_obj):
+        parents.extend(get_parents_for_mro(origin_obj))
+        return
+
+    for obj_ in origin_obj.__orig_bases__:
+        origin_obj = strip_alias(obj_)
+        if is_ignored_type(origin_obj):
+            continue
+
+        type_vars_map.update(create_type_vars_map(obj_))
+        parents.append(create_generic_class(origin_obj, type_vars_map) or obj_)
+        recursion_get_parents_for_generic_class(
+            obj_,
+            parents,
+            type_vars_map.copy(),
+        )
+
+
+def get_parents_for_mro(obj: TypeHint) -> list[TypeHint]:
+    return [
+        obj_ for obj_ in obj.mro()
+        if not is_ignored_type(strip_alias(obj_))
+    ]
+
+
+def get_parents(obj: TypeHint) -> list[TypeHint]:
+    if is_ignored_type(strip_alias(obj)):
+        raise ValueError(f"The starting class {obj!r} is in ignored types")
+
+    if isinstance(obj, GenericAlias):
+        type_vars_map = create_type_vars_map(obj)
+        parents = [
+            create_generic_class(
+                origin_obj=strip_alias(obj),
+                type_vars_map=type_vars_map,
+            ) or obj,
+        ]
+        recursion_get_parents_for_generic_class(
+            obj=obj,
+            parents=parents,
+            type_vars_map=type_vars_map,
+        )
+    elif has_orig_bases(obj):
+        parents = [obj]
+        recursion_get_parents_for_generic_class(
+            obj=obj,
+            parents=parents,
+            type_vars_map={},
+        )
+    else:
+        parents = get_parents_for_mro(obj)
+    return parents
+
+
+if TYPE_CHECKING:
+    from typing import Union as WithParents
+else:
+    class WithParents:
+        def __class_getitem__(
+            cls, item: TypeHint,
+        ) -> TypeHint | ProvideMultiple:
+            parents = get_parents(item)
+            if len(parents) > 1:
+                return ProvideMultiple(parents)
+            return parents[0]


Test patch:
diff --git a/tests/unit/container/test_with_parents.py b/tests/unit/container/test_with_parents.py
new file mode 100644
index 0000000..abb3e40
--- /dev/null
+++ b/tests/unit/container/test_with_parents.py
@@ -0,0 +1,161 @@
+from abc import ABC
+from typing import Annotated, Any, Generic, Protocol, TypeVar
+
+import pytest
+
+from dishka import (
+    DEFAULT_COMPONENT,
+    FromComponent,
+    Provider,
+    Scope,
+    make_container,
+)
+from dishka._adaptix.feature_requirement import HAS_PY_311
+from dishka.entities.with_parents import WithParents
+from dishka.exceptions import NoFactoryError
+
+if HAS_PY_311:
+    from typing import TypeVarTuple, Unpack
+
+    Ts = TypeVarTuple("TS")  # noqa: PLC0132
+
+T = TypeVar("T")
+B = TypeVar("B")
+
+
+def test_simple_inheritance() -> None:
+    class A1: ...
+    class A2(A1): ...
+    class A3(A2): ...
+
+    provider = Provider(scope=Scope.APP)
+    provider.provide(lambda: A3(), provides=WithParents[A3])
+    container = make_container(provider)
+    assert (
+        container.get(A3)
+        is container.get(A2)
+        is container.get(A1)
+    )
+
+
+@pytest.mark.parametrize(
+    ("obj", "value", "component"),
+    [
+        (type("A1", (Protocol,), {}), Protocol, DEFAULT_COMPONENT),
+        (type("A2", (object,), {}), object, DEFAULT_COMPONENT),
+        (type("A3", (ABC,), {}), ABC, DEFAULT_COMPONENT),
+        (type("A3", (ABC,), {}), ABC, "TTT"),
+    ],
+)
+def test_ignore_parent_type(obj: Any, value: Any, component: Any) -> None:
+    provider = Provider(scope=Scope.APP, component=component)
+    provider.provide(
+        source=lambda: obj(),
+        provides=Annotated[WithParents[obj], FromComponent(component)],
+    )
+    container = make_container(provider)
+
+    with pytest.raises(NoFactoryError):
+        container.get(value, component=component)
+
+
+def test_type_var() -> None:
+    class A1(Generic[T]): ...
+    class A2(A1[str]): ...
+
+    provider = Provider(scope=Scope.APP)
+    provider.provide(lambda: A2(), provides=WithParents[A2])
+
+    container = make_container(provider)
+
+    assert(
+        container.get(A2)
+        is container.get(A1[str])
+    )
+
+
+@pytest.mark.skipif(
+    not HAS_PY_311,
+    reason="test for python >= 3.11",
+)
+def test_type_var_tuple() -> None:
+    class A1(Generic[Unpack[Ts]]): ...
+    class A2(A1[str, int, type]): ...
+
+    provider = Provider(scope=Scope.APP)
+    provider.provide(lambda: A2(), provides=WithParents[A2])
+
+    container = make_container(provider)
+
+    assert (
+        container.get(A2)
+        is container.get(A1[str, int, type])
+    )
+
+if HAS_PY_311:
+    class A1(Generic[T, Unpack[Ts]]): ...
+    class A2(A1[str, int, type], int): ...
+
+    class B1(Generic[Unpack[Ts], T], int): ...
+    class B2(B1[int, tuple[str, ...], type], int): ...
+
+    class C1(Generic[B, Unpack[Ts], T]): ...
+    class C2(C1[int, type, str, tuple[str, ...]], int): ...
+
+    params = [
+        (A2, A1[str, int, type], int),
+        (B2, B1[int, tuple[str, ...], type], int),
+        (C2, C1[int, type, str, tuple[str, ...]], int),
+    ]
+else:
+    params = []
+
+@pytest.mark.parametrize(
+    ("obj", "val1", "val2"),
+    params,
+)
+@pytest.mark.skipif(
+    not HAS_PY_311,
+reason="test for python >= 3.11",
+)
+def test_type_var_and_type_var_tuple(
+    obj: Any,
+    val1: Any,
+    val2: Any,
+) -> None:
+    provider = Provider(scope=Scope.APP)
+    provider.provide(lambda: obj(), provides=WithParents[obj])
+    container = make_container(provider)
+
+    assert (
+        container.get(obj)
+        is container.get(val1)
+        is container.get(val2)
+    )
+
+def test_deep_inheritance() -> None:
+    class A1(Generic[T], float): ...
+    class A2(A1[T], Generic[T]): ...
+
+    class B1: ...
+    class B2(B1): ...
+    class B3(B2): ...
+
+    class C1(Generic[T], B3): ...
+    class D1(A2[int], C1[str]): ...
+
+    provider = Provider(scope=Scope.APP)
+    provider.provide(lambda: D1(), provides=WithParents[D1])
+    container = make_container(provider)
+
+    assert(
+        container.get(D1)
+        is container.get(A2[int])
+        is container.get(A1[int])
+        is container.get(float)
+        is container.get(C1[str])
+        is container.get(B3)
+        is container.get(B2)
+        is container.get(B1)
+        is container.get(D1)
+    )