48 lines
1.5 KiB
Python
48 lines
1.5 KiB
Python
"""Abstract base class for in-app regulatory-signal notifications.
|
|
|
|
A notification is created once per triggering event (a brand-new regulation,
|
|
or a significant change to an existing one) and broadcast to every logged-in
|
|
user. There is no per-user subscription targeting — see the design doc for why.
|
|
Per-user "read" state is tracked separately from the notification itself, so
|
|
one notification row serves every user rather than being fanned out on create.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from abc import ABC, abstractmethod
|
|
|
|
|
|
class BaseNotificationStore(ABC):
|
|
"""Port interface for perception notification persistence."""
|
|
|
|
@abstractmethod
|
|
def create(
|
|
self,
|
|
*,
|
|
event_id: str,
|
|
kind: str,
|
|
title: str,
|
|
impact_level: str | None,
|
|
summary: str | None,
|
|
) -> None:
|
|
"""Record a new notification. kind is 'new' or 'changed'."""
|
|
|
|
@abstractmethod
|
|
def list_for_user(self, user_id: str, limit: int = 20) -> list[dict]:
|
|
"""Return the most recent notifications, newest first.
|
|
|
|
Each item includes a "read" boolean reflecting whether `user_id` has
|
|
marked it read.
|
|
"""
|
|
|
|
@abstractmethod
|
|
def unread_count(self, user_id: str) -> int:
|
|
"""Return how many notifications `user_id` has not yet read."""
|
|
|
|
@abstractmethod
|
|
def mark_all_read(self, user_id: str) -> int:
|
|
"""Mark every currently-unread notification read for `user_id`.
|
|
|
|
Returns the number of notifications newly marked.
|
|
"""
|