118 lines
3.5 KiB
Python
118 lines
3.5 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from app.core.config import settings
|
|
|
|
SETTINGS_ENV_KEYS: dict[str, str] = {
|
|
"enable_rate_limit": "ENABLE_RATE_LIMIT",
|
|
"enable_docs": "ENABLE_DOCS",
|
|
"cookie_secure": "COOKIE_SECURE",
|
|
"jwt_access_ttl_min": "JWT_ACCESS_TTL_MIN",
|
|
"auth_lockout_attempts": "AUTH_LOCKOUT_ATTEMPTS",
|
|
"auth_lockout_minutes": "AUTH_LOCKOUT_MINUTES",
|
|
"cors_origins": "CORS_ORIGINS",
|
|
"frontend_url": "FRONTEND_URL",
|
|
"public_base_url": "PUBLIC_BASE_URL",
|
|
"smtp_host": "SMTP_HOST",
|
|
"smtp_port": "SMTP_PORT",
|
|
"smtp_from": "SMTP_FROM",
|
|
"avatar_max_bytes": "AVATAR_MAX_BYTES",
|
|
"media_url_ttl_seconds": "MEDIA_URL_TTL_SECONDS",
|
|
"log_level": "LOG_LEVEL",
|
|
"audit_retention_days": "AUDIT_RETENTION_DAYS",
|
|
"jwt_refresh_ttl_days": "JWT_REFRESH_TTL_DAYS",
|
|
}
|
|
|
|
MANAGED_KEYS = tuple(SETTINGS_ENV_KEYS.keys())
|
|
|
|
|
|
def _settings_file() -> Path:
|
|
return Path(settings.compton_settings_path)
|
|
|
|
|
|
def get_settings_values() -> dict[str, Any]:
|
|
return {key: getattr(settings, key) for key in MANAGED_KEYS}
|
|
|
|
|
|
def _coerce_value(key: str, value: Any) -> Any:
|
|
current = getattr(settings, key)
|
|
if isinstance(current, bool):
|
|
if isinstance(value, bool):
|
|
return value
|
|
if isinstance(value, str):
|
|
return value.lower() in {"1", "true", "yes", "on"}
|
|
return bool(value)
|
|
if isinstance(current, int):
|
|
return int(value)
|
|
if isinstance(current, list):
|
|
if isinstance(value, list):
|
|
return [str(item) for item in value]
|
|
if isinstance(value, str):
|
|
return [item.strip() for item in value.split(",") if item.strip()]
|
|
raise ValueError(f"INVALID_LIST_{key}")
|
|
return value
|
|
|
|
|
|
def env_locks() -> dict[str, bool]:
|
|
return {key: os.getenv(env_key) is not None for key, env_key in SETTINGS_ENV_KEYS.items()}
|
|
|
|
|
|
def apply_settings_to_app(values: dict[str, Any]) -> None:
|
|
for key, value in values.items():
|
|
if key not in MANAGED_KEYS:
|
|
continue
|
|
setattr(settings, key, _coerce_value(key, value))
|
|
|
|
|
|
def read_settings() -> dict[str, Any]:
|
|
path = _settings_file()
|
|
if not path.exists():
|
|
return {}
|
|
with path.open("r", encoding="utf-8") as file:
|
|
payload = json.load(file)
|
|
if not isinstance(payload, dict):
|
|
return {}
|
|
return {key: payload[key] for key in MANAGED_KEYS if key in payload}
|
|
|
|
|
|
def write_settings(partial: dict[str, Any]) -> dict[str, Any]:
|
|
locks = env_locks()
|
|
current = get_settings_values()
|
|
for key, value in partial.items():
|
|
if key not in MANAGED_KEYS:
|
|
continue
|
|
if locks[key]:
|
|
continue
|
|
current[key] = _coerce_value(key, value)
|
|
|
|
path = _settings_file()
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
with path.open("w", encoding="utf-8") as file:
|
|
json.dump(current, file, ensure_ascii=False, indent=2)
|
|
return current
|
|
|
|
|
|
def bootstrap_settings() -> None:
|
|
apply_settings_to_app(read_settings())
|
|
|
|
|
|
def get_settings_payload() -> dict[str, Any]:
|
|
locks = env_locks()
|
|
values = get_settings_values()
|
|
secrets = {
|
|
"jwt_access_secret_configured": bool(settings.jwt_access_secret),
|
|
"jwt_refresh_pepper_configured": bool(settings.jwt_refresh_pepper),
|
|
"smtp_password_configured": bool(settings.smtp_password),
|
|
"s3_secret_key_configured": bool(settings.s3_secret_key),
|
|
}
|
|
return {
|
|
"values": values,
|
|
"locks": locks,
|
|
"settings_path": str(_settings_file()),
|
|
"secrets": secrets,
|
|
}
|