X-Financial/.tmp/lightrag_inspect/lightrag_pkg/lightrag/api/runtime_validation.py

"""Helpers for validating startup runtime expectations from `.env`."""

from __future__ import annotations

import os
from dataclasses import dataclass
from pathlib import Path

from dotenv import dotenv_values

_CONTAINER_RUNTIME_TARGETS = {"compose", "docker"}


@dataclass(frozen=True)
class RuntimeEnvironment:
    """Describes whether the current process is running in a container runtime."""

    in_container: bool
    in_docker: bool
    in_kubernetes: bool

    @property
    def label(self) -> str:
        if self.in_kubernetes:
            return "Kubernetes"
        if self.in_docker:
            return "Docker"
        return "host"


def _read_cgroup_content() -> str:
    """Best-effort read of cgroup metadata for container detection."""

    for candidate in ("/proc/1/cgroup", "/proc/self/cgroup"):
        try:
            return Path(candidate).read_text(encoding="utf-8")
        except OSError:
            continue
    return ""


def detect_runtime_environment(
    environ: dict[str, str] | None = None,
) -> RuntimeEnvironment:
    """Detect whether the current process is running on host, Docker, or Kubernetes."""

    environ = environ or os.environ
    cgroup_content = _read_cgroup_content().lower()

    in_kubernetes = bool(
        environ.get("KUBERNETES_SERVICE_HOST")
        or Path("/var/run/secrets/kubernetes.io/serviceaccount").exists()
        or "kubepods" in cgroup_content
        or "kubernetes" in cgroup_content
    )
    in_docker = bool(
        Path("/.dockerenv").exists()
        or Path("/run/.containerenv").exists()
        or any(
            marker in cgroup_content
            for marker in ("docker", "containerd", "libpod", "podman")
        )
    )

    return RuntimeEnvironment(
        in_container=in_kubernetes or in_docker,
        in_docker=in_docker,
        in_kubernetes=in_kubernetes,
    )


def load_runtime_target_from_env_file(env_path: str | Path = ".env") -> str | None:
    """Return the raw LIGHTRAG_RUNTIME_TARGET value from the `.env` file, if present."""

    env_values = dotenv_values(str(env_path))
    runtime_target = env_values.get("LIGHTRAG_RUNTIME_TARGET")
    if runtime_target is None:
        return None
    return runtime_target.strip()


def validate_runtime_target(
    runtime_target: str | None,
    runtime_environment: RuntimeEnvironment | None = None,
) -> tuple[bool, str | None]:
    """Validate `.env` runtime target against the current runtime environment."""

    if runtime_target is None:
        return True, None

    normalized_target = runtime_target.strip().lower()
    runtime_environment = runtime_environment or detect_runtime_environment()

    if normalized_target == "host":
        if runtime_environment.in_container:
            return (
                False,
                "Configuration error in .env: LIGHTRAG_RUNTIME_TARGET=host.\n"
                "This value from .env requires the server process to run on the host, "
                f"but the current process is running inside {runtime_environment.label}.",
            )
        return True, None

    if normalized_target in _CONTAINER_RUNTIME_TARGETS:
        if runtime_environment.in_container:
            return True, None
        return (
            False,
            f"Configuration error in .env: LIGHTRAG_RUNTIME_TARGET={runtime_target}.\n"
            "This value from .env requires the server process to run inside Docker or "
            "Kubernetes, but the current process is running on the host.",
        )

    return (
        False,
        f"Configuration error in .env: LIGHTRAG_RUNTIME_TARGET={runtime_target!r}.\n"
        "This value from .env must be 'host' or 'compose' (alias: 'docker').",
    )


def validate_runtime_target_from_env_file(
    env_path: str | Path = ".env",
    runtime_environment: RuntimeEnvironment | None = None,
) -> tuple[bool, str | None]:
    """Load LIGHTRAG_RUNTIME_TARGET from `.env` and validate it if declared."""

    runtime_target = load_runtime_target_from_env_file(env_path)
    return validate_runtime_target(runtime_target, runtime_environment)