lookingglass/hyperglass/models/system.py

"""hyperglass System Settings model."""

# Standard Library
import typing as t
from pathlib import Path
from ipaddress import ip_address

# Third Party
from pydantic import (
    FilePath,
    RedisDsn,
    SecretStr,
    DirectoryPath,
    IPvAnyAddress,
    ValidationInfo,
    field_validator,
)
from pydantic_settings import BaseSettings, SettingsConfigDict

# Project
from hyperglass.util import at_least, cpu_count

if t.TYPE_CHECKING:
    # Third Party
    from rich.console import Console, RenderResult, ConsoleOptions

ListenHost = t.Union[None, IPvAnyAddress, t.Literal["localhost"]]

_default_app_path = Path("/etc/hyperglass")


class HyperglassSettings(BaseSettings):
    """hyperglass system settings, required to start hyperglass."""

    model_config = SettingsConfigDict(env_prefix="hyperglass_")

    config_file_names: t.ClassVar[t.Tuple[str, ...]] = ("config", "devices", "directives")
    default_app_path: t.ClassVar[Path] = _default_app_path
    original_app_path: Path = _default_app_path

    debug: bool = False
    dev_mode: bool = False
    disable_ui: bool = False
    app_path: DirectoryPath = _default_app_path
    redis_host: str = "localhost"
    redis_password: t.Optional[SecretStr] = None
    redis_db: int = 1
    redis_dsn: RedisDsn = None
    host: IPvAnyAddress = None
    port: int = 8001
    ca_cert: t.Optional[FilePath] = None
    container: bool = False

    def __init__(self, **kwargs) -> None:
        """Create hyperglass Settings instance."""
        super().__init__(**kwargs)
        if self.container:
            self.app_path = self.default_app_path

    def __rich_console__(self, console: "Console", options: "ConsoleOptions") -> "RenderResult":
        """Render a Rich table representation of hyperglass settings."""
        # Third Party
        from rich.panel import Panel
        from rich.style import Style
        from rich.table import Table, box
        from rich.pretty import Pretty

        table = Table(box=box.MINIMAL, border_style="subtle")
        table.add_column("Environment Variable", style=Style(color="#118ab2", bold=True))
        table.add_column("Value")
        params = sorted(
            (
                "debug",
                "dev_mode",
                "app_path",
                "redis_host",
                "redis_db",
                "redis_dsn",
                "host",
                "port",
            )
        )
        for attr in params:
            table.add_row(f"hyperglass_{attr}".upper(), Pretty(getattr(self, attr)))

        yield Panel.fit(table, title="hyperglass settings", border_style="subtle")

    @field_validator("host", mode="before")
    def validate_host(
        cls: "HyperglassSettings", value: t.Any, info: ValidationInfo
    ) -> IPvAnyAddress:
        """Set default host based on debug mode."""

        if value is None:
            if info.data.get("debug") is False:
                return ip_address("::1")
            if info.data.get("debug") is True:
                return ip_address("::")

        if isinstance(value, str):
            if value != "localhost":
                try:
                    return ip_address(value)
                except ValueError as err:
                    raise ValueError(str(value)) from err

            elif value == "localhost":
                return ip_address("::1")

        raise ValueError(str(value))

    @field_validator("redis_dsn", mode="before")
    def validate_redis_dsn(cls, value: t.Any, info: ValidationInfo) -> RedisDsn:
        """Construct a Redis DSN if none is provided."""
        if value is None:
            host = info.data.get("redis_host")
            db = info.data.get("redis_db")
            dsn = "redis://{}/{!s}".format(host, db)
            password = info.data.get("redis_password")
            if password is not None:
                dsn = "redis://:{}@{}/{!s}".format(password.get_secret_value(), host, db)
            return dsn
        return value

    def bind(self: "HyperglassSettings") -> str:
        """Format a listen_address. Wraps IPv6 address in brackets."""
        if self.host.version == 6:
            return f"[{self.host!s}]:{self.port!s}"
        return f"{self.host!s}:{self.port!s}"

    @property
    def log_level(self: "HyperglassSettings") -> str:
        """Get log level as string, inferred from debug mode."""
        if self.debug:
            return "DEBUG"
        return "WARNING"

    @property
    def workers(self: "HyperglassSettings") -> int:
        """Get worker count, inferred from debug mode."""
        if self.debug:
            return 1
        return cpu_count(2)

    @property
    def redis(self: "HyperglassSettings") -> t.Dict[str, t.Union[None, int, str]]:
        """Get redis parameters as a dict for convenient connection setups."""
        password = None
        if self.redis_password is not None:
            password = self.redis_password.get_secret_value()

        return {
            "db": self.redis_db,
            "host": self.redis_host,
            "password": password,
        }

    @property
    def redis_connection_pool(self: "HyperglassSettings") -> t.Dict[str, t.Any]:
        """Get Redis ConnectionPool keyword arguments."""
        return {"url": str(self.redis_dsn), "max_connections": at_least(8, cpu_count(2))}

    @property
    def dev_url(self: "HyperglassSettings") -> str:
        """Get the hyperglass URL for when dev_mode is enabled."""
        return f"http://localhost:{self.port!s}/"

    @property
    def prod_url(self: "HyperglassSettings") -> str:
        """Get the UI-facing hyperglass URL/path."""
        return "/api/"

    @property
    def static_path(self: "HyperglassSettings") -> Path:
        """Get static asset path."""
        return Path(self.app_path / "static")