"""Application configuration loaded from environment + DB-backed admin settings.""" from __future__ import annotations import os import secrets from functools import lru_cache from typing import List from pydantic import Field from pydantic_settings import BaseSettings, SettingsConfigDict def _parse_cors_origins(raw: str) -> List[str]: """Parse a CORS_ORIGINS env value into a list of origin strings. Accepts (in priority order): - JSON array string: '["http://a","http://b"]' - comma-separated: "http://a,http://b" - single value: "http://a" Strips whitespace and drops empties. Returns the default list if input is empty/blank. This helper exists because pydantic-settings' EnvSettingsSource treats `List[str]` as a "complex" type and tries to JSON-decode the raw env value BEFORE any field validator runs — so a plain comma-separated string from docker-compose crashes Settings() at import time on older pydantic-settings versions. Storing the field as `str` sidesteps that entirely; this helper parses it on demand. """ if not raw: return ["http://localhost:5173"] v = raw.strip() if not v: return ["http://localhost:5173"] if v.startswith("["): import json try: parsed = json.loads(v) if isinstance(parsed, list): return [str(o).strip() for o in parsed if str(o).strip()] except Exception: pass # fall through to comma-split return [o.strip() for o in v.split(",") if o.strip()] class Settings(BaseSettings): model_config = SettingsConfigDict(env_file=".env", extra="ignore", case_sensitive=False) # Database database_url: str = "postgresql+asyncpg://airpg:airpg_secret@localhost:5432/airpg" # Redis redis_url: str = "redis://localhost:6379/0" # Qdrant qdrant_url: str = "http://localhost:6333" # Auth jwt_secret: str = Field(default_factory=lambda: secrets.token_hex(32)) jwt_algorithm: str = "HS256" access_token_expire_minutes: int = 60 * 24 * 7 # 7 days # Admin setup # If empty, will be generated at first run and printed to console. admin_setup_token: str = "" # CORS # Stored as a raw string (NOT List[str]) so pydantic-settings' env source # treats it as a simple scalar and never attempts JSON-decoding. The # parsed list is exposed via the `cors_origins_list` property below. # Accepts either a comma-separated string ("http://a,http://b") or a # JSON-array string ('["http://a","http://b"]'). cors_origins: str = "http://localhost:5173" @property def cors_origins_list(self) -> List[str]: """Parsed list of allowed CORS origins (see `_parse_cors_origins`).""" return _parse_cors_origins(self.cors_origins) # Logging log_level: str = "INFO" # Default LLM (used to seed DB on first run; overridable via admin panel) default_llm_base_url: str = "http://localhost:1234/v1" default_llm_api_key: str = "dummy" default_llm_model: str = "local-model" # Default embeddings / RAG settings (overridable via admin panel) # provider="hash" is a deterministic offline fallback (no semantic quality). # Switch to "openai" and point embedding.base_url at an OpenAI-compatible /embeddings endpoint # for real semantic search. default_embedding_provider: str = "hash" default_embedding_base_url: str = "" # empty = reuse llm.base_url default_embedding_api_key: str = "" # empty = reuse llm.api_key default_embedding_model: str = "text-embedding-3-small" default_embedding_dim: int = 0 # 0 = auto-probe from endpoint default_embedding_request_timeout: int = 60 # Context manager defaults (admin-overridable) default_recent_messages: int = 10 default_compress_threshold: int = 20 default_summary_messages: int = 10 # Worker mode flag worker_mode: bool = False @property def is_worker(self) -> bool: return bool(os.getenv("WORKER_MODE")) or self.worker_mode @lru_cache def get_settings() -> Settings: return Settings() settings = get_settings()