180 lines
6.9 KiB
Python
180 lines
6.9 KiB
Python
"""Compact/sidebar-summary menu override API.
|
|
|
|
Routes (prefix ``/api/compact-menu-overrides``):
|
|
GET "" read effective/default/personal compact-menu overrides.
|
|
``menu_type`` omitted / empty → type A (original legacy data).
|
|
PUT "" replace that type in the caller's editable scope.
|
|
``menu_type`` omitted / empty → type A, does not wipe type B.
|
|
DELETE "" reset that type in the caller's editable scope.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
from typing import Literal
|
|
|
|
from fastapi import APIRouter, HTTPException, Query, Request
|
|
from pydantic import BaseModel, Field
|
|
|
|
from app.gateway.auth.disabled_mode import is_auth_disabled
|
|
from app.gateway.catalog_cache import cached_catalog, catalog_key, invalidate_catalog
|
|
from app.gateway.deps import get_optional_user_from_request
|
|
from deerflow.persistence.compact_menu_overrides import (
|
|
DEFAULT_COMPACT_MENU_TYPE,
|
|
CompactMenuOverrideStore,
|
|
normalize_compact_menu_type,
|
|
)
|
|
|
|
router = APIRouter(prefix="/api/compact-menu-overrides", tags=["compact-menu-overrides"])
|
|
|
|
|
|
class CompactMenuOverride(BaseModel):
|
|
node_id: str = Field(..., min_length=1, max_length=128)
|
|
placement: Literal["sidebar", "settings", "toolbox"] = "sidebar"
|
|
sort_order: int = 0
|
|
label: str | None = Field(default=None, max_length=255)
|
|
disabled: bool = False
|
|
|
|
|
|
class CompactMenuOverrideResponse(CompactMenuOverride):
|
|
menu_type: str = DEFAULT_COMPACT_MENU_TYPE
|
|
updated_by: str | None = None
|
|
updated_at: datetime | str | None = None
|
|
|
|
|
|
class CompactMenuOverrideListResponse(BaseModel):
|
|
menu_type: str = DEFAULT_COMPACT_MENU_TYPE
|
|
overrides: list[CompactMenuOverrideResponse]
|
|
default_overrides: list[CompactMenuOverrideResponse] = Field(default_factory=list)
|
|
user_overrides: list[CompactMenuOverrideResponse] = Field(default_factory=list)
|
|
has_user_override: bool = False
|
|
editable_scope: Literal["default", "user"] = "default"
|
|
|
|
|
|
class CompactMenuOverrideReplaceRequest(BaseModel):
|
|
menu_type: str | None = Field(
|
|
default=None,
|
|
max_length=32,
|
|
description="A or B. Omitted/empty = A (original compact-menu set).",
|
|
)
|
|
overrides: list[CompactMenuOverride] = Field(default_factory=list)
|
|
|
|
|
|
def _get_store(request: Request) -> CompactMenuOverrideStore:
|
|
store = getattr(request.app.state, "compact_menu_override_store", None)
|
|
if store is None:
|
|
raise HTTPException(status_code=503, detail="Compact menu override store not available")
|
|
return store
|
|
|
|
|
|
async def _actor(request: Request) -> tuple[str | None, bool]:
|
|
"""Return ``(user_id, is_admin)`` while preserving auth-disabled mode."""
|
|
user = await get_optional_user_from_request(request)
|
|
if user is None:
|
|
if is_auth_disabled():
|
|
return None, True
|
|
raise HTTPException(status_code=401, detail="Not authenticated")
|
|
return str(user.id), getattr(user, "system_role", None) == "admin"
|
|
|
|
|
|
def _merge_overrides(default_rows: list[dict], user_rows: list[dict] | None) -> list[dict]:
|
|
"""Overlay personal rows and retain administrator rows for new nodes."""
|
|
if user_rows is None:
|
|
return [dict(row) for row in default_rows]
|
|
merged: list[dict] = []
|
|
seen: set[str] = set()
|
|
for row in user_rows:
|
|
node_id = str(row.get("node_id") or "")
|
|
if not node_id or node_id in seen:
|
|
continue
|
|
merged.append(dict(row))
|
|
seen.add(node_id)
|
|
for row in default_rows:
|
|
node_id = str(row.get("node_id") or "")
|
|
if node_id and node_id not in seen:
|
|
merged.append(dict(row))
|
|
seen.add(node_id)
|
|
return merged
|
|
|
|
|
|
async def _read_response(
|
|
request: Request,
|
|
store: CompactMenuOverrideStore,
|
|
menu_type: str,
|
|
*,
|
|
actor: tuple[str | None, bool] | None = None,
|
|
) -> CompactMenuOverrideListResponse:
|
|
user_id, is_admin = actor if actor is not None else await _actor(request)
|
|
default_rows = await store.list_overrides(menu_type)
|
|
if is_admin or user_id is None:
|
|
return CompactMenuOverrideListResponse(
|
|
menu_type=menu_type,
|
|
overrides=[CompactMenuOverrideResponse(**row) for row in default_rows],
|
|
default_overrides=[CompactMenuOverrideResponse(**row) for row in default_rows],
|
|
user_overrides=[],
|
|
has_user_override=False,
|
|
editable_scope="default",
|
|
)
|
|
user_rows = await store.list_user_overrides(user_id, menu_type)
|
|
effective = _merge_overrides(default_rows, user_rows)
|
|
return CompactMenuOverrideListResponse(
|
|
menu_type=menu_type,
|
|
overrides=[CompactMenuOverrideResponse(**row) for row in effective],
|
|
default_overrides=[CompactMenuOverrideResponse(**row) for row in default_rows],
|
|
user_overrides=[CompactMenuOverrideResponse(**row) for row in (user_rows or [])],
|
|
has_user_override=user_rows is not None,
|
|
editable_scope="user",
|
|
)
|
|
|
|
|
|
@router.get("", response_model=CompactMenuOverrideListResponse)
|
|
async def list_compact_menu_overrides(
|
|
request: Request,
|
|
menu_type: str | None = Query(
|
|
default=None,
|
|
max_length=32,
|
|
description="A or B. Omitted/empty = A, returning the original compact-menu rows.",
|
|
),
|
|
) -> CompactMenuOverrideListResponse:
|
|
t = normalize_compact_menu_type(menu_type)
|
|
actor = await _actor(request)
|
|
store = _get_store(request)
|
|
return await cached_catalog(
|
|
request, "compact_menu_overrides", catalog_key(*actor, t),
|
|
lambda: _read_response(request, store, t, actor=actor), CompactMenuOverrideListResponse,
|
|
)
|
|
|
|
|
|
@router.put("", response_model=CompactMenuOverrideListResponse)
|
|
async def replace_compact_menu_overrides(
|
|
request: Request,
|
|
body: CompactMenuOverrideReplaceRequest,
|
|
) -> CompactMenuOverrideListResponse:
|
|
t = normalize_compact_menu_type(body.menu_type)
|
|
store = _get_store(request)
|
|
user_id, is_admin = await _actor(request)
|
|
payload = [o.model_dump() for o in body.overrides]
|
|
if is_admin or user_id is None:
|
|
await store.replace_all(payload, menu_type=t, updated_by=user_id)
|
|
else:
|
|
await store.replace_user_overrides(user_id, payload, menu_type=t, updated_by=user_id)
|
|
await invalidate_catalog(request, "compact_menu_overrides")
|
|
return await _read_response(request, store, t)
|
|
|
|
|
|
@router.delete("", response_model=CompactMenuOverrideListResponse)
|
|
async def reset_compact_menu_overrides(
|
|
request: Request,
|
|
menu_type: str | None = Query(default=None, max_length=32),
|
|
) -> CompactMenuOverrideListResponse:
|
|
"""Clear the caller's personal compact layout for one variant."""
|
|
t = normalize_compact_menu_type(menu_type)
|
|
store = _get_store(request)
|
|
user_id, is_admin = await _actor(request)
|
|
if is_admin or user_id is None:
|
|
await store.replace_all([], menu_type=t, updated_by=user_id)
|
|
else:
|
|
await store.replace_user_overrides(user_id, [], menu_type=t, updated_by=user_id)
|
|
await invalidate_catalog(request, "compact_menu_overrides")
|
|
return await _read_response(request, store, t)
|