118 lines
4.1 KiB
Python
118 lines
4.1 KiB
Python
"""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()
|