70 lines
2.1 KiB
Python
70 lines
2.1 KiB
Python
"""Abstract notification-template persistence API.
|
|
|
|
Global / shared store: no ``user_id`` scoping. The router gates every endpoint
|
|
behind admin. ``created_by`` is audit-only.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from abc import ABC, abstractmethod
|
|
from typing import Any
|
|
|
|
|
|
class DuplicateNotificationTemplateNameError(ValueError):
|
|
"""Raised when creating/updating would collide on the unique name."""
|
|
|
|
|
|
class VersionConflictError(Exception):
|
|
"""Raised when ``expected_version`` does not match the stored version."""
|
|
|
|
def __init__(self, server: dict[str, Any]) -> None:
|
|
super().__init__("version conflict")
|
|
self.server = server
|
|
|
|
|
|
class NotificationTemplateStore(ABC):
|
|
@abstractmethod
|
|
async def list_templates(
|
|
self,
|
|
*,
|
|
keyword: str | None = None,
|
|
enabled: bool | None = None,
|
|
tag: str | None = None,
|
|
page: int = 1,
|
|
page_size: int = 20,
|
|
) -> tuple[list[dict[str, Any]], int]:
|
|
"""Return ``(items, total)`` newest-first with optional filters."""
|
|
raise NotImplementedError
|
|
|
|
@abstractmethod
|
|
async def get_template(self, template_id: str) -> dict[str, Any] | None:
|
|
raise NotImplementedError
|
|
|
|
@abstractmethod
|
|
async def create_template(self, data: dict[str, Any]) -> dict[str, Any]:
|
|
"""Create a template. Raises ``DuplicateNotificationTemplateNameError``."""
|
|
raise NotImplementedError
|
|
|
|
@abstractmethod
|
|
async def update_template(
|
|
self,
|
|
template_id: str,
|
|
data: dict[str, Any],
|
|
*,
|
|
expected_version: int,
|
|
) -> dict[str, Any] | None:
|
|
"""Update with optimistic lock. Returns None if missing.
|
|
|
|
Raises ``VersionConflictError`` (with ``.server``) on mismatch, or
|
|
``DuplicateNotificationTemplateNameError`` on a name clash.
|
|
"""
|
|
raise NotImplementedError
|
|
|
|
@abstractmethod
|
|
async def delete_template(self, template_id: str) -> bool:
|
|
raise NotImplementedError
|
|
|
|
@abstractmethod
|
|
async def name_exists(self, name: str, *, exclude_id: str | None = None) -> bool:
|
|
raise NotImplementedError
|