from __future__ import annotations
from onegov.core.orm.types import UTCDateTime
from uuid import uuid4
from onegov.core.collection import GenericCollection
from sedate import utcnow
from sqlalchemy import (
Column,
String,
ForeignKey,
Integer,
UniqueConstraint,
)
from sqlalchemy.orm import relationship, backref
from onegov.core.orm.types import JSON, UUID
from onegov.core.orm import Base
from typing import Any, ClassVar, TYPE_CHECKING
if TYPE_CHECKING:
import uuid
from sqlalchemy.orm import Query, Session
from onegov.org.models import News
[docs]
class PushNotification(Base):
"""
Keeps track of all outbound push notifications to prevent duplicates.
"""
[docs]
__tablename__: ClassVar[str] = 'push_notification'
#: The internal ID of the notification
[docs]
id: Column[uuid.UUID] = Column(
UUID, # type: ignore[arg-type]
nullable=False,
primary_key=True,
default=uuid4
)
#: The ID of the news item that triggered the notification
[docs]
news_id: Column[int] = Column(
Integer,
ForeignKey('pages.id', ondelete='CASCADE'),
nullable=False
)
[docs]
news: relationship[News] = relationship(
'News',
backref=backref('sent_notifications', passive_deletes=True),
foreign_keys=[news_id]
)
#: The topic/channel the notification was sent to
[docs]
topic_id: Column[str] = Column(String, nullable=False)
#: When the notification was sent
[docs]
sent_at = Column(UTCDateTime, nullable=False, default=utcnow)
#: Response information from the notification service
[docs]
response_data: Column[dict[str, Any] | None] = Column(JSON, nullable=True)
@classmethod
[docs]
def was_notification_sent(
cls, session: Session, news_id: int, topic_id: str
) -> bool:
return (
session.query(cls)
.filter(cls.news_id == news_id, cls.topic_id == topic_id)
.first()
is not None
)
@classmethod
[docs]
def record_sent_notification(
cls,
session: Session,
news_id: int,
topic_id: str,
response_data: dict[str, Any] | None
) -> PushNotification:
notification = cls(
news_id=news_id,
topic_id=topic_id,
sent_at=utcnow(),
response_data=response_data
)
session.add(notification)
session.flush()
return notification
# For each topic_id inside a news, there should be only one notification
[docs]
__table_args__ = (
UniqueConstraint('news_id', 'topic_id', name='uix_news_topic'),
)
[docs]
class PushNotificationCollection(GenericCollection[PushNotification]):
"""Simple collection for sent notifications."""
@property
[docs]
def model_class(self) -> type[PushNotification]:
return PushNotification
[docs]
def query(self) -> Query[PushNotification]:
return super().query()