Conditional Providers¶

Introduction¶

In many applications, the correct implementation of a service depends on the runtime environment. You might use Redis in production but an in-memory store in development, enable debug tooling only when a flag is set, or activate an adapter only when a companion service is registered.

waku supports conditional providers through the when= parameter available on every provider helper (singleton, scoped, transient, object_, many). Combined with markers and activator functions, this lets you register multiple implementations of the same interface and let the framework choose the right one at startup.

Markers¶

A Marker is a named boolean flag that controls whether a provider is active. Markers are created as simple instances and used in two places: on the when= parameter of a provider, and in an activator() call that decides the marker's value at startup.

1
2
3
4
5

from waku.di import Marker

USE_REDIS = Marker('use_redis')
PRODUCTION = Marker('production')
DEBUG = Marker('debug')

Markers are inert on their own — they only become meaningful once you wire them to an activator function and attach them to providers via when=.

Activator functions¶

An activator is a callable that returns bool. At container construction time, Dishka calls every registered activator and uses the result to decide which markers are active.

The activator() helper creates a Provider that you register alongside your other providers:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

from dataclasses import dataclass

from waku.di import Marker, activator


@dataclass
class AppConfig:
    use_redis: bool = False
    environment: str = 'development'
    debug: bool = False


USE_REDIS = Marker('use_redis')
PRODUCTION = Marker('production')
DEBUG = Marker('debug')


def is_redis(config: AppConfig) -> bool:
    return config.use_redis


def is_production(config: AppConfig) -> bool:
    return config.environment == 'production'


def is_debug(config: AppConfig) -> bool:
    return config.debug

Providing config via context¶

To make configuration available to activator functions, register a contextual provider at the APP scope and pass the actual instance through the context parameter of WakuFactory.

When the container starts:

1. The container resolves AppConfig from the context dictionary.
2. Each activator function receives the config and returns True or False.
3. Providers whose when= marker evaluated to True become active; the rest are skipped.

The example below demonstrates this pattern end-to-end.

Full example: environment-based service selection¶

A self-contained example that selects a cache implementation based on configuration:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65

from dataclasses import dataclass
from typing import Protocol

from waku import WakuFactory, module
from waku.di import Marker, Scope, activator, contextual, scoped


@dataclass
class AppConfig:
    use_redis: bool = False


USE_REDIS = Marker('use_redis')


def is_redis(config: AppConfig) -> bool:
    return config.use_redis


class ICache(Protocol):
    def get(self, key: str) -> str | None: ...


@dataclass
class RedisCache(ICache):
    def get(self, key: str) -> str | None:
        return f'redis:{key}'


@dataclass
class InMemoryCache(ICache):
    def get(self, key: str) -> str | None:
        return f'memory:{key}'


@module(
    providers=[
        contextual(AppConfig, scope=Scope.APP),
        activator(is_redis, USE_REDIS),
        scoped(ICache, RedisCache, when=USE_REDIS),
        scoped(ICache, InMemoryCache, when=~USE_REDIS),
    ],
)
class AppModule:
    pass


async def main() -> None:
    app = WakuFactory(
        AppModule,
        context={AppConfig: AppConfig(use_redis=True)},
    ).create()

    async with app, app.container() as container:
        cache = await container.get(ICache)
        assert isinstance(cache, RedisCache)

    app = WakuFactory(
        AppModule,
        context={AppConfig: AppConfig(use_redis=False)},
    ).create()

    async with app, app.container() as container:
        cache = await container.get(ICache)
        assert isinstance(cache, InMemoryCache)

Has(Type) — presence-based activation¶

Has(Type) activates a provider only when the specified type is registered somewhere in the container. No activator function is needed — the container checks its own registry at build time.

This is useful for feature-flag-style activation where a feature is enabled by the mere presence of a dependency:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

from dataclasses import dataclass

from waku import WakuFactory, module
from waku.di import Has, scoped


@dataclass
class FeatureA:
    pass


@dataclass
class FeatureAConsumer:
    a: FeatureA


@module(
    providers=[
        scoped(FeatureA),
        scoped(FeatureAConsumer, when=Has(FeatureA)),
    ],
)
class AppModule:
    pass


async def main() -> None:
    app = WakuFactory(AppModule).create()

    async with app, app.container() as container:
        consumer = await container.get(FeatureAConsumer)
        assert isinstance(consumer.a, FeatureA)

If FeatureA is removed from the providers list, the container fails to build with GraphMissingFactoryError during graph validation.

Marker composition¶

Markers support boolean-style composition using Python operators. This lets you express complex activation conditions without writing custom activator logic:

Negation (~)¶

Activate when a marker is not active:

from waku.di import Marker, scoped

USE_REDIS = Marker('use_redis')

scoped(ICache, RedisCache, when=USE_REDIS)
scoped(ICache, InMemoryCache, when=~USE_REDIS)

AND (&)¶

Activate only when both markers are active:

from waku.di import Marker, scoped

DEBUG = Marker('debug')
PRODUCTION = Marker('production')

scoped(DebugProductionService, when=DEBUG & PRODUCTION)

OR (|)¶

Activate when either marker is active:

from waku.di import Has

scoped(SharedConsumer, when=Has(FeatureA) | Has(FeatureB))

activator() reference¶

from waku.di import activator

def activator(fn: Callable[..., bool], *markers: Any) -> Provider:
    """Create a Provider with an activator for simple cases.

    Args:
        fn: Callable that returns bool to determine marker activation.
        *markers: Marker instances or types to activate.

    Returns:
        Provider with the activator registered.
    """

A single activator call can activate multiple markers:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

from waku.di import Marker, activator

DEBUG = Marker('debug')
VERBOSE = Marker('verbose')


def is_debug_mode() -> bool:
    return True


activator(is_debug_mode, DEBUG, VERBOSE)

Further reading¶

• Providers — provider types and scopes
• Modules — module system and provider registration
• Dishka conditional activation — advanced patterns from the underlying DI framework