Configuration¶

class Config(BaseModel):
    """A fully loaded configuration.

    ``extra="forbid"`` rejects unknown top-level sections, and applies inside
    ``nats`` and ``service`` too. It deliberately does not reach inside a
    plugin's table: a plugin validates its own table against the model it
    declares as ``Meta.config_model``.

    ``hide_input_in_errors=True`` drops the offending value from a validation
    error's rendering. This is the model whose validation pass constructs the
    ``ValidationError`` that ``load_config`` turns into a ``ConfigError``, so it
    is the model this must be set on: the same setting on a nested model such as
    ``NATSConfig`` has no effect on the error that actually gets raised. Without
    it, a failed cross-field check such as
    ``_reject_multiple_authentication_styles`` would echo the raw, still-unwrapped
    ``[nats]`` table — including a plaintext password or token — into the message
    that ends up in a startup log.

    Attributes:
        nats: Broker connection settings.
        service: Service-level settings.
        plugins: Per-plugin configuration tables, keyed by plugin id. A plugin
            normally reads through ``PluginBase.options`` rather than this
            directly, so its table is validated against the model it declares.

    """

    model_config = ConfigDict(extra="forbid", frozen=True, hide_input_in_errors=True)

    nats: NATSConfig = Field(default_factory=NATSConfig)
    service: ServiceConfig = Field(default_factory=ServiceConfig)
    plugins: Dict[str, Dict[str, Any]] = Field(default_factory=dict)

    # --- Concrete Methods ---

    def plugin_config(self, plugin_id: str) -> Dict[str, Any]:
        """Return the configuration table declared for a plugin.

        The result is a copy, so a caller that mutates it cannot reconfigure
        later readers in the process.

        Args:
            plugin_id: The plugin's own id, as assigned during ``connect``.

        Returns:
            The plugin's configuration table, or an empty mapping when no table
            is declared for that id.

        """
        return copy.deepcopy(dict(self.plugins.get(plugin_id, {})))

class NATSConfig(BaseModel):
    """Broker connection settings.

    Every attribute except ``url`` and ``reconnect_time_wait_s`` defaults to
    ``None`` meaning unset. An unset attribute is omitted from the connect call
    rather than passed as a value the framework guessed at, so the client
    library's own default applies.

    Numeric attributes are strict: a boolean is rejected rather than coerced to
    an integer, which pydantic's default mode would otherwise do.

    Attributes:
        url: Broker to connect to.
        user: Username for user/password authentication.
        password: Password for user/password authentication. Wrapped so it
            cannot leak through ``repr`` or a traceback.
        token: Token for token authentication. Wrapped so it cannot leak
            through ``repr`` or a traceback.
        credentials_file: Path to an nkey/JWT credentials file.
        tls_ca_file: Path to a CA bundle, verified readable at load time.
        tls_hostname: Hostname the broker's certificate is expected to present.
        reconnect_time_wait_s: Seconds to wait between reconnection attempts.
        max_reconnect_attempts: Retry ceiling. Unset means the client library's
            own default, which is 60 attempts rather than unlimited.

    """

    model_config = ConfigDict(extra="forbid", frozen=True)

    url: str = DEFAULT_NATS_URL
    user: str | None = None
    password: SecretStr | None = None
    token: SecretStr | None = None
    credentials_file: str | None = None
    tls_ca_file: str | None = None
    tls_hostname: str | None = None
    reconnect_time_wait_s: StrictInt | StrictFloat = DEFAULT_RECONNECT_TIME_WAIT_S
    max_reconnect_attempts: StrictInt | None = None

    # --- Validators ---

    @model_validator(mode="after")
    def _reject_multiple_authentication_styles(self) -> NATSConfig:
        """Reject more than one authentication style.

        Preferring one style over another silently would hide a deployment
        mistake, so a combination is refused rather than resolved.

        Returns:
            The validated settings.

        Raises:
            ValueError: More than one authentication style is configured.

        """
        styles = {
            "user/password": bool(self.user or self.password),
            "token": bool(self.token),
            "credentials_file": bool(self.credentials_file),
        }
        configured = sorted(name for name, present in styles.items() if present)
        if len(configured) > 1:
            raise ValueError(f"only one authentication style may be configured, found {' and '.join(configured)}")
        return self

    @model_validator(mode="after")
    def _verify_tls_ca_file(self) -> NATSConfig:
        """Verify the CA bundle is readable, so a bad one fails at startup.

        The message must name the CA file: callers rely on the path appearing in
        the error.

        Returns:
            The validated settings.

        Raises:
            ValueError: The CA file does not exist or could not be loaded.

        """
        if self.tls_ca_file is None:
            return self

        ca_path = Path(self.tls_ca_file)
        if not ca_path.is_file():
            raise ValueError(f"TLS CA file does not exist: {ca_path}")
        try:
            ssl.create_default_context().load_verify_locations(cafile=str(ca_path))
        except (OSError, ssl.SSLError) as exc:
            raise ValueError(f"TLS CA file could not be loaded: {ca_path}: {exc}") from exc
        return self

    # --- Properties ---

    @cached_property
    def tls_context(self) -> ssl.SSLContext | None:
        """SSL context built from ``tls_ca_file``, or ``None`` when TLS is unset.

        The client library takes a prepared context rather than a CA path. Built
        on first access and cached; the file was already verified readable at
        load time, so this cannot fail for a configuration that loaded.

        """
        if self.tls_ca_file is None:
            return None
        context = ssl.create_default_context()
        context.load_verify_locations(cafile=self.tls_ca_file)
        return context

class ServiceConfig(BaseModel):
    """Service-level settings.

    Attributes:
        heartbeat_interval_ms: Heartbeat cadence in milliseconds.

    """

    model_config = ConfigDict(extra="forbid", frozen=True)

    heartbeat_interval_ms: StrictInt = DEFAULT_HEARTBEAT_INTERVAL_MS

class ConfigError(Exception):
    """Raised when a configuration cannot be used.

    Always raised at load time, never mid-run, and names the offending file.
    Validation failures carry pydantic's report, which names every offending
    key at once.

    Args:
        message: Description of what is wrong.
        path: Configuration file the problem was found in, if any.

    Attributes:
        path: The offending file as a string, or ``None``.

    """

    def __init__(self, message: str, *, path: str | Path | None = None) -> None:
        self.path = str(path) if path is not None else None
        super().__init__(f"{self.path}: {message}" if self.path is not None else message)