rex/startup_validation.py

Rama master · réplica de lectura del repositorio original.

"""Startup environment variable validation for Rex.

Validates required and optional environment variables before any other
initialization, so misconfigured deployments fail loudly with clear
error messages rather than silently misbehaving.

Usage::

    from rex.startup_validation import check_startup_env

    # Call early in main() before any other initialization
    check_startup_env()  # uses built-in _ENV_SPECS, exits 1 on failure

Custom specs can be passed::

    from rex.startup_validation import EnvVarSpec, check_startup_env, _validate_int

    check_startup_env([
        EnvVarSpec("MY_REQUIRED_KEY", required=True),
        EnvVarSpec("MY_PORT", required=False, validator=_validate_int),
    ])
"""

from __future__ import annotations

import os
import sys
from collections.abc import Callable
from dataclasses import dataclass
from urllib.parse import urlparse

# ---------------------------------------------------------------------------
# Built-in validators
# ---------------------------------------------------------------------------


def _validate_int(value: str) -> str | None:
    """Return None if *value* is a valid integer string, else an error string."""
    try:
        int(value)
        return None
    except (ValueError, TypeError):
        return f"expected integer, got {value!r}"


def _validate_url(value: str) -> str | None:
    """Return None if *value* is a valid http/https URL, else an error string."""
    try:
        parsed = urlparse(value)
    except Exception:  # noqa: BLE001
        return f"cannot be parsed as a URL: {value!r}"
    if parsed.scheme not in ("http", "https"):
        return f"expected http/https URL, got {value!r}"
    if not parsed.netloc:
        return f"expected http/https URL, got {value!r}"
    return None


# ---------------------------------------------------------------------------
# EnvVarSpec
# ---------------------------------------------------------------------------


@dataclass
class EnvVarSpec:
    """Specification for a single environment variable.

    Attributes:
        name: Environment variable name (e.g. ``"REX_SPEAK_PORT"``).
        required: Whether the variable *must* be present.  Optional by default.
        validator: Optional callable that receives the string value and returns
            ``None`` on success or a non-empty error string on failure.
        description: Human-readable description (used in error messages).
    """

    name: str
    required: bool = False
    validator: Callable[[str], str | None] | None = None
    description: str = ""


# ---------------------------------------------------------------------------
# Built-in specs for known Rex env vars
# ---------------------------------------------------------------------------

_ENV_SPECS: list[EnvVarSpec] = [
    EnvVarSpec(
        "REX_SPEAK_PORT",
        required=False,
        validator=_validate_int,
        description="Port for the TTS speak API server",
    ),
    EnvVarSpec(
        "OPENAI_BASE_URL",
        required=False,
        validator=_validate_url,
        description="Custom base URL for the OpenAI-compatible API",
    ),
    EnvVarSpec(
        "HA_BASE_URL",
        required=False,
        validator=_validate_url,
        description="Home Assistant base URL",
    ),
    EnvVarSpec(
        "OLLAMA_BASE_URL",
        required=False,
        validator=_validate_url,
        description="Ollama API base URL",
    ),
]


# ---------------------------------------------------------------------------
# Core validation functions
# ---------------------------------------------------------------------------


def validate_startup_env(
    specs: list[EnvVarSpec] | None = None,
    *,
    env: dict[str, str] | None = None,
) -> list[str]:
    """Validate environment variables against *specs*.

    Args:
        specs: List of :class:`EnvVarSpec` to validate.  When ``None``, the
            built-in :data:`_ENV_SPECS` list is used.
        env: Mapping to use instead of :data:`os.environ`.  Useful for testing.

    Returns:
        A list of human-readable error strings.  Empty when all checks pass.
    """
    if specs is None:
        specs = _ENV_SPECS
    if env is None:
        env = dict(os.environ)

    errors: list[str] = []

    for spec in specs:
        value = env.get(spec.name)
        if not value:
            # Treat missing (None) and empty string the same: not provided.
            if spec.required:
                errors.append(f"Required environment variable {spec.name!r} is not set.")
        else:
            if spec.validator is not None:
                err = spec.validator(value)
                if err:
                    errors.append(f"Environment variable {spec.name!r} is invalid: {err}.")

    return errors


def check_startup_env(
    specs: list[EnvVarSpec] | None = None,
    *,
    env: dict[str, str] | None = None,
    exit_on_error: bool = True,
) -> list[str]:
    """Validate environment variables and optionally exit on failure.

    This is the primary entry point.  Call it as early as possible in
    ``main()`` before any other initialization.

    Args:
        specs: Specs to validate (defaults to built-in :data:`_ENV_SPECS`).
        env: Override environment mapping (defaults to :data:`os.environ`).
        exit_on_error: When ``True`` (the default) and validation fails, print
            all errors to stderr and call :func:`sys.exit(1)`.

    Returns:
        List of error strings.  Empty when all checks pass.
    """
    errors = validate_startup_env(specs, env=env)
    if errors and exit_on_error:
        for err in errors:
            print(f"ERROR: {err}", file=sys.stderr)
        sys.exit(1)
    return errors


__all__ = [
    "EnvVarSpec",
    "validate_startup_env",
    "check_startup_env",
    "_validate_int",
    "_validate_url",
    "_ENV_SPECS",
]