fix: Has(T) registers type to dishka for when= usage

[Stefanqn]

currently conditional activation is documented with from_context(RedisConfig, scope=Scope.APP) to register the type to dishka.

This forces a user to provide the type via context and blocks it to provide it with a provider.

Detailed Bug description is:

Title

@provide(when=Has(T)) still fails graph validation when T is not registered

Body

Hi, I think there is a mismatch between the documented behavior of Has(...) and actual behavior in dishka==1.9.1.

Expected

A factory guarded with @provide(when=Has(T)) should be inactive when T is absent, but container creation should still succeed.

Actual

Container creation fails with GraphMissingFactoryError if T is not registered in the graph at all, even though the factory is guarded by Has(T).

Minimal reproducer

from dishka import Has, Provider, Scope, make_container, provide
import pytest


def test_has_guard_does_not_allow_missing_dependency() -> None:
    class SecurityConfig:
        pass

    class AppStartSideEffect:
        pass

    class SecurityProvider(Provider):
        scope = Scope.APP

        @provide(when=Has(SecurityConfig))
        def cors_setup(self, cfg: SecurityConfig) -> AppStartSideEffect:
            return AppStartSideEffect()

    with pytest.raises(Exception) as exc_info:
        make_container(SecurityProvider())

    assert exc_info.type.__name__ == "GraphMissingFactoryError"
    assert "SecurityConfig" in str(exc_info.value)

Observed behavior

The factory parameter cfg: SecurityConfig is validated unconditionally during graph build, so Has(SecurityConfig) does not make that dependency optional.

Workaround

If SecurityConfig is registered first, for example via from_context(SecurityConfig, scope=Scope.APP), then the guarded factory behaves as expected.

from dishka import Has, Provider, Scope, from_context, make_container, provide, collect


def test_has_guard_works_once_type_is_registered() -> None:
    class SecurityConfig:
        pass

    class AppStartSideEffect:
        pass

    class SecurityProvider(Provider):
        scope = Scope.APP
        security_config = from_context(SecurityConfig, scope=Scope.APP)

        @provide(when=Has(SecurityConfig))
        def cors_setup(self, cfg: SecurityConfig) -> AppStartSideEffect:
            return AppStartSideEffect()

        all_side_effects = collect(AppStartSideEffect)

    container = make_container(SecurityProvider(), context={})
    assert container.get(list[AppStartSideEffect]) == []

Why this is surprising

The docs for conditional activation suggest that Has(T) can be used to guard factories based on whether a type is available. That reads like "factory becomes inactive if T is absent," but current behavior is "T must still already be registered in the graph."

[Tishka17]

According to your description this is definitely a bug. But I cannot get what is your PR about. Please, do not reformat unrelated code

[Stefanqn]

Please, do not reformat unrelated code

ok, I manually reverted that.

But I cannot get what is your PR about.

please look at the tests. I added a provider.declare(T) method to declare a type without providing an instance. This facilitates using Conditional activation when an instance of T is absent without limiting the wiring of T to come from the context, like the from_context examples in the documentation.

Also I updated the documentation:

Use declare(T, ...) when you only need to register the type in graph without
adding a real source for it. This helps graph validation, but Has(T) still
stays false until some real provider or context value is available.

[Tishka17]

Ok, looks interesting. I am not yet sure we want extend API here, probably we can just fix the bug, but I'll keep this PR open for a while

[Tishka17]

Relates to #642

[Stefanqn]

I am not yet sure we want extend API here, probably we can just fix the bug

please check the latest commit. It fixes the bug by making Has(T) registration implicit.

In short: Has(T) now implicitly “declares” T for validation, but does not pretend that T exists at runtime.

It does it in two separate layers:

• graph-building/validation gets a synthetic factory for T
• runtime activation for Has(T) still evaluates to false unless T is actually obtainable

Mechanically:

• In src/dishka/graph_builder/builder.py, build() now calls _add_implicit_has_factories(...) before scope fixing and validation.
• _add_implicit_has_factories(...) walks all marker expressions on factories, including nested when_dependencies, via _iter_factory_markers(...).
• For every Has(T) marker it finds, if no factory already exists for T, it creates one with _make_implicit_has_factory(...).
• That synthetic factory is registered under DependencyKey(T, component), so validator lookup can now resolve T and graph validation no longer raises “missing factory”.

So the hidden factory is:

• visible enough for validation
• permanently inactive for runtime Has(T) checks

[Tishka17]

Yeah, that make sense. Probably we can use it as a temporary solution. I'd prefer more general solution but it will take some time, I'll consider merging this PR and then reworking in more unified way

[Stefanqn]

I'll consider merging this PR and then reworking in more unified way

this would be nice - this bug strongly limits conditional activation for my use case. I've implemented a generic ConfigProvider for traversing a config tree and automatically provide decorated sub-sections.

[chirizxc]

.. _when:

Conditional activation
============================================

