registry.py

Go to the documentation of this file.

"""Explicit-import connector registration.
 
Dependencies
------------
- oc_diagnostics (_dbg structured logger)
- oc_connect.connectors.telegram (lazy; the shipped Telegram bundle, imported for its registration side effect)
"""
from __future__ import annotations
 
from oc_diagnostics import _dbg
 
 
class ConnectorRegistry:
    """Small in-memory registry with fail-loud conflict handling."""
 
    def __init__(self) -> None:
        self._connectors: dict[str, object] = {}
 
    def register(self, name: str, connector: object) -> object:
        """Register a connector object, allowing only identical repeats."""
        if not name.strip():
            raise ValueError("connector name must not be blank")
        existing = self._connectors.get(name)
        if existing is not None and existing is not connector:
            raise ValueError(f"connector {name!r} is already registered")
        self._connectors[name] = connector
        _dbg("OC_CONNECT.REGISTRY", "register", f"registered connector {name}", 3)
        return connector
 
    def get(self, name: str) -> object:
        """Return a registered connector or fail with its public name."""
        try:
            return self._connectors[name]
        except KeyError:
            raise KeyError(f"connector {name!r} is not registered") from None
 
    def names(self) -> tuple[str, ...]:
        """Return registered names in deterministic order."""
        return tuple(sorted(self._connectors))
 
    def clear(self) -> None:
        """Remove registrations; intended for isolated tests."""
        self._connectors.clear()
 
 
registry = ConnectorRegistry()
 
 
def load_builtin_connectors() -> ConnectorRegistry:
    """Load shipped bundles by explicit import, never entry points."""
    from .connectors import telegram as _telegram
 
    _dbg(
        "OC_CONNECT.REGISTRY",
        "load_builtin_connectors",
        f"loaded bundle module {_telegram.__name__}",
        3,
    )
    return registry
 
 
__all__ = ["ConnectorRegistry", "load_builtin_connectors", "registry"]