"""Admin settings service (DB-backed).""" from __future__ import annotations from typing import Any, Dict, List, Optional from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.models import Setting # Settings that can be edited by admin via the admin panel EDITABLE_SETTING_KEYS = { "llm.base_url": str, "llm.api_key": str, "llm.model": str, "llm.temperature": float, "llm.step_temperature": float, "llm.summary_temperature": float, "llm.max_tokens": int, "llm.request_timeout": int, "llm.streaming": bool, "context.recent_messages": int, "context.compress_threshold": int, "context.summary_messages": int, "context.max_tokens_total": int, "triggers.enabled": bool, "triggers.check_interval": int, # Embeddings / RAG "embedding.provider": str, # "hash" | "openai" "embedding.base_url": str, # OpenAI-compatible base URL (e.g. http://localhost:1234/v1) "embedding.api_key": str, # API key (may be empty for local servers) "embedding.model": str, # e.g. text-embedding-3-small, bge-m3, nomic-embed-text "embedding.dim": int, # vector dimension; 0 = auto-probe from endpoint "embedding.request_timeout": int, # request timeout, seconds } async def get_all_settings(db: AsyncSession) -> Dict[str, Any]: result = await db.execute(select(Setting)) return {row.key: row.value for row in result.scalars().all()} async def get_setting(db: AsyncSession, key: str, default: Any = None) -> Any: result = await db.execute(select(Setting).where(Setting.key == key)) row = result.scalars().first() return row.value if row else default async def update_settings(db: AsyncSession, updates: Dict[str, Any]) -> Dict[str, Any]: for key, value in updates.items(): if key not in EDITABLE_SETTING_KEYS: continue expected = EDITABLE_SETTING_KEYS[key] try: if expected is bool: value = bool(value) elif expected is int: value = int(value) elif expected is float: value = float(value) else: value = str(value) except (TypeError, ValueError): continue result = await db.execute(select(Setting).where(Setting.key == key)) row = result.scalars().first() if row is None: db.add(Setting(key=key, value=value)) else: row.value = value await db.commit() return await get_all_settings(db) def cast_setting(key: str, value: Any) -> Any: """Cast raw DB value to the expected type for use.""" if key not in EDITABLE_SETTING_KEYS: return value expected = EDITABLE_SETTING_KEYS[key] try: if expected is bool: if isinstance(value, str): return value.lower() in ("1", "true", "yes", "on") return bool(value) if expected is int: return int(value) if expected is float: return float(value) return str(value) except (TypeError, ValueError): return value