yau-plant-assistant/api/config.py

"""Configuration, read once from the environment.

Values come from ~/ai/api.env on lin001 (0600, not in Git). .env.example in the
repo root lists every key with no values. Nothing here has a secret default,
and nothing here is ever logged.
"""

from __future__ import annotations

import os
from functools import lru_cache

from pydantic import BaseModel


class Settings(BaseModel):
    # --- imh (pending) -----------------------------------------------------
    use_fixtures: bool = True

    # --- local -------------------------------------------------------------
    pghost: str = "pg-ai"
    pgport: int = 5432
    pgdatabase: str = "plant"
    pguser: str = "agent_ro"
    pgpassword: str = ""

    # --- Azure OpenAI ------------------------------------------------------
    azure_openai_endpoint: str = ""
    azure_openai_api_key: str = ""
    azure_openai_api_version: str = ""
    chat_deployment: str = ""       # flagship - final prose only
    cheap_deployment: str = ""      # classifier, entities, tool selection
    embed_deployment: str = ""      # text-embedding-3-small

    # --- no-LLM stub mode --------------------------------------------------
    # Off by default and must stay that way. See api/stub.py for what it does
    # and, more importantly, what it does not prove.
    no_llm_stub: bool = False

    # --- behaviour ---------------------------------------------------------
    classifier_confidence_threshold: float = 0.7
    site_timezone: str = "Australia/Sydney"
    # Where the CONTROLLED copy of a procedure actually lives. A site fact, not
    # a document fact - it is the same for every document, so it is not in
    # doc_chunks. It is here rather than left to the model because sending an
    # operator to a controlled copy that does not exist is worse than telling
    # them nothing. The default says who to ask, which is always true.
    controlled_copy_location: str = (
        "Ask the WRPS document controller - this assistant does not hold "
        "controlled copies."
    )
    max_rows_returned: int = 5000
    query_timeout_seconds: int = 30
    max_output_tokens: int = 1200

    # --- Cube --------------------------------------------------------------
    cubejs_api_url: str = "http://cube:4000/cubejs-api/v1"
    cubejs_api_secret: str = ""

    # --- Langfuse ----------------------------------------------------------
    langfuse_host: str = "http://langfuse:3000"
    langfuse_public_key: str = ""
    langfuse_secret_key: str = ""

    def dsn(self) -> str:
        """Postgres DSN. Never log the result - it carries the password."""
        return (
            f"postgresql://{self.pguser}:{self.pgpassword}"
            f"@{self.pghost}:{self.pgport}/{self.pgdatabase}"
        )

    def redacted(self) -> dict[str, object]:
        """Safe to log and safe to return from /healthz."""
        secret = {"pgpassword", "azure_openai_api_key", "cubejs_api_secret",
                  "langfuse_secret_key"}
        return {
            k: ("set" if v else "unset") if k in secret else v
            for k, v in self.model_dump().items()
        }


@lru_cache
def settings() -> Settings:
    env = os.environ
    return Settings(
        use_fixtures=env.get("USE_FIXTURES", "true").lower() == "true",
        pghost=env.get("PGHOST", "pg-ai"),
        pgport=int(env.get("PGPORT", "5432")),
        pgdatabase=env.get("PGDATABASE", "plant"),
        pguser=env.get("PGUSER", "agent_ro"),
        pgpassword=env.get("PGPASSWORD", ""),
        no_llm_stub=env.get("NO_LLM_STUB", "false").lower() == "true",
        azure_openai_endpoint=env.get("AZURE_OPENAI_ENDPOINT", ""),
        azure_openai_api_key=env.get("AZURE_OPENAI_API_KEY", ""),
        azure_openai_api_version=env.get("AZURE_OPENAI_API_VERSION", ""),
        chat_deployment=env.get("CHAT_DEPLOYMENT", ""),
        cheap_deployment=env.get("CHEAP_DEPLOYMENT", ""),
        embed_deployment=env.get("EMBED_DEPLOYMENT", ""),
        classifier_confidence_threshold=float(
            env.get("CLASSIFIER_CONFIDENCE_THRESHOLD", "0.7")
        ),
        site_timezone=env.get("SITE_TIMEZONE", "Australia/Sydney"),
        controlled_copy_location=env.get(
            "CONTROLLED_COPY_LOCATION",
            "Ask the WRPS document controller - this assistant does not hold "
            "controlled copies.",
        ),
        max_rows_returned=int(env.get("MAX_ROWS_RETURNED", "5000")),
        query_timeout_seconds=int(env.get("QUERY_TIMEOUT_SECONDS", "30")),
        max_output_tokens=int(env.get("MAX_OUTPUT_TOKENS", "1200")),
        cubejs_api_url=env.get("CUBEJS_API_URL", "http://cube:4000/cubejs-api/v1"),
        cubejs_api_secret=env.get("CUBEJS_API_SECRET", ""),
        langfuse_host=env.get("LANGFUSE_HOST", "http://langfuse:3000"),
        langfuse_public_key=env.get("LANGFUSE_PUBLIC_KEY", ""),
        langfuse_secret_key=env.get("LANGFUSE_SECRET_KEY", ""),
    )