b570f8415f
Adds complete super_admin panel with 9 pages and account owner categories page. Backend includes 5 new DB tables, ~25 API endpoints, settings manager with in-memory cache, and 29 integration tests. Frontend includes reusable admin components (DataTable, Pagination, ActionMenu, etc.) with code-split lazy loading. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
97 lines
3.1 KiB
Python
97 lines
3.1 KiB
Python
"""Runtime platform settings with in-memory cache."""
|
|
import json
|
|
import time
|
|
import uuid
|
|
from datetime import datetime, timezone
|
|
from typing import Any, Optional
|
|
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy import select
|
|
|
|
from app.models.platform_setting import PlatformSetting
|
|
|
|
|
|
class SettingsManager:
|
|
"""Manage runtime platform settings with in-memory cache (60s TTL)."""
|
|
|
|
_cache: dict[str, Any] = {}
|
|
_cache_time: float = 0
|
|
CACHE_TTL = 60
|
|
|
|
@classmethod
|
|
async def get(cls, key: str, db: AsyncSession, default: Any = None) -> Any:
|
|
if time.time() - cls._cache_time < cls.CACHE_TTL and key in cls._cache:
|
|
return cls._cache[key]
|
|
|
|
result = await db.execute(
|
|
select(PlatformSetting).where(PlatformSetting.setting_key == key)
|
|
)
|
|
setting = result.scalar_one_or_none()
|
|
if not setting:
|
|
return default
|
|
|
|
value = cls._parse_value(setting.setting_value, setting.data_type)
|
|
cls._cache[key] = value
|
|
cls._cache_time = time.time()
|
|
return value
|
|
|
|
@classmethod
|
|
async def set(cls, key: str, value: Any, db: AsyncSession, user_id: uuid.UUID) -> None:
|
|
result = await db.execute(
|
|
select(PlatformSetting).where(PlatformSetting.setting_key == key)
|
|
)
|
|
setting = result.scalar_one_or_none()
|
|
|
|
str_value = json.dumps(value) if isinstance(value, (dict, list)) else str(value).lower() if isinstance(value, bool) else str(value)
|
|
|
|
if setting:
|
|
setting.setting_value = str_value
|
|
setting.updated_by_id = user_id
|
|
setting.updated_at = datetime.now(timezone.utc)
|
|
else:
|
|
setting = PlatformSetting(
|
|
setting_key=key,
|
|
setting_value=str_value,
|
|
data_type=cls._infer_type(value),
|
|
updated_by_id=user_id,
|
|
)
|
|
db.add(setting)
|
|
|
|
# Invalidate cache
|
|
cls._cache.pop(key, None)
|
|
cls._cache_time = 0
|
|
|
|
@classmethod
|
|
async def get_all(cls, db: AsyncSession, include_sensitive: bool = False) -> dict[str, Any]:
|
|
result = await db.execute(select(PlatformSetting))
|
|
settings = result.scalars().all()
|
|
out = {}
|
|
for s in settings:
|
|
if s.is_sensitive and not include_sensitive:
|
|
out[s.setting_key] = "***"
|
|
else:
|
|
out[s.setting_key] = cls._parse_value(s.setting_value, s.data_type)
|
|
return out
|
|
|
|
@staticmethod
|
|
def _parse_value(value: Optional[str], data_type: str) -> Any:
|
|
if value is None:
|
|
return None
|
|
if data_type == "boolean":
|
|
return value.lower() == "true"
|
|
if data_type == "integer":
|
|
return int(value)
|
|
if data_type == "json":
|
|
return json.loads(value)
|
|
return value
|
|
|
|
@staticmethod
|
|
def _infer_type(value: Any) -> str:
|
|
if isinstance(value, bool):
|
|
return "boolean"
|
|
if isinstance(value, int):
|
|
return "integer"
|
|
if isinstance(value, (dict, list)):
|
|
return "json"
|
|
return "string"
|