feat: implement OmegaConf-based configuration management with uv
Configuration changes: - Replace pydantic-settings with OmegaConf for flexible config management - Support YAML config files via CONFIG_FILE environment variable - Support environment variables with STEWARD__SECTION__KEY format - Add config_schema.yaml as default configuration schema - Create pydantic models for each configuration section for type safety - Maintain backward compatibility via properties on Settings class - Add CONFIGURATION.md with comprehensive setup and usage guide - Update pyproject.toml to include config_schema.yaml in package data Build/deployment changes: - Replace pip with uv in Dockerfile for faster dependency installation - Create docker-compose.dev.yml for local development (build .) - Keep docker-compose.yml for production (uses ghcr.io/djw4/steward:latest) - Update .env.example with new STEWARD__* variable format This setup is designed for Kubernetes deployment: - Non-sensitive config goes in ConfigMap (config.yaml) - Secrets go in Secret resources (environment variables) - Single unified configuration system for all environments
This commit is contained in:
Daniel Wagner
+233
-32
@@ -1,45 +1,246 @@
|
||||
"""Configuration management for Steward."""
|
||||
"""Configuration management for Steward.
|
||||
|
||||
from pydantic_settings import BaseSettings, SettingsConfigDict
|
||||
Supports configuration via:
|
||||
1. YAML config file (passed via CONFIG_FILE env var)
|
||||
2. Environment variables (STEWARD__SECTION__KEY=value format)
|
||||
3. Default schema values
|
||||
|
||||
Environment variables take precedence over config file values.
|
||||
"""
|
||||
|
||||
import logging
|
||||
import os
|
||||
from pathlib import Path
|
||||
|
||||
from omegaconf import OmegaConf
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class Settings(BaseSettings):
|
||||
"""Application settings loaded from environment variables or .env file."""
|
||||
class TelegramConfig(BaseModel):
|
||||
"""Telegram bot configuration."""
|
||||
|
||||
model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8", extra="ignore")
|
||||
bot_token: str = Field(default="", description="Telegram bot token")
|
||||
allowed_user_ids: list[int] = Field(default_factory=list, description="Allowed user IDs")
|
||||
group_ids: list[int] = Field(default_factory=list, description="Allowed group/channel IDs")
|
||||
|
||||
# Telegram
|
||||
telegram_bot_token: str = ""
|
||||
telegram_allowed_user_ids: list[int] = []
|
||||
telegram_group_ids: list[int] = []
|
||||
|
||||
# LLM (OpenAI-compatible)
|
||||
openai_api_key: str = ""
|
||||
openai_base_url: str = "https://api.openai.com/v1"
|
||||
openai_model: str = "gpt-4o"
|
||||
openai_system_prompt: str = (
|
||||
"You are Steward, a persistent, trustworthy AI-assisted personal operations platform. "
|
||||
"You reduce cognitive load by observing, remembering, planning, and proposing actions. "
|
||||
"You are conservative, transparent, and policy-aware. "
|
||||
"Always explain your reasoning."
|
||||
class OpenAIConfig(BaseModel):
|
||||
"""OpenAI/LLM configuration."""
|
||||
|
||||
api_key: str = Field(default="", description="OpenAI API key")
|
||||
base_url: str = Field(default="https://api.openai.com/v1", description="API base URL")
|
||||
model: str = Field(default="gpt-4o", description="Model to use")
|
||||
system_prompt: str = Field(
|
||||
default=(
|
||||
"You are Steward, a persistent, trustworthy AI-assisted personal operations platform. "
|
||||
"You reduce cognitive load by observing, remembering, planning, and proposing actions. "
|
||||
"You are conservative, transparent, and policy-aware. "
|
||||
"Always explain your reasoning."
|
||||
),
|
||||
description="System prompt for the LLM",
|
||||
)
|
||||
|
||||
# Analysis / proposal worker
|
||||
analysis_target_url: str = ""
|
||||
analysis_target_api_key: str = ""
|
||||
analysis_cron_hour: int = 8
|
||||
analysis_cron_minute: int = 0
|
||||
|
||||
# Thread memory
|
||||
thread_memory_path: str = "thread_memory.json"
|
||||
class AnalysisConfig(BaseModel):
|
||||
"""Analysis/proposal worker configuration."""
|
||||
|
||||
# MCP / OpenAPI tool server (open-webui/openapi-servers compatible)
|
||||
# Set MCP_SERVER_URL to enable tool calling. The service fetches
|
||||
# /openapi.json from this URL to discover available tools.
|
||||
mcp_server_url: str = ""
|
||||
mcp_server_api_key: str = ""
|
||||
target_url: str = Field(default="", description="Target URL for analysis")
|
||||
target_api_key: str = Field(default="", description="API key for target")
|
||||
cron_hour: int = Field(default=8, description="Hour for cron schedule")
|
||||
cron_minute: int = Field(default=0, description="Minute for cron schedule")
|
||||
|
||||
|
||||
class MemoryConfig(BaseModel):
|
||||
"""Memory/persistence configuration."""
|
||||
|
||||
thread_memory_path: str = Field(default="thread_memory.json", description="Thread memory path")
|
||||
|
||||
|
||||
class ToolsConfig(BaseModel):
|
||||
"""Tools/MCP configuration."""
|
||||
|
||||
mcp_server_url: str = Field(default="", description="MCP server URL")
|
||||
mcp_server_api_key: str = Field(default="", description="MCP server API key")
|
||||
|
||||
|
||||
class Settings(BaseModel):
|
||||
"""Application settings with OmegaConf and pydantic integration."""
|
||||
|
||||
telegram: TelegramConfig = Field(default_factory=TelegramConfig)
|
||||
openai: OpenAIConfig = Field(default_factory=OpenAIConfig)
|
||||
analysis: AnalysisConfig = Field(default_factory=AnalysisConfig)
|
||||
memory: MemoryConfig = Field(default_factory=MemoryConfig)
|
||||
tools: ToolsConfig = Field(default_factory=ToolsConfig)
|
||||
|
||||
class Config:
|
||||
"""Pydantic config."""
|
||||
|
||||
arbitrary_types_allowed = True
|
||||
|
||||
# Compatibility properties for existing code
|
||||
@property
|
||||
def telegram_bot_token(self) -> str:
|
||||
"""Legacy property for backward compatibility."""
|
||||
return self.telegram.bot_token
|
||||
|
||||
@property
|
||||
def telegram_allowed_user_ids(self) -> list[int]:
|
||||
"""Legacy property for backward compatibility."""
|
||||
return self.telegram.allowed_user_ids
|
||||
|
||||
@property
|
||||
def telegram_group_ids(self) -> list[int]:
|
||||
"""Legacy property for backward compatibility."""
|
||||
return self.telegram.group_ids
|
||||
|
||||
@property
|
||||
def openai_api_key(self) -> str:
|
||||
"""Legacy property for backward compatibility."""
|
||||
return self.openai.api_key
|
||||
|
||||
@property
|
||||
def openai_base_url(self) -> str:
|
||||
"""Legacy property for backward compatibility."""
|
||||
return self.openai.base_url
|
||||
|
||||
@property
|
||||
def openai_model(self) -> str:
|
||||
"""Legacy property for backward compatibility."""
|
||||
return self.openai.model
|
||||
|
||||
@property
|
||||
def openai_system_prompt(self) -> str:
|
||||
"""Legacy property for backward compatibility."""
|
||||
return self.openai.system_prompt
|
||||
|
||||
@property
|
||||
def analysis_target_url(self) -> str:
|
||||
"""Legacy property for backward compatibility."""
|
||||
return self.analysis.target_url
|
||||
|
||||
@property
|
||||
def analysis_target_api_key(self) -> str:
|
||||
"""Legacy property for backward compatibility."""
|
||||
return self.analysis.target_api_key
|
||||
|
||||
@property
|
||||
def analysis_cron_hour(self) -> int:
|
||||
"""Legacy property for backward compatibility."""
|
||||
return self.analysis.cron_hour
|
||||
|
||||
@property
|
||||
def analysis_cron_minute(self) -> int:
|
||||
"""Legacy property for backward compatibility."""
|
||||
return self.analysis.cron_minute
|
||||
|
||||
@property
|
||||
def thread_memory_path(self) -> str:
|
||||
"""Legacy property for backward compatibility."""
|
||||
return self.memory.thread_memory_path
|
||||
|
||||
@property
|
||||
def mcp_server_url(self) -> str:
|
||||
"""Legacy property for backward compatibility."""
|
||||
return self.tools.mcp_server_url
|
||||
|
||||
@property
|
||||
def mcp_server_api_key(self) -> str:
|
||||
"""Legacy property for backward compatibility."""
|
||||
return self.tools.mcp_server_api_key
|
||||
|
||||
|
||||
_settings_instance: Settings | None = None
|
||||
|
||||
|
||||
def _load_config_from_file(config_path: str | Path) -> dict:
|
||||
"""Load configuration from YAML file."""
|
||||
config_path = Path(config_path)
|
||||
if not config_path.exists():
|
||||
logger.warning("Config file not found: %s", config_path)
|
||||
return {}
|
||||
|
||||
logger.info("Loading config from %s", config_path)
|
||||
cfg = OmegaConf.load(config_path)
|
||||
return OmegaConf.to_container(cfg, resolve=True) or {}
|
||||
|
||||
|
||||
def _load_config_from_env() -> dict:
|
||||
"""Load configuration from environment variables (STEWARD__SECTION__KEY format)."""
|
||||
cfg = {}
|
||||
prefix = "STEWARD__"
|
||||
|
||||
for key, value in os.environ.items():
|
||||
if not key.startswith(prefix):
|
||||
continue
|
||||
|
||||
# Parse STEWARD__SECTION__KEY=value
|
||||
parts = key[len(prefix) :].lower().split("__")
|
||||
if len(parts) < 2:
|
||||
continue
|
||||
|
||||
section = parts[0]
|
||||
setting_key = "__".join(parts[1:])
|
||||
|
||||
if section not in cfg:
|
||||
cfg[section] = {}
|
||||
|
||||
# Try to parse value as JSON first (for lists, etc.)
|
||||
if isinstance(cfg[section], dict):
|
||||
try:
|
||||
import json
|
||||
|
||||
cfg[section][setting_key] = json.loads(value)
|
||||
except (json.JSONDecodeError, ValueError):
|
||||
cfg[section][setting_key] = value
|
||||
|
||||
return cfg
|
||||
|
||||
|
||||
def get_settings() -> Settings:
|
||||
"""Return application settings singleton."""
|
||||
return Settings()
|
||||
"""Get or create the settings singleton.
|
||||
|
||||
Configuration is loaded in order of precedence (highest to lowest):
|
||||
1. Environment variables (STEWARD__SECTION__KEY=value)
|
||||
2. YAML config file (path via CONFIG_FILE env var)
|
||||
3. Default values from schema
|
||||
|
||||
Returns:
|
||||
Settings: The application settings.
|
||||
"""
|
||||
global _settings_instance
|
||||
|
||||
if _settings_instance is not None:
|
||||
return _settings_instance
|
||||
|
||||
# Start with schema defaults
|
||||
schema_path = Path(__file__).parent / "config_schema.yaml"
|
||||
base_cfg = OmegaConf.load(schema_path)
|
||||
|
||||
# Merge in config file if specified
|
||||
config_file = os.environ.get("CONFIG_FILE")
|
||||
if config_file:
|
||||
file_cfg = OmegaConf.create(_load_config_from_file(config_file))
|
||||
base_cfg = OmegaConf.merge(base_cfg, file_cfg)
|
||||
|
||||
# Merge in environment variables (takes precedence)
|
||||
env_cfg = OmegaConf.create(_load_config_from_env())
|
||||
if env_cfg:
|
||||
base_cfg = OmegaConf.merge(base_cfg, env_cfg)
|
||||
|
||||
# Convert to dict and create Settings instance
|
||||
config_dict = OmegaConf.to_container(base_cfg, resolve=True) or {}
|
||||
_settings_instance = Settings(**config_dict)
|
||||
|
||||
logger.info("Configuration loaded successfully")
|
||||
logger.debug("Active configuration: %s", OmegaConf.to_yaml(base_cfg))
|
||||
|
||||
return _settings_instance
|
||||
|
||||
|
||||
def reload_settings() -> Settings:
|
||||
"""Reload configuration from source (mainly for testing)."""
|
||||
global _settings_instance
|
||||
_settings_instance = None
|
||||
return get_settings()
|
||||
|
||||
Reference in New Issue
Block a user