feat: add static evaluation of activators

Classify activators as STATIC/DYNAMIC and evaluate static ones
at container creation time, stripping inactive providers from
the dependency graph.

Author: fadedDexofan

src/dishka/activator_classifier.py

@@ -0,0 +1,183 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from enum import Enum, auto
+from typing import TYPE_CHECKING
+
+from dishka.entities.factory_type import FactoryType
+from dishka.entities.key import DependencyKey
+from dishka.entities.marker import Marker
+
+if TYPE_CHECKING:
+    from dishka.dependency_source.activator import Activator
+    from dishka.dependency_source.factory import Factory
+    from dishka.entities.scope import BaseScope
+    from dishka.registry import Registry
+
+
+class ActivatorType(Enum):
+    STATIC = auto()
+    DYNAMIC = auto()
+
+
+@dataclass(frozen=True, slots=True)
+class ClassifiedActivator:
+    key: DependencyKey
+    activator: Activator
+    type: ActivatorType
+    dependencies: frozenset[DependencyKey]
+
+
+class ActivatorClassifier:
+    def __init__(
+        self,
+        registries: tuple[Registry, ...],
+        activators: dict[DependencyKey, Activator],
+        root_scope: BaseScope,
+    ) -> None:
+        self._registries = registries
+        self._activators = activators
+        self._root_scope = root_scope
+        self._all_factories: dict[DependencyKey, Factory] = {}
+        self._context_keys_at_root: set[DependencyKey] = set()
+        self._build_factory_index()
+
+    def _build_factory_index(self) -> None:
+        for registry in self._registries:
+            for key, factory in registry.factories.items():
+                self._all_factories[key] = factory
+                if (
+                    factory.type == FactoryType.CONTEXT
+                    and factory.scope == self._root_scope
+                ):
+                    self._context_keys_at_root.add(key)
+
+    def _is_async_factory(self, factory: Factory) -> bool:
+        return factory.type in (
+            FactoryType.ASYNC_FACTORY,
+            FactoryType.ASYNC_GENERATOR,
+        )
+
+    def _is_marker_dependency(
+        self,
+        activator: Activator,
+        dep: DependencyKey,
+    ) -> bool:
+        """Check if dependency is the activator's marker (auto-injected)."""
+        return (
+            dep.type_hint is activator.marker_type
+            or dep.type_hint is Marker
+        )
+
+    def _get_activator_dependencies(
+        self,
+        activator: Activator,
+    ) -> frozenset[DependencyKey]:
+        factory = activator.factory
+        all_deps = list(factory.dependencies) + list(
+            factory.kw_dependencies.values(),
+        )
+        return frozenset(
+            dep for dep in all_deps
+            if dep in self._activators
+            and not self._is_marker_dependency(activator, dep)
+        )
+
+    def _get_all_dependencies(
+        self,
+        activator: Activator,
+    ) -> list[DependencyKey]:
+        factory = activator.factory
+        all_deps = list(factory.dependencies) + list(
+            factory.kw_dependencies.values(),
+        )
+        return [
+            dep for dep in all_deps
+            if not self._is_marker_dependency(activator, dep)
+        ]
+
+    def _is_root_context_dep(self, dep: DependencyKey) -> bool:
+        return dep in self._context_keys_at_root
+
+    def _is_registered(self, dep: DependencyKey) -> bool:
+        return dep in self._all_factories or dep in self._activators
+
+    def _topological_sort(
+        self,
+        activator_deps: dict[DependencyKey, frozenset[DependencyKey]],
+    ) -> list[DependencyKey]:
+        result: list[DependencyKey] = []
+        visited: set[DependencyKey] = set()
+
+        def visit(key: DependencyKey) -> None:
+            if key in visited:
+                return
+            for dep in activator_deps.get(key, frozenset()):
+                if dep in activator_deps:
+                    visit(dep)
+            visited.add(key)
+            result.append(key)
+
+        for key in activator_deps:
+            visit(key)
+
+        return result
+
+    def classify(self) -> dict[DependencyKey, ClassifiedActivator]:
+        activator_deps: dict[DependencyKey, frozenset[DependencyKey]] = {}
+        for key, activator in self._activators.items():
+            activator_deps[key] = self._get_activator_dependencies(activator)
+
+        eval_order = self._topological_sort(activator_deps)
+
+        classification: dict[DependencyKey, ClassifiedActivator] = {}
+
+        for key in eval_order:
+            activator = self._activators[key]
+            activator_type = self._classify_single(
+                activator,
+                activator_deps[key],
+                classification,
+            )
+            classification[key] = ClassifiedActivator(
+                key=key,
+                activator=activator,
+                type=activator_type,
+                dependencies=activator_deps[key],
+            )
+
+        return classification
+
+    def _classify_single(
+        self,
+        activator: Activator,
+        activator_dependencies: frozenset[DependencyKey],
+        already_classified: dict[DependencyKey, ClassifiedActivator],
+    ) -> ActivatorType:
+        factory = activator.factory
+
+        if self._is_async_factory(factory):
+            return ActivatorType.DYNAMIC
+
+        all_deps = self._get_all_dependencies(activator)
+
+        if not all_deps:
+            return ActivatorType.STATIC
+
+        for dep in activator_dependencies:
+            classified = already_classified.get(dep)
+            if classified and classified.type == ActivatorType.DYNAMIC:
+                return ActivatorType.DYNAMIC
+
+        non_activator_deps = [
+            dep for dep in all_deps if dep not in self._activators
+        ]
+
+        for dep in non_activator_deps:
+            if self._is_root_context_dep(dep):
+                continue
+            if not self._is_registered(dep):
+                continue
+            return ActivatorType.DYNAMIC
+
+        return ActivatorType.STATIC