There are some cases when you want to declare a factory or decorator in a provider but use only when a certain condition is met. For example:

* Apply decorators in debug mode
* Use cache if redis config provided in context
* Implement A/B testing with different implementations based on HTTP header
* Provide different identity provider classes based on available context objects: web request or queued messages.

This can be achieved with "activation" approach. Key concepts here:

* **Marker** - special object to distinguish which implementations should be used.
* **Activator** or **activation function** - special function registered in provider and taking decision if marker is active or not.
* **activation condition** - expression with marker objects set in dependency source dynamically associated with activators to select between multiple implementations or enable decorators


.. note::

    The activation feature makes the application harder to analyze and can also affect performance, so use it wisely.

Basic usage
---------------------------------

To set conditional activation you create special ``Marker`` objects and use them in ``when=`` condition inside ``provide``, ``decorate`` or ``alias``.

.. code-block:: python

-    from dishka import Provider, provide, Scope
+    from dishka import Marker, Provider, Scope, provide

-    class MyProvider(Provider)
+    class MyProvider(Provider):
        @provide(scope=Scope.APP)
        def base_impl(self) -> Cache:
            return NormalCacheImpl()

        @provide(when=Marker("debug"), scope=Scope.APP)
        def debug_impl(self) -> Cache:
            return DebugCacheImpl()

In this code you can see 2 factories providing same type ``Cache``.
The second one is used whenever ``Marker("debug")`` is treated as as active.
The base implementation will be used in all other cases as it has no condition set.
The overall rule is "last wins" like it worked with overriding.

Second step is to provide logic of marker activation. You write a function returning ``bool`` and register it in provider using ``@activate`` decorator.
It can be the same or another provider while you pass when creating a container.

.. code-block:: python

-    from dishka import activate, Provider
+    from dishka import Marker, Provider, activate

-    class MyProvider(Provider)
+    class MyProvider(Provider):
        @activate(Marker("debug"))
        def is_debug(self) -> bool:
            return False

This function can use other objects as well. For example, we can pass config using context

.. code-block:: python

-    class MyProvider(Provider)
+    class MyProvider(Provider):
        config = from_context(Config, scope=Scope.APP)

        @activate(Marker("debug"))
        def is_debug(self, config: Config) -> bool:
            return config.debug

Activation on marker type
--------------------------------

More general pattern is to create own marker type and register a single activator on all instances. You can request marker as an activator parameter.

.. code-block::

    class EnvMarker(Marker):
        pass

-    class MyProvider(Provider)
+    class MyProvider(Provider):
        config = from_context(Config, scope=Scope.APP)

        @activate(EnvMarker)
        def is_debug(self, marker: EnvMarker, config: Config) -> bool:
            return config.environment == marker.value


Combining markers
------------------------------------------

Markers support simple combination logic when used in ``when=`` using ``|`` (or), ``&`` (and) and ``~`` (not) operators

.. code-block:: python


        @provide(when=Marker("debug") | EnvMarker("preprod"))
        def debug_impl(self) -> Cache:
            return DebugCacheImpl()

        @provide(when=~Marker("debug") & EnvMarker("preprod"))
        def test_impl(self) -> Cache:
            return TestCacheImpl()


Provider-level activation
-------------------------

You can set ``when=`` on the entire provider to apply a condition to all factories, aliases, and decorators within it. This reduces boilerplate when all dependencies in a provider share the same activation condition.

.. code-block:: python

    from dishka import Marker, Provider, Scope, provide

    class DebugProvider(Provider):
        when = Marker("debug")
        scope = Scope.APP

        @provide
        def debug_cache(self) -> Cache:
            return DebugCacheImpl()

        @provide
        def debug_logger(self) -> Logger:
            return VerboseLogger()

The provider's ``when`` can also be set via constructor:

.. code-block:: python

    provider = DebugProvider(when=Marker("debug"))

When both provider and individual source have ``when=``, conditions are combined with AND logic:

.. code-block:: python

    class FeatureProvider(Provider):
        when = Marker("prod")  # prerequisite
        scope = Scope.APP

        @provide(when=Has(RedisConfig))  # additional condition
        def redis_cache(self, config: RedisConfig) -> Cache:
            return RedisCache(config)
        # Effective: Marker("prod") & Has(RedisConfig)

The provider's ``when`` acts as a prerequisite; individual sources add further constraints. If a factory shouldn't inherit the provider's condition, move it to a different provider.

Checking graph elements
---------------------------------------

In case you want to activate some features when specific objects are available you can use ``Has`` marker. It checks whether

* requested class is registered in container with appropriate scope
* it is activated
* if it actually presents in context while being registered as ``from_context``

``Has(T)`` implicitly registers ``T`` for graph validation.

Use ``from_context(T, ...)`` when ``Has(T)`` should become true only after a real
context value is passed.

