Skip to content

notifications.registry

registry

Deprecated compatibility imports for :mod:codex_platform.messaging.registry.

Classes

ChannelRegistry

Registry for config-driven delivery channel factories.

Source code in src/codex_platform/messaging/registry.py 
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
class ChannelRegistry:
    """Registry for config-driven delivery channel factories."""

    def __init__(self) -> None:
        self._factories: list[tuple[str, Callable[[Any], DeliveryChannel | None]]] = []

    def register(
        self,
        name: str,
        factory: Callable[[Any], DeliveryChannel | None],
    ) -> None:
        """Register a channel factory."""

        if any(existing == name for existing, _factory in self._factories):
            log.warning("ChannelRegistry | duplicate registration for %s", name)
        self._factories.append((name, factory))

    def build_channels(self, config: Any) -> list[DeliveryChannel]:
        """Build available channels from registered factories."""

        channels: list[DeliveryChannel] = []
        for name, factory in self._factories:
            try:
                channel = factory(config)
                if channel is not None and channel.is_available():
                    channels.append(channel)
                    log.info("ChannelRegistry | %s enabled", name)
                else:
                    log.debug("ChannelRegistry | %s skipped (not configured)", name)
            except Exception:
                log.exception("ChannelRegistry | %s factory failed", name)
        return channels
Functions
register(name, factory)

Register a channel factory.

Source code in src/codex_platform/messaging/registry.py 
20
21
22
23
24
25
26
27
28
29
def register(
    self,
    name: str,
    factory: Callable[[Any], DeliveryChannel | None],
) -> None:
    """Register a channel factory."""

    if any(existing == name for existing, _factory in self._factories):
        log.warning("ChannelRegistry | duplicate registration for %s", name)
    self._factories.append((name, factory))
build_channels(config)

Build available channels from registered factories.

Source code in src/codex_platform/messaging/registry.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
def build_channels(self, config: Any) -> list[DeliveryChannel]:
    """Build available channels from registered factories."""

    channels: list[DeliveryChannel] = []
    for name, factory in self._factories:
        try:
            channel = factory(config)
            if channel is not None and channel.is_available():
                channels.append(channel)
                log.info("ChannelRegistry | %s enabled", name)
            else:
                log.debug("ChannelRegistry | %s skipped (not configured)", name)
        except Exception:
            log.exception("ChannelRegistry | %s factory failed", name)
    return channels