86 lines
2.8 KiB
Python
86 lines
2.8 KiB
Python
"""User preferences API — per-user key/value store backed by a JSON file."""
|
|
import json
|
|
import os
|
|
from typing import Optional
|
|
from fastapi import APIRouter, Request
|
|
from src.auth_helpers import get_current_user
|
|
|
|
PREFS_FILE = os.path.join("data", "user_prefs.json")
|
|
|
|
|
|
def _load():
|
|
"""Load the raw prefs file (internal use only)."""
|
|
try:
|
|
with open(PREFS_FILE, "r", encoding="utf-8") as f:
|
|
return json.load(f)
|
|
except (FileNotFoundError, json.JSONDecodeError):
|
|
return {}
|
|
|
|
|
|
def _save(prefs):
|
|
os.makedirs(os.path.dirname(PREFS_FILE), exist_ok=True)
|
|
with open(PREFS_FILE, "w", encoding="utf-8") as f:
|
|
json.dump(prefs, f, indent=2)
|
|
|
|
|
|
def _load_for_user(user: Optional[str] = None) -> dict:
|
|
"""Load preferences for a specific user."""
|
|
all_prefs = _load()
|
|
if "_users" in all_prefs:
|
|
if user is None:
|
|
# Auth disabled — return first user's prefs for backward compat
|
|
users = all_prefs["_users"]
|
|
return dict(next(iter(users.values()), {}))
|
|
return dict(all_prefs["_users"].get(user, {}))
|
|
# Legacy flat format — return as-is
|
|
return dict(all_prefs)
|
|
|
|
|
|
def _save_for_user(user: Optional[str], prefs: dict):
|
|
"""Save preferences for a specific user."""
|
|
all_prefs = _load()
|
|
if user is None:
|
|
# Auth disabled. If the store is already multi-user (e.g. auth was
|
|
# turned off on a deployment that previously ran multi-user), writing
|
|
# `prefs` flat would overwrite the whole `_users` map and destroy every
|
|
# other user's preferences. Instead write back into the same (first)
|
|
# slot _load_for_user(None) reads from, preserving the others.
|
|
if "_users" in all_prefs:
|
|
users = all_prefs["_users"]
|
|
first_key = next(iter(users), None)
|
|
if first_key is not None:
|
|
users[first_key] = prefs
|
|
_save(all_prefs)
|
|
return
|
|
_save(prefs)
|
|
return
|
|
if "_users" not in all_prefs:
|
|
all_prefs = {"_users": {}}
|
|
all_prefs["_users"][user] = prefs
|
|
_save(all_prefs)
|
|
|
|
|
|
def setup_prefs_routes():
|
|
router = APIRouter(prefix="/api/prefs", tags=["preferences"])
|
|
|
|
@router.get("")
|
|
async def get_all_prefs(request: Request):
|
|
user = get_current_user(request)
|
|
return _load_for_user(user)
|
|
|
|
@router.get("/{key}")
|
|
async def get_pref(request: Request, key: str):
|
|
user = get_current_user(request)
|
|
prefs = _load_for_user(user)
|
|
return {"key": key, "value": prefs.get(key)}
|
|
|
|
@router.put("/{key}")
|
|
async def set_pref(request: Request, key: str, body: dict):
|
|
user = get_current_user(request)
|
|
prefs = _load_for_user(user)
|
|
prefs[key] = body.get("value")
|
|
_save_for_user(user, prefs)
|
|
return {"key": key, "value": prefs[key]}
|
|
|
|
return router
|