Files
deerflow-code/offline-backend-20260512/backend/app/gateway/routers/compact_menu_overrides.py
T

77 lines
2.6 KiB
Python

"""Compact/sidebar-summary menu override API.
Routes (prefix ``/api/compact-menu-overrides``):
GET "" list all compact-menu overrides
PUT "" replace the entire override set (admin)
"""
from __future__ import annotations
from datetime import datetime
from typing import Literal
from fastapi import APIRouter, HTTPException, Request
from pydantic import BaseModel, Field
from app.gateway.deps import get_current_user, get_optional_user_from_request
from deerflow.persistence.compact_menu_overrides import CompactMenuOverrideStore
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"] = "sidebar"
sort_order: int = 0
label: str | None = Field(default=None, max_length=255)
class CompactMenuOverrideResponse(CompactMenuOverride):
updated_by: str | None = None
updated_at: datetime | str | None = None
class CompactMenuOverrideListResponse(BaseModel):
overrides: list[CompactMenuOverrideResponse]
class CompactMenuOverrideReplaceRequest(BaseModel):
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 _require_admin(request: Request) -> None:
user = await get_optional_user_from_request(request)
if user is None:
return
if getattr(user, "system_role", None) != "admin":
raise HTTPException(status_code=403, detail="简介菜单管理仅限管理员")
@router.get("", response_model=CompactMenuOverrideListResponse)
async def list_compact_menu_overrides(request: Request) -> CompactMenuOverrideListResponse:
store = _get_store(request)
rows = await store.list_overrides()
return CompactMenuOverrideListResponse(overrides=[CompactMenuOverrideResponse(**r) for r in rows])
@router.put("", response_model=CompactMenuOverrideListResponse)
async def replace_compact_menu_overrides(
request: Request,
body: CompactMenuOverrideReplaceRequest,
) -> CompactMenuOverrideListResponse:
await _require_admin(request)
store = _get_store(request)
user_id = await get_current_user(request)
rows = await store.replace_all(
[o.model_dump() for o in body.overrides],
updated_by=user_id,
)
return CompactMenuOverrideListResponse(overrides=[CompactMenuOverrideResponse(**r) for r in rows])