deerflow-code/offline-backend-20260512/backend/app/gateway/routers/compact_menu_overrides.py
2026-09-07 18:24:55 +08:00

106 lines
3.5 KiB
Python

"""Compact/sidebar-summary menu override API.
Routes (prefix ``/api/compact-menu-overrides``):
GET "" list compact-menu overrides.
``menu_type`` omitted / empty → type A (original legacy data).
PUT "" replace that type's override set (admin).
``menu_type`` omitted / empty → type A, does not wipe type B.
"""
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.deps import get_current_user, 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)
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]
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 _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,
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)
store = _get_store(request)
rows = await store.list_overrides(t)
return CompactMenuOverrideListResponse(
menu_type=t,
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)
t = normalize_compact_menu_type(body.menu_type)
store = _get_store(request)
user_id = await get_current_user(request)
rows = await store.replace_all(
[o.model_dump() for o in body.overrides],
menu_type=t,
updated_by=user_id,
)
return CompactMenuOverrideListResponse(
menu_type=t,
overrides=[CompactMenuOverrideResponse(**r) for r in rows],
)