862e6b7b25
Co-authored-by: Cursor <cursoragent@cursor.com>
45 lines
1.3 KiB
Python
45 lines
1.3 KiB
Python
from datetime import datetime, timezone
|
|
|
|
from app.models import NotificationItem
|
|
|
|
|
|
class NotificationStore:
|
|
def __init__(self) -> None:
|
|
self._items: list[NotificationItem] = []
|
|
|
|
def list_all(self) -> list[NotificationItem]:
|
|
return sorted(self._items, key=lambda n: n.created_at, reverse=True)
|
|
|
|
def unread_count(self) -> int:
|
|
return sum(1 for item in self._items if not item.read)
|
|
|
|
def add(self, title: str, text: str, link_tab: str | None = None) -> NotificationItem:
|
|
item = NotificationItem(
|
|
id=f"ntf-{int(datetime.now(timezone.utc).timestamp() * 1000)}",
|
|
title=title,
|
|
text=text,
|
|
created_at=datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S"),
|
|
read=False,
|
|
link_tab=link_tab, # type: ignore[arg-type]
|
|
)
|
|
self._items.insert(0, item)
|
|
return item
|
|
|
|
def mark_read(self, notification_id: str) -> bool:
|
|
for item in self._items:
|
|
if item.id == notification_id:
|
|
item.read = True
|
|
return True
|
|
return False
|
|
|
|
def mark_all_read(self) -> int:
|
|
count = 0
|
|
for item in self._items:
|
|
if not item.read:
|
|
item.read = True
|
|
count += 1
|
|
return count
|
|
|
|
|
|
notification_store = NotificationStore()
|