feat: implement full admin panel with dashboard, user management, and platform settings
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>
This commit is contained in:
96
backend/app/core/settings_manager.py
Normal file
96
backend/app/core/settings_manager.py
Normal file
@@ -0,0 +1,96 @@
|
||||
"""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"
|
||||
Reference in New Issue
Block a user