refactor: Now analyze type hints with type-analyzer

Author: remimd

cq/_core/handler.py

@@ -3,11 +3,12 @@
 from collections.abc import Awaitable, Callable, Iterator
 from dataclasses import dataclass, field
 from functools import partial
-from inspect import Parameter, getmro, isclass
+from inspect import Parameter, isclass
 from inspect import signature as inspect_signature
 from typing import TYPE_CHECKING, Any, Protocol, Self, overload, runtime_checkable
 
 import injection
+from type_analyzer import MatchingTypesConfig, iter_matching_types, matching_types
 
 type HandlerType[**P, T] = type[Handler[P, T]]
 type HandlerFactory[**P, T] = Callable[..., Awaitable[Handler[P, T]]]
@@ -49,12 +50,14 @@ def handlers_from(
         self,
         input_type: type[I],
     ) -> Iterator[Callable[[I], Awaitable[O]]]:
-        for it in getmro(input_type):
-            for factory in self.__factories.get(it, ()):
+        for key_type in _iter_key_types(input_type):
+            for factory in self.__factories.get(key_type, ()):
                 yield _make_handle_function(factory)
 
     def subscribe(self, input_type: type[I], factory: HandlerFactory[[I], O]) -> Self:
-        self.__factories[input_type].append(factory)
+        for key_type in _build_key_types(input_type):
+            self.__factories[key_type].append(factory)
+
         return self
 
 
@@ -69,18 +72,20 @@ def handlers_from(
         self,
         input_type: type[I],
     ) -> Iterator[Callable[[I], Awaitable[O]]]:
-        for it in getmro(input_type):
-            factory = self.__factories.get(it, None)
+        for key_type in _iter_key_types(input_type):
+            factory = self.__factories.get(key_type, None)
             if factory is not None:
                 yield _make_handle_function(factory)
 
     def subscribe(self, input_type: type[I], factory: HandlerFactory[[I], O]) -> Self:
-        if input_type in self.__factories:
-            raise RuntimeError(
-                f"A handler is already registered for the input type: `{input_type}`."
-            )
+        for key_type in _build_key_types(input_type):
+            if key_type in self.__factories:
+                raise RuntimeError(
+                    f"A handler is already registered for the input type: `{key_type}`."
+                )
+
+            self.__factories[key_type] = factory
 
-        self.__factories[input_type] = factory
         return self
 
 
@@ -152,6 +157,20 @@ def __decorator(
         return wrapped
 
 
+def _build_key_types(input_type: Any) -> tuple[Any, ...]:
+    config = MatchingTypesConfig(ignore_none=True)
+    return matching_types(input_type, config)
+
+
+def _iter_key_types(input_type: Any) -> Iterator[Any]:
+    config = MatchingTypesConfig(
+        with_bases=True,
+        with_origin=True,
+        with_type_alias_value=True,
+    )
+    return iter_matching_types(input_type, config)
+
+
 def _resolve_input_type[I, O](handler_type: HandlerType[[I], O]) -> type[I]:
     fake_method = handler_type.handle.__get__(NotImplemented, handler_type)
     signature = inspect_signature(fake_method, eval_str=True)

pyproject.toml

@@ -50,6 +50,7 @@ classifiers = [
 dependencies = [
     "anyio",
     "python-injection",
+    "type-analyzer",
 ]
 
 [project.urls]