The implicit registration only helps validation. ``Has(T)`` still stays false
until some real provider or real context value is available.


For example:

.. code-block:: python

-    from dishka import Provider, from_context, provide, Scope
+    from dishka import Has, Provider, Scope, from_context, make_container, provide

-    class MyProvider(Provider)
+    class MyProvider(Provider):
        config = from_context(RedisConfig, scope=Scope.APP)

        @provide(scope=Scope.APP)
        def base_impl(self) -> Cache:
            return NormalCacheImpl()

        @provide(when=Has(RedisConfig), scope=Scope.APP)
        def redis_impl(self, config: RedisConfig) -> Cache:
            return RedisCache(config)

        @provide(when=Has(MemcachedConfig), scope=Scope.APP)
        def memcached_impl(self, config: MemcachedConfig) -> Cache:
            return MemcachedCache(config)


    container = make_container(MyProvider, context={})


In this case,

* ``memcached_impl`` is not used because no real factory for ``MemcachedConfig`` is provided
* ``redis_impl`` is not used while it is registered as ``from_context`` but no real value is provided.
* ``base_impl`` is used as a default one, because none of later is active

[chirizxc]

Maybe something like:

    def _make_implicit_has_factory(
        self,
        key: DependencyKey,
        root_scope: BaseScope,
    ) -> Factory:
        return Factory(
            scope=root_scope,
            source=key.type_hint,
            provides=key,
            is_to_bind=False,
            dependencies=[],
            kw_dependencies={},
            type_=FactoryType.CONTEXT,
            cache=False,
            when_override=None,
            when_active=BoolMarker(False),
            when_component=key.component,
            when_dependencies=[],
        )

    def _add_implicit_has_factories(
        self,
        factories: dict[DependencyKey, Factory],
    ) -> None:
        root_scope = next(iter(self.scopes))
        seen = set(factories)
        missing_keys: list[DependencyKey] = []
        stack = list(factories.values())

        while stack:
            factory = stack.pop()
            stack.extend(factory.when_dependencies)

            for marker in unpack_marker(factory.when_active):
                if isinstance(marker, Has):
                    key = DependencyKey(marker.value, factory.when_component)
                    if key not in seen:
                        seen.add(key)
                        missing_keys.append(key)

            for marker in unpack_marker(factory.when_override):
                if isinstance(marker, Has):
                    key = DependencyKey(marker.value, factory.when_component)
                    if key not in seen:
                        seen.add(key)
                        missing_keys.append(key)

        factories.update({
            key: self._make_implicit_has_factory(key, root_scope)
            for key in missing_keys
        })

_iter_factory_markers create extra function calls and yield from frames for each node in tree (while and stack.pop() cheaper)

[Stefanqn]

thanks for the review. Please take a look at my latest commit.

I updated src/dishka/graph_builder/builder.py to use the iterative stack traversal and pass root_scope explicitly into _make_implicit_has_factory(...), following your suggestion.

[Tishka17]

This certainly fixes the case

@provide(when=Has(SecurityConfig))
def cors_setup(self, cfg: SecurityConfig) -> AppStartSideEffect:
   return AppStartSideEffect()

But also hides the problem in case:

@provide(when=~Has(SecurityConfig))
def cors_setup(self, cfg: SecurityConfig) -> AppStartSideEffect:
   return AppStartSideEffect()

[sonarqubecloud]

Quality Gate passed

Issues
1 New issue
0 Accepted issues

Measures
0 Security Hotspots
0.0% Coverage on New Code
0.0% Duplication on New Code

See analysis details on SonarQube Cloud

[Stefanqn]

I reworked this so Has(T) is handled in validation rather than by registering synthetic factories.

• when=Has(T): if T has no real factory, the condition is statically false, so validator skips that branch
• when=~Has(T): the branch stays validatable, so a factory that still requires T now correctly fails validation
  So the case you pointed out is no longer masked:
  @provide(when=~Has(SecurityConfig))
  def cors_setup(self, cfg: SecurityConfig) -> AppStartSideEffect:
  ...
  This now fails at graph validation again.
  I also updated tests to cover:
• Has(T) guarded branches with missing T
• the contradictory ~Has(T) + dependency-on-T case
• from_context(T) remaining runtime-dependent on actual context presence

[Tishka17]

I think I have implemented static evaluation: see #696

[Stefanqn]

def is_zero(value: int) -> bool:
    return value == 0
def fallback() -> str:
    return "a"
def needs_float(value: float) -> str:
    return str(value)

def test_surprise():
    provider = Provider(scope=Scope.APP)
    provider.provide(lambda: 1, provides=int)
    provider.activate(is_zero, Marker("ZERO")) # false
    provider.provide(fallback, provides=str)
    provider.provide(needs_float, provides=str, when=Marker("ZERO"))
    container = make_container(provider)

[Tishka17]

I am closing this in favor of static evaluation, which is merged. We can continue discussion in related PR about deferring checks