"""SQLAlchemy-backed notification template store.""" from __future__ import annotations import uuid from datetime import datetime from typing import Any from sqlalchemy import delete, func, or_, select from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from deerflow.persistence.notification_templates.base import NotificationTemplateStore from deerflow.persistence.notification_templates.model import NotificationTemplateRow def _to_dict(row: NotificationTemplateRow) -> dict[str, Any]: data = row.to_dict() for key in ("created_at", "updated_at"): value = data.get(key) if isinstance(value, datetime): data[key] = value.isoformat() data["tags"] = data.get("tags") or [] data["enabled"] = bool(data.get("enabled", True)) data["version"] = int(data.get("version", 1)) return data class NotificationTemplateRepository(NotificationTemplateStore): def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None: self._sf = session_factory async def list_templates( self, *, keyword: str | None = None, enabled: bool | None = None, tag: str | None = None, page: int = 1, page_size: int = 20, ) -> dict[str, Any]: stmt = select(NotificationTemplateRow) count_stmt = select(func.count()).select_from(NotificationTemplateRow) if keyword: kw = f"%{keyword}%" filt = or_( NotificationTemplateRow.name.ilike(kw), NotificationTemplateRow.title_template.ilike(kw), NotificationTemplateRow.content_template.ilike(kw), ) stmt = stmt.where(filt) count_stmt = count_stmt.where(filt) if enabled is not None: stmt = stmt.where(NotificationTemplateRow.enabled == enabled) count_stmt = count_stmt.where(NotificationTemplateRow.enabled == enabled) if tag: all_rows_stmt = stmt all_rows_count_stmt = count_stmt # Filter by tag in JSON array — load all and filter in Python for portability stmt = select(NotificationTemplateRow) count_stmt = select(func.count()).select_from(NotificationTemplateRow) if keyword: kw = f"%{keyword}%" filt = or_( NotificationTemplateRow.name.ilike(kw), NotificationTemplateRow.title_template.ilike(kw), NotificationTemplateRow.content_template.ilike(kw), ) stmt = stmt.where(filt) count_stmt = count_stmt.where(filt) if enabled is not None: stmt = stmt.where(NotificationTemplateRow.enabled == enabled) count_stmt = count_stmt.where(enabled == NotificationTemplateRow.enabled) async with self._sf() as session: result = await session.execute(stmt.order_by(NotificationTemplateRow.created_at.desc())) all_rows = list(result.scalars()) filtered = [r for r in all_rows if tag in (r.tags or [])] total = len(filtered) offset = (page - 1) * page_size page_rows = filtered[offset : offset + page_size] return { "total": total, "items": [_to_dict(r) for r in page_rows], "page": page, "page_size": page_size, } async with self._sf() as session: total_result = await session.execute(count_stmt) total = total_result.scalar() or 0 offset = (page - 1) * page_size stmt = stmt.order_by(NotificationTemplateRow.created_at.desc()).offset(offset).limit(page_size) result = await session.execute(stmt) rows = list(result.scalars()) return { "total": total, "items": [_to_dict(r) for r in rows], "page": page, "page_size": page_size, } async def get_template(self, template_id: str) -> dict[str, Any] | None: async with self._sf() as session: row = await session.get(NotificationTemplateRow, template_id) return _to_dict(row) if row else None async def create_template(self, data: dict[str, Any], *, created_by: str | None = None) -> dict[str, Any]: row = NotificationTemplateRow( id=str(data.get("id") or uuid.uuid4().hex), name=str(data["name"]), title_template=str(data.get("title_template") or ""), content_template=str(data.get("content_template") or ""), tags=list(data.get("tags") or []), enabled=bool(data.get("enabled", True)), version=1, created_by=created_by, updated_by=created_by, ) async with self._sf() as session: session.add(row) await session.commit() await session.refresh(row) return _to_dict(row) async def update_template( self, template_id: str, data: dict[str, Any], *, expected_version: int, updated_by: str | None = None ) -> dict[str, Any]: async with self._sf() as session: row = await session.get(NotificationTemplateRow, template_id) if row is None: raise ValueError("NOT_FOUND") if row.version != expected_version: raise ValueError("VERSION_CONFLICT") if "name" in data: row.name = str(data["name"]) if "title_template" in data: row.title_template = str(data["title_template"]) if "content_template" in data: row.content_template = str(data["content_template"]) if "tags" in data: row.tags = list(data["tags"]) if "enabled" in data: row.enabled = bool(data["enabled"]) row.version = row.version + 1 row.updated_by = updated_by await session.commit() await session.refresh(row) return _to_dict(row) async def delete_template(self, template_id: str) -> bool: async with self._sf() as session: row = await session.get(NotificationTemplateRow, template_id) if row is None: return False await session.delete(row) await session.commit() return True async def duplicate_template(self, template_id: str, *, created_by: str | None = None) -> dict[str, Any]: async with self._sf() as session: row = await session.get(NotificationTemplateRow, template_id) if row is None: raise ValueError("NOT_FOUND") base_name = f"{row.name} - 副本" new_name = base_name counter = 1 while True: existing = await session.execute( select(NotificationTemplateRow).where(NotificationTemplateRow.name == new_name) ) if existing.scalar_one_or_none() is None: break counter += 1 new_name = f"{base_name} {counter}" new_row = NotificationTemplateRow( id=uuid.uuid4().hex, name=new_name, title_template=row.title_template, content_template=row.content_template, tags=list(row.tags or []), enabled=row.enabled, version=1, created_by=created_by, updated_by=created_by, ) session.add(new_row) await session.commit() await session.refresh(new_row) return _to_dict(new_row) async def name_exists(self, name: str, exclude_id: str | None = None) -> bool: async with self._sf() as session: stmt = select(NotificationTemplateRow).where(NotificationTemplateRow.name == name) if exclude_id: stmt = stmt.where(NotificationTemplateRow.id != exclude_id) result = await session.execute(stmt) return result.scalar_one_or_none() is not None