src/dishka/async_container.py

@@ -30,6 +30,7 @@
 from .provider import BaseProvider, make_root_context_provider
 from .registry import Registry
 from .registry_builder import RegistryBuilder
+from .static_evaluator import apply_static_evaluation
 
 T = TypeVar("T")
 
@@ -307,14 +308,19 @@ def make_async_container(
 ) -> AsyncContainer:
     context_provider = make_root_context_provider(providers, context, scopes)
     has_provider = HasProvider()
-    registries = RegistryBuilder(
+    builder = RegistryBuilder(
         scopes=scopes,
         container_key=CONTAINER_KEY,
         multicomponent_providers=[has_provider],
         providers=(*providers, context_provider),
         skip_validation=skip_validation,
         validation_settings=validation_settings,
-    ).build()
+    )
+    registries = builder.build()
+    registries = apply_static_evaluation(
+        registries, builder.activators, context, start_scope,
+    )
+
     container = AsyncContainer(
         *registries,
         context=context,

src/dishka/container.py

@@ -30,6 +30,7 @@
 from .provider import BaseProvider, make_root_context_provider
 from .registry import Registry
 from .registry_builder import RegistryBuilder
+from .static_evaluator import apply_static_evaluation
 
 T = TypeVar("T")
 
@@ -304,14 +305,19 @@ def make_container(
 ) -> Container:
     context_provider = make_root_context_provider(providers, context, scopes)
     has_provider = HasProvider()
-    registries = RegistryBuilder(
+    builder = RegistryBuilder(
         scopes=scopes,
         container_key=CONTAINER_KEY,
         multicomponent_providers=[has_provider],
         providers=(*providers, context_provider),
         skip_validation=skip_validation,
         validation_settings=validation_settings,
-    ).build()
+    )
+    registries = builder.build()
+    registries = apply_static_evaluation(
+        registries, builder.activators, context, start_scope,
+    )
+
     container = Container(
         *registries,
         context=context,

src/dishka/registry_filter.py

@@ -0,0 +1,130 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from dishka.entities.component import DEFAULT_COMPONENT
+from dishka.entities.key import DependencyKey
+from dishka.entities.marker import (
+    AndMarker,
+    BaseMarker,
+    BoolMarker,
+    Marker,
+    NotMarker,
+    OrMarker,
+)
+from dishka.registry import Registry
+
+if TYPE_CHECKING:
+    from dishka.dependency_source.factory import Factory
+
+
+class RegistryFilter:
+    def __init__(
+        self,
+        activation_results: dict[DependencyKey, bool],
+    ) -> None:
+        self._activation_results = activation_results
+
+    def _eval_simple_marker(
+        self,
+        marker: Marker,
+        provides_key: DependencyKey | None,
+    ) -> bool | None:
+        key = DependencyKey(
+            marker,
+            provides_key.component if provides_key else DEFAULT_COMPONENT,
+        )
+        return self._activation_results.get(key)
+
+    def _eval_not_marker(
+        self,
+        marker: NotMarker,
+        provides_key: DependencyKey | None,
+    ) -> bool | None:
+        inner = self._is_marker_active(marker.marker, provides_key)
+        return None if inner is None else not inner
+
+    def _eval_or_marker(
+        self,
+        marker: OrMarker,
+        provides_key: DependencyKey | None,
+    ) -> bool | None:
+        left = self._is_marker_active(marker.left, provides_key)
+        right = self._is_marker_active(marker.right, provides_key)
+        if left is True or right is True:
+            return True
+        if left is None or right is None:
+            return None
+        return False
+
+    def _eval_and_marker(
+        self,
+        marker: AndMarker,
+        provides_key: DependencyKey | None,
+    ) -> bool | None:
+        left = self._is_marker_active(marker.left, provides_key)
+        right = self._is_marker_active(marker.right, provides_key)
+        if left is False or right is False:
+            return False
+        if left is None or right is None:
+            return None
+        return True
+
+    def _is_marker_active(
+        self,
+        marker: BaseMarker | None,
+        provides_key: DependencyKey | None,
+    ) -> bool | None:
+        result: bool | None
+        match marker:
+            case None:
+                result = True
+            case BoolMarker():
+                result = None if not marker.value else True
+            case NotMarker():
+                result = self._eval_not_marker(marker, provides_key)
+            case OrMarker():
+                result = self._eval_or_marker(marker, provides_key)
+            case AndMarker():
+                result = self._eval_and_marker(marker, provides_key)
+            case Marker():
+                result = self._eval_simple_marker(marker, provides_key)
+            case _:
+                result = None
+        return result
+
+    def _should_include_factory(self, factory: Factory) -> bool:
+        if factory.when_active is None:
+            return True
+
+        result = self._is_marker_active(
+            factory.when_active,
+            factory.provides,
+        )
+
+        if result is None:
+            return True
+        return result
+
+    def filter(
+        self,
+        registries: tuple[Registry, ...],
+    ) -> tuple[Registry, ...]:
+        if not self._activation_results:
+            return registries
+
+        filtered: list[Registry] = []
+
+        for registry in registries:
+            new_registry = Registry(
+                registry.scope,
+                has_fallback=registry.has_fallback,
+            )
+
+            for key, factory in registry.factories.items():
+                if self._should_include_factory(factory):
+                    new_registry.add_factory(factory, key)
+
+            filtered.append(new_registry)
+
+        return tuple(filtered)