67 lines
2.2 KiB
Python
67 lines
2.2 KiB
Python
"""Per-user UI preferences (account-bound, follow the user across devices).
|
|
|
|
Currently exposes a single toggle: ``hide_admin_prompt_prefix`` — when true the
|
|
chat input hides the admin-configured prompt-prefix switch so the user only uses
|
|
their own prompts. Kept as a small key-value (JSON) store so new toggles do not
|
|
each need a schema change.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
|
|
from fastapi import APIRouter, HTTPException, Request
|
|
from pydantic import BaseModel, Field
|
|
|
|
from deerflow.runtime.user_context import get_effective_user_id
|
|
|
|
logger = logging.getLogger(__name__)
|
|
router = APIRouter(prefix="/api/user-preferences", tags=["user-preferences"])
|
|
|
|
|
|
class UserPreferencesResponse(BaseModel):
|
|
hide_admin_prompt_prefix: bool = False
|
|
|
|
|
|
class UserPreferencesUpdateRequest(BaseModel):
|
|
hide_admin_prompt_prefix: bool | None = Field(default=None)
|
|
|
|
|
|
def _current_user_id(request: Request) -> str:
|
|
user = getattr(request.state, "user", None)
|
|
if user is not None:
|
|
return str(user.id)
|
|
return get_effective_user_id()
|
|
|
|
|
|
def _get_store(request: Request):
|
|
store = getattr(request.app.state, "user_preferences_store", None)
|
|
if store is None:
|
|
raise HTTPException(status_code=503, detail="User preferences store not available")
|
|
return store
|
|
|
|
|
|
def _to_response(prefs: dict) -> UserPreferencesResponse:
|
|
return UserPreferencesResponse(
|
|
hide_admin_prompt_prefix=bool(prefs.get("hide_admin_prompt_prefix", False)),
|
|
)
|
|
|
|
|
|
@router.get("", response_model=UserPreferencesResponse)
|
|
async def get_user_preferences(request: Request) -> UserPreferencesResponse:
|
|
store = _get_store(request)
|
|
user_id = _current_user_id(request)
|
|
prefs = await store.get_for_user(user_id)
|
|
return _to_response(prefs)
|
|
|
|
|
|
@router.put("", response_model=UserPreferencesResponse)
|
|
async def update_user_preferences(request: Request, body: UserPreferencesUpdateRequest) -> UserPreferencesResponse:
|
|
store = _get_store(request)
|
|
user_id = _current_user_id(request)
|
|
patch = body.model_dump(exclude_unset=True, exclude_none=True)
|
|
if not patch:
|
|
return _to_response(await store.get_for_user(user_id))
|
|
prefs = await store.merge_for_user(user_id, patch)
|
|
return _to_response(prefs)
|