141 lines
5.3 KiB
Python
141 lines
5.3 KiB
Python
"""SQLAlchemy-backed sidebar-menu override store (global / shared)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime
|
|
from typing import Any
|
|
|
|
from sqlalchemy import delete, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
|
|
|
|
from deerflow.persistence.menu_overrides.base import MenuOverrideStore
|
|
from deerflow.persistence.menu_overrides.model import MenuOverrideRow, UserMenuOverrideConfigRow
|
|
|
|
|
|
def _to_dict(row: MenuOverrideRow) -> dict[str, Any]:
|
|
data = row.to_dict()
|
|
value = data.get("updated_at")
|
|
if isinstance(value, datetime):
|
|
data["updated_at"] = value.isoformat()
|
|
return data
|
|
|
|
|
|
def _row_from(data: dict[str, Any], updated_by: str | None) -> MenuOverrideRow:
|
|
return MenuOverrideRow(
|
|
node_id=str(data["node_id"]),
|
|
parent_id=(str(data["parent_id"]) if data.get("parent_id") not in (None, "") else None),
|
|
sort_order=int(data.get("sort_order", 0)),
|
|
label=(data.get("label") or None),
|
|
disabled=bool(data.get("disabled", False)),
|
|
updated_by=updated_by,
|
|
)
|
|
|
|
|
|
def _normalize_user_row(data: dict[str, Any]) -> dict[str, Any]:
|
|
"""Keep only the public override fields in the personal JSON document."""
|
|
return {
|
|
"node_id": str(data["node_id"]),
|
|
"parent_id": (str(data["parent_id"]) if data.get("parent_id") not in (None, "") else None),
|
|
"sort_order": int(data.get("sort_order", 0)),
|
|
"label": (data.get("label") or None),
|
|
"disabled": bool(data.get("disabled", False)),
|
|
}
|
|
|
|
|
|
def _user_rows(row: UserMenuOverrideConfigRow) -> list[dict[str, Any]]:
|
|
value = row.overrides
|
|
if isinstance(value, str):
|
|
# PortableJSON handles MySQL automatically, but this keeps old rows
|
|
# readable if a driver returned the raw TEXT value.
|
|
import json
|
|
|
|
try:
|
|
value = json.loads(value)
|
|
except (TypeError, ValueError):
|
|
value = []
|
|
if not isinstance(value, list):
|
|
value = []
|
|
updated_at = row.updated_at.isoformat() if isinstance(row.updated_at, datetime) else row.updated_at
|
|
return [
|
|
{
|
|
**_normalize_user_row(item),
|
|
"updated_by": row.updated_by,
|
|
"updated_at": updated_at,
|
|
}
|
|
for item in value
|
|
if isinstance(item, dict) and item.get("node_id")
|
|
]
|
|
|
|
|
|
class MenuOverrideRepository(MenuOverrideStore):
|
|
def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None:
|
|
self._sf = session_factory
|
|
|
|
async def list_overrides(self) -> list[dict[str, Any]]:
|
|
stmt = select(MenuOverrideRow).order_by(MenuOverrideRow.sort_order.asc())
|
|
async with self._sf() as session:
|
|
result = await session.execute(stmt)
|
|
return [_to_dict(row) for row in result.scalars()]
|
|
|
|
async def replace_all(self, rows: list[dict[str, Any]], *, updated_by: str | None = None) -> list[dict[str, Any]]:
|
|
# De-dupe by node_id (last wins) so a malformed payload can't violate the PK.
|
|
by_id: dict[str, dict[str, Any]] = {}
|
|
for r in rows:
|
|
node_id = r.get("node_id")
|
|
if node_id:
|
|
by_id[str(node_id)] = r
|
|
async with self._sf() as session:
|
|
await session.execute(delete(MenuOverrideRow))
|
|
new_rows = [_row_from(r, updated_by) for r in by_id.values()]
|
|
session.add_all(new_rows)
|
|
await session.commit()
|
|
for row in new_rows:
|
|
await session.refresh(row)
|
|
return [_to_dict(row) for row in new_rows]
|
|
|
|
async def list_user_overrides(self, user_id: str) -> list[dict[str, Any]] | None:
|
|
async with self._sf() as session:
|
|
row = await session.get(UserMenuOverrideConfigRow, str(user_id))
|
|
return _user_rows(row) if row is not None else None
|
|
|
|
async def replace_user_overrides(
|
|
self,
|
|
user_id: str,
|
|
rows: list[dict[str, Any]],
|
|
*,
|
|
updated_by: str | None = None,
|
|
) -> list[dict[str, Any]] | None:
|
|
key = str(user_id)
|
|
by_id: dict[str, dict[str, Any]] = {}
|
|
for data in rows:
|
|
node_id = data.get("node_id")
|
|
if node_id:
|
|
normalized = _normalize_user_row(data)
|
|
by_id[normalized["node_id"]] = normalized
|
|
async with self._sf() as session:
|
|
row = await session.get(UserMenuOverrideConfigRow, key)
|
|
if not by_id:
|
|
if row is not None:
|
|
await session.delete(row)
|
|
await session.commit()
|
|
return None
|
|
now = datetime.now(UTC)
|
|
payload = list(by_id.values())
|
|
if row is None:
|
|
row = UserMenuOverrideConfigRow(
|
|
user_id=key,
|
|
overrides=payload,
|
|
updated_by=updated_by or key,
|
|
)
|
|
session.add(row)
|
|
else:
|
|
row.overrides = payload
|
|
row.updated_by = updated_by or key
|
|
row.updated_at = now
|
|
await session.commit()
|
|
# The shared session factory uses ``expire_on_commit=False``. Build
|
|
# the response from the writer-side instance instead of issuing a
|
|
# post-commit refresh that a read/write-splitting endpoint could
|
|
# route to a lagging replica.
|
|
return _user_rows(row)
|