101 lines
3.1 KiB
Python
101 lines
3.1 KiB
Python
"""Configuration loading for Memory System API."""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from pathlib import Path
|
|
from typing import Any, Literal
|
|
|
|
import yaml
|
|
from pydantic import BaseModel, Field
|
|
|
|
|
|
class ServerConfig(BaseModel):
|
|
host: str = "127.0.0.1"
|
|
port: int = 1934
|
|
api_key: str = ""
|
|
|
|
|
|
class OpenVikingConfig(BaseModel):
|
|
url: str = "http://127.0.0.1:1933"
|
|
api_key: str = ""
|
|
timeout: int = 30
|
|
verify_ssl: bool = True
|
|
|
|
|
|
class EverOSConfig(BaseModel):
|
|
url: str = "http://127.0.0.1:1995"
|
|
api_key: str = ""
|
|
timeout: int = 180
|
|
verify_ssl: bool = True
|
|
health_path: str = "/health"
|
|
|
|
|
|
class StorageConfig(BaseModel):
|
|
sqlite_path: str = "/home/tom/memory-gateway/memory_system_api.sqlite3"
|
|
|
|
|
|
class LoggingConfig(BaseModel):
|
|
level: str = "INFO"
|
|
|
|
|
|
class Config(BaseModel):
|
|
server: ServerConfig = Field(default_factory=ServerConfig)
|
|
openviking: OpenVikingConfig = Field(default_factory=OpenVikingConfig)
|
|
everos: EverOSConfig = Field(default_factory=EverOSConfig)
|
|
storage: StorageConfig = Field(default_factory=StorageConfig)
|
|
logging: LoggingConfig = Field(default_factory=LoggingConfig)
|
|
|
|
|
|
_config: Config | None = None
|
|
|
|
|
|
def load_config(config_path: str | None = None) -> Config:
|
|
path = Path(config_path or os.environ.get("MEMORY_SYSTEM_CONFIG", "config.yaml"))
|
|
if not path.exists():
|
|
return _apply_env_overrides(Config())
|
|
with path.open("r", encoding="utf-8") as handle:
|
|
data = yaml.safe_load(handle) or {}
|
|
config = Config(
|
|
server=ServerConfig(**data.get("server", {})),
|
|
openviking=OpenVikingConfig(**data.get("openviking", {})),
|
|
everos=EverOSConfig(**data.get("everos", {})),
|
|
storage=StorageConfig(**data.get("storage", {})),
|
|
logging=LoggingConfig(**data.get("logging", {})),
|
|
)
|
|
return _apply_env_overrides(config)
|
|
|
|
|
|
def get_config() -> Config:
|
|
global _config
|
|
if _config is None:
|
|
_config = load_config()
|
|
return _config
|
|
|
|
|
|
def set_config(config: Config) -> None:
|
|
global _config
|
|
_config = config
|
|
|
|
|
|
def _apply_env_overrides(config: Config) -> Config:
|
|
updates: dict[str, dict[str, Any]] = {
|
|
"server": _env_updates("MEMORY_SYSTEM_SERVER", {"API_KEY": "api_key", "HOST": "host", "PORT": "port"}),
|
|
"openviking": _env_updates("OPENVIKING", {"URL": "url", "BASE_URL": "url", "API_KEY": "api_key", "TIMEOUT": "timeout"}),
|
|
"everos": _env_updates("EVEROS", {"URL": "url", "BASE_URL": "url", "API_KEY": "api_key", "TIMEOUT": "timeout"}),
|
|
"storage": _env_updates("MEMORY_SYSTEM_STORAGE", {"SQLITE_PATH": "sqlite_path"}),
|
|
}
|
|
for section, values in updates.items():
|
|
if values:
|
|
setattr(config, section, getattr(config, section).model_copy(update=values))
|
|
return config
|
|
|
|
|
|
def _env_updates(prefix: str, mapping: dict[str, str]) -> dict[str, Any]:
|
|
values: dict[str, Any] = {}
|
|
for env_name, field_name in mapping.items():
|
|
raw = os.environ.get(f"{prefix}_{env_name}")
|
|
if raw is None:
|
|
continue
|
|
values[field_name] = int(raw) if field_name in {"port", "timeout"} else raw
|
|
return values